feat: add us-cpa preparation workflow

This commit is contained in:
Stefano Fiorini
2026-03-15 01:28:22 -05:00
parent decf3132d5
commit 82cf3d9010
8 changed files with 309 additions and 10 deletions

View File

@@ -191,6 +191,76 @@ class UsCpaCliSmokeTests(unittest.TestCase):
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())
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,75 @@
from __future__ import annotations
import json
import tempfile
import unittest
from io import BytesIO
from pathlib import Path
from reportlab.pdfgen import canvas
from us_cpa.cases import CaseManager
from us_cpa.prepare import EfileExporter, PrepareEngine
from us_cpa.sources import TaxYearCorpus, bootstrap_irs_catalog
class PrepareEngineTests(unittest.TestCase):
def build_case(self, temp_dir: str) -> tuple[CaseManager, TaxYearCorpus]:
case_dir = Path(temp_dir) / "2025-jane-doe"
manager = CaseManager(case_dir)
manager.create_case(case_label="Jane Doe", tax_year=2025)
manager.intake(
tax_year=2025,
user_facts={
"taxpayer.fullName": "Jane Doe",
"filingStatus": "single",
"wages": 50000,
"taxableInterest": 100,
"federalWithholding": 6000,
},
document_paths=[],
)
corpus = TaxYearCorpus(cache_root=Path(temp_dir) / "cache")
def fake_fetch(url: str) -> bytes:
buffer = BytesIO()
pdf = canvas.Canvas(buffer)
pdf.drawString(72, 720, f"Template for {url}")
pdf.save()
return buffer.getvalue()
corpus.download_catalog(2025, bootstrap_irs_catalog(2025), fetcher=fake_fetch)
return manager, corpus
def test_prepare_creates_normalized_return_and_artifacts(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
manager, corpus = self.build_case(temp_dir)
engine = PrepareEngine(corpus=corpus)
result = engine.prepare_case(manager.case_dir)
self.assertEqual(result["status"], "prepared")
self.assertEqual(result["summary"]["requiredForms"], ["f1040"])
self.assertEqual(result["summary"]["reviewRequiredArtifacts"], ["f1040"])
self.assertTrue((manager.case_dir / "return" / "normalized-return.json").exists())
self.assertTrue((manager.case_dir / "output" / "artifacts.json").exists())
normalized = json.loads((manager.case_dir / "return" / "normalized-return.json").read_text())
self.assertEqual(normalized["totals"]["adjustedGrossIncome"], 50100.0)
self.assertEqual(normalized["totals"]["taxableIncome"], 34350.0)
def test_exporter_writes_efile_ready_payload(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
manager, corpus = self.build_case(temp_dir)
engine = PrepareEngine(corpus=corpus)
engine.prepare_case(manager.case_dir)
export = EfileExporter().export_case(manager.case_dir)
self.assertEqual(export["status"], "draft")
self.assertTrue((manager.case_dir / "output" / "efile-ready.json").exists())
self.assertEqual(export["returnSummary"]["requiredForms"], ["f1040"])
if __name__ == "__main__":
unittest.main()

View File

@@ -31,7 +31,7 @@ class QuestionEngineTests(unittest.TestCase):
self.assertEqual(analysis["issue"], "standard_deduction")
self.assertEqual(analysis["taxYear"], 2025)
self.assertEqual(analysis["conclusion"]["answer"], "$15,000")
self.assertEqual(analysis["conclusion"]["answer"], "$15,750")
self.assertEqual(analysis["confidence"], "high")
self.assertTrue(analysis["authorities"])
self.assertEqual(analysis["authorities"][0]["sourceClass"], "irs_instructions")
@@ -57,7 +57,7 @@ class QuestionEngineTests(unittest.TestCase):
"factsUsed": [{"field": "filingStatus", "value": "single"}],
"missingFacts": [],
"authorities": [{"title": "Instructions for Form 1040 and Schedules 1-3"}],
"conclusion": {"answer": "$15,000", "summary": "Single filers use a $15,000 standard deduction for tax year 2025."},
"conclusion": {"answer": "$15,750", "summary": "Single filers use a $15,750 standard deduction for tax year 2025."},
"confidence": "high",
"followUpQuestions": [],
"primaryLawRequired": False,
@@ -66,7 +66,7 @@ class QuestionEngineTests(unittest.TestCase):
conversation = render_analysis(analysis)
memo = render_memo(analysis)
self.assertIn("$15,000", conversation)
self.assertIn("$15,750", conversation)
self.assertIn("Issue", memo)
self.assertIn("Authorities", memo)