390 lines
14 KiB
Python
390 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
SKILL_DIR = Path(__file__).resolve().parents[1]
|
|
SRC_DIR = SKILL_DIR / "src"
|
|
|
|
|
|
def _pyproject_text() -> str:
|
|
return (SKILL_DIR / "pyproject.toml").read_text()
|
|
|
|
|
|
class UsCpaCliSmokeTests(unittest.TestCase):
|
|
def test_skill_scaffold_files_exist(self) -> None:
|
|
self.assertTrue((SKILL_DIR / "SKILL.md").exists())
|
|
self.assertTrue((SKILL_DIR / "pyproject.toml").exists())
|
|
self.assertTrue((SKILL_DIR / "README.md").exists())
|
|
self.assertTrue((SKILL_DIR / "scripts" / "us-cpa").exists())
|
|
self.assertTrue(
|
|
(SKILL_DIR.parent.parent / "docs" / "us-cpa.md").exists()
|
|
)
|
|
|
|
def test_pyproject_declares_runtime_and_dev_dependencies(self) -> None:
|
|
pyproject = _pyproject_text()
|
|
self.assertIn('"pypdf>=', pyproject)
|
|
self.assertIn('"reportlab>=', pyproject)
|
|
self.assertIn("[project.optional-dependencies]", pyproject)
|
|
self.assertIn('"pytest>=', pyproject)
|
|
|
|
def test_readme_documents_install_and_script_usage(self) -> None:
|
|
readme = (SKILL_DIR / "README.md").read_text()
|
|
self.assertIn("pip install -e .[dev]", readme)
|
|
self.assertIn("scripts/us-cpa", readme)
|
|
self.assertIn("python -m unittest", readme)
|
|
|
|
def test_docs_explain_openclaw_installation_flow(self) -> None:
|
|
readme = (SKILL_DIR / "README.md").read_text()
|
|
operator_doc = (SKILL_DIR.parent.parent / "docs" / "us-cpa.md").read_text()
|
|
skill_doc = (SKILL_DIR / "SKILL.md").read_text()
|
|
|
|
self.assertIn("OpenClaw installation", readme)
|
|
self.assertIn("~/.openclaw/workspace/skills/us-cpa", readme)
|
|
self.assertIn(".venv/bin/python", readme)
|
|
self.assertIn("OpenClaw installation", operator_doc)
|
|
self.assertIn("rsync -a --delete", operator_doc)
|
|
self.assertIn("~/.openclaw/workspace/skills/us-cpa/scripts/us-cpa", skill_doc)
|
|
|
|
def test_wrapper_prefers_local_virtualenv_python(self) -> None:
|
|
wrapper = (SKILL_DIR / "scripts" / "us-cpa").read_text()
|
|
self.assertIn('.venv/bin/python', wrapper)
|
|
self.assertIn('PYTHON_BIN', wrapper)
|
|
|
|
def test_fixture_directories_exist(self) -> None:
|
|
fixtures_dir = SKILL_DIR / "tests" / "fixtures"
|
|
for name in ("irs", "facts", "documents", "returns"):
|
|
self.assertTrue((fixtures_dir / name).exists())
|
|
|
|
def run_cli(self, *args: str) -> subprocess.CompletedProcess[str]:
|
|
env = os.environ.copy()
|
|
env["PYTHONPATH"] = str(SRC_DIR)
|
|
return subprocess.run(
|
|
[sys.executable, "-m", "us_cpa.cli", *args],
|
|
text=True,
|
|
capture_output=True,
|
|
env=env,
|
|
)
|
|
|
|
def test_help_lists_all_commands(self) -> None:
|
|
result = self.run_cli("--help")
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
for command in (
|
|
"question",
|
|
"prepare",
|
|
"review",
|
|
"fetch-year",
|
|
"extract-docs",
|
|
"render-forms",
|
|
"export-efile-ready",
|
|
):
|
|
self.assertIn(command, result.stdout)
|
|
|
|
def test_question_command_emits_json_by_default(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
env = os.environ.copy()
|
|
env["PYTHONPATH"] = str(SRC_DIR)
|
|
env["US_CPA_CACHE_DIR"] = temp_dir
|
|
subprocess.run(
|
|
[sys.executable, "-m", "us_cpa.cli", "fetch-year", "--tax-year", "2025"],
|
|
text=True,
|
|
capture_output=True,
|
|
env=env,
|
|
check=True,
|
|
)
|
|
result = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
"-m",
|
|
"us_cpa.cli",
|
|
"question",
|
|
"--tax-year",
|
|
"2025",
|
|
"--question",
|
|
"What is the standard deduction?",
|
|
],
|
|
text=True,
|
|
capture_output=True,
|
|
env=env,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
payload = json.loads(result.stdout)
|
|
self.assertEqual(payload["command"], "question")
|
|
self.assertEqual(payload["format"], "json")
|
|
self.assertEqual(payload["question"], "What is the standard deduction?")
|
|
self.assertEqual(payload["status"], "answered")
|
|
self.assertIn("analysis", payload)
|
|
|
|
def test_prepare_requires_case_dir(self) -> None:
|
|
result = self.run_cli("prepare", "--tax-year", "2025")
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("case directory", result.stderr.lower())
|
|
|
|
def test_extract_docs_can_create_case_and_register_facts(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
case_dir = Path(temp_dir) / "2025-jane-doe"
|
|
facts_path = Path(temp_dir) / "facts.json"
|
|
facts_path.write_text(json.dumps({"filingStatus": "single"}))
|
|
|
|
result = self.run_cli(
|
|
"extract-docs",
|
|
"--tax-year",
|
|
"2025",
|
|
"--case-dir",
|
|
str(case_dir),
|
|
"--create-case",
|
|
"--case-label",
|
|
"Jane Doe",
|
|
"--facts-json",
|
|
str(facts_path),
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
payload = json.loads(result.stdout)
|
|
self.assertEqual(payload["status"], "accepted")
|
|
self.assertEqual(payload["factCount"], 1)
|
|
self.assertTrue((case_dir / "case-manifest.json").exists())
|
|
|
|
def test_extract_docs_stops_on_conflicts(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
case_dir = Path(temp_dir) / "2025-jane-doe"
|
|
first_facts = Path(temp_dir) / "facts-1.json"
|
|
second_facts = Path(temp_dir) / "facts-2.json"
|
|
first_facts.write_text(json.dumps({"filingStatus": "single"}))
|
|
second_facts.write_text(json.dumps({"filingStatus": "married_filing_jointly"}))
|
|
|
|
first = self.run_cli(
|
|
"extract-docs",
|
|
"--tax-year",
|
|
"2025",
|
|
"--case-dir",
|
|
str(case_dir),
|
|
"--create-case",
|
|
"--case-label",
|
|
"Jane Doe",
|
|
"--facts-json",
|
|
str(first_facts),
|
|
)
|
|
self.assertEqual(first.returncode, 0, first.stderr)
|
|
|
|
second = self.run_cli(
|
|
"extract-docs",
|
|
"--tax-year",
|
|
"2025",
|
|
"--case-dir",
|
|
str(case_dir),
|
|
"--facts-json",
|
|
str(second_facts),
|
|
)
|
|
self.assertNotEqual(second.returncode, 0)
|
|
payload = json.loads(second.stdout)
|
|
self.assertEqual(payload["status"], "needs_resolution")
|
|
self.assertEqual(payload["issueType"], "fact_conflict")
|
|
|
|
def test_question_markdown_memo_mode_renders_tax_memo(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
env = os.environ.copy()
|
|
env["PYTHONPATH"] = str(SRC_DIR)
|
|
env["US_CPA_CACHE_DIR"] = temp_dir
|
|
subprocess.run(
|
|
[sys.executable, "-m", "us_cpa.cli", "fetch-year", "--tax-year", "2025"],
|
|
text=True,
|
|
capture_output=True,
|
|
env=env,
|
|
check=True,
|
|
)
|
|
result = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
"-m",
|
|
"us_cpa.cli",
|
|
"question",
|
|
"--tax-year",
|
|
"2025",
|
|
"--format",
|
|
"markdown",
|
|
"--style",
|
|
"memo",
|
|
"--question",
|
|
"What is the standard deduction?",
|
|
],
|
|
text=True,
|
|
capture_output=True,
|
|
env=env,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn("# Tax Memo", result.stdout)
|
|
self.assertIn("## Conclusion", result.stdout)
|
|
|
|
def test_prepare_command_generates_return_package(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
env = os.environ.copy()
|
|
env["PYTHONPATH"] = str(SRC_DIR)
|
|
env["US_CPA_CACHE_DIR"] = str(Path(temp_dir) / "cache")
|
|
subprocess.run(
|
|
[sys.executable, "-m", "us_cpa.cli", "fetch-year", "--tax-year", "2025"],
|
|
text=True,
|
|
capture_output=True,
|
|
env=env,
|
|
check=True,
|
|
)
|
|
|
|
case_dir = Path(temp_dir) / "2025-jane-doe"
|
|
facts_path = Path(temp_dir) / "facts.json"
|
|
facts_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"taxpayer.fullName": "Jane Doe",
|
|
"filingStatus": "single",
|
|
"wages": 50000,
|
|
"taxableInterest": 100,
|
|
"federalWithholding": 6000,
|
|
}
|
|
)
|
|
)
|
|
subprocess.run(
|
|
[
|
|
sys.executable,
|
|
"-m",
|
|
"us_cpa.cli",
|
|
"extract-docs",
|
|
"--tax-year",
|
|
"2025",
|
|
"--case-dir",
|
|
str(case_dir),
|
|
"--create-case",
|
|
"--case-label",
|
|
"Jane Doe",
|
|
"--facts-json",
|
|
str(facts_path),
|
|
],
|
|
text=True,
|
|
capture_output=True,
|
|
env=env,
|
|
check=True,
|
|
)
|
|
|
|
result = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
"-m",
|
|
"us_cpa.cli",
|
|
"prepare",
|
|
"--tax-year",
|
|
"2025",
|
|
"--case-dir",
|
|
str(case_dir),
|
|
],
|
|
text=True,
|
|
capture_output=True,
|
|
env=env,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
payload = json.loads(result.stdout)
|
|
self.assertEqual(payload["status"], "prepared")
|
|
self.assertEqual(payload["summary"]["requiredForms"], ["f1040"])
|
|
self.assertTrue((case_dir / "output" / "artifacts.json").exists())
|
|
|
|
def test_review_command_returns_findings(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
env = os.environ.copy()
|
|
env["PYTHONPATH"] = str(SRC_DIR)
|
|
env["US_CPA_CACHE_DIR"] = str(Path(temp_dir) / "cache")
|
|
subprocess.run(
|
|
[sys.executable, "-m", "us_cpa.cli", "fetch-year", "--tax-year", "2025"],
|
|
text=True,
|
|
capture_output=True,
|
|
env=env,
|
|
check=True,
|
|
)
|
|
case_dir = Path(temp_dir) / "2025-jane-doe"
|
|
facts_path = Path(temp_dir) / "facts.json"
|
|
facts_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"taxpayer.fullName": "Jane Doe",
|
|
"filingStatus": "single",
|
|
"wages": 50000,
|
|
"taxableInterest": 100,
|
|
"federalWithholding": 6000,
|
|
}
|
|
)
|
|
)
|
|
subprocess.run(
|
|
[
|
|
sys.executable,
|
|
"-m",
|
|
"us_cpa.cli",
|
|
"extract-docs",
|
|
"--tax-year",
|
|
"2025",
|
|
"--case-dir",
|
|
str(case_dir),
|
|
"--create-case",
|
|
"--case-label",
|
|
"Jane Doe",
|
|
"--facts-json",
|
|
str(facts_path),
|
|
],
|
|
text=True,
|
|
capture_output=True,
|
|
env=env,
|
|
check=True,
|
|
)
|
|
subprocess.run(
|
|
[
|
|
sys.executable,
|
|
"-m",
|
|
"us_cpa.cli",
|
|
"prepare",
|
|
"--tax-year",
|
|
"2025",
|
|
"--case-dir",
|
|
str(case_dir),
|
|
],
|
|
text=True,
|
|
capture_output=True,
|
|
env=env,
|
|
check=True,
|
|
)
|
|
normalized_path = case_dir / "return" / "normalized-return.json"
|
|
normalized = json.loads(normalized_path.read_text())
|
|
normalized["totals"]["adjustedGrossIncome"] = 99999.0
|
|
normalized_path.write_text(json.dumps(normalized, indent=2))
|
|
|
|
result = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
"-m",
|
|
"us_cpa.cli",
|
|
"review",
|
|
"--tax-year",
|
|
"2025",
|
|
"--case-dir",
|
|
str(case_dir),
|
|
],
|
|
text=True,
|
|
capture_output=True,
|
|
env=env,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
payload = json.loads(result.stdout)
|
|
self.assertEqual(payload["status"], "reviewed")
|
|
self.assertEqual(payload["findingCount"], 2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|