114 lines
4.5 KiB
Python
114 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from us_cpa.cases import CaseConflictError, CaseManager
|
|
|
|
|
|
class CaseManagerTests(unittest.TestCase):
|
|
def test_create_case_builds_expected_layout(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
case_dir = Path(temp_dir) / "2025-jane-doe"
|
|
manager = CaseManager(case_dir)
|
|
|
|
manifest = manager.create_case(case_label="Jane Doe", tax_year=2025)
|
|
|
|
self.assertEqual(manifest["caseLabel"], "Jane Doe")
|
|
self.assertEqual(manifest["taxYear"], 2025)
|
|
for name in (
|
|
"input",
|
|
"extracted",
|
|
"return",
|
|
"output",
|
|
"reports",
|
|
"issues",
|
|
"sources",
|
|
):
|
|
self.assertTrue((case_dir / name).is_dir())
|
|
self.assertTrue((case_dir / "case-manifest.json").exists())
|
|
|
|
def test_intake_registers_documents_and_user_facts(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
root = Path(temp_dir)
|
|
case_dir = root / "2025-jane-doe"
|
|
document = root / "w2.txt"
|
|
document.write_text("sample w2")
|
|
manager = CaseManager(case_dir)
|
|
manager.create_case(case_label="Jane Doe", tax_year=2025)
|
|
|
|
result = manager.intake(
|
|
tax_year=2025,
|
|
user_facts={"filingStatus": "single", "taxpayer.ssnLast4": "1234"},
|
|
document_paths=[document],
|
|
)
|
|
|
|
self.assertEqual(result["status"], "accepted")
|
|
self.assertEqual(len(result["registeredDocuments"]), 1)
|
|
self.assertTrue((case_dir / "input" / "w2.txt").exists())
|
|
facts = json.loads((case_dir / "extracted" / "facts.json").read_text())
|
|
self.assertEqual(facts["facts"]["filingStatus"]["value"], "single")
|
|
|
|
def test_intake_extracts_machine_usable_facts_from_text_documents(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
root = Path(temp_dir)
|
|
case_dir = root / "2025-jane-doe"
|
|
w2 = root / "w2.txt"
|
|
w2.write_text(
|
|
"Form W-2 Wage and Tax Statement\n"
|
|
"Employee: Jane Doe\n"
|
|
"Box 1 Wages, tips, other compensation 50000\n"
|
|
"Box 2 Federal income tax withheld 6000\n"
|
|
)
|
|
interest = root / "1099-int.txt"
|
|
interest.write_text(
|
|
"Form 1099-INT\n"
|
|
"Recipient: Jane Doe\n"
|
|
"Box 1 Interest Income 1750\n"
|
|
)
|
|
manager = CaseManager(case_dir)
|
|
manager.create_case(case_label="Jane Doe", tax_year=2025)
|
|
|
|
result = manager.intake(
|
|
tax_year=2025,
|
|
user_facts={"filingStatus": "single"},
|
|
document_paths=[w2, interest],
|
|
)
|
|
|
|
self.assertEqual(result["status"], "accepted")
|
|
facts = json.loads((case_dir / "extracted" / "facts.json").read_text())
|
|
self.assertEqual(facts["facts"]["wages"]["value"], 50000.0)
|
|
self.assertEqual(facts["facts"]["federalWithholding"]["value"], 6000.0)
|
|
self.assertEqual(facts["facts"]["taxableInterest"]["value"], 1750.0)
|
|
self.assertEqual(facts["facts"]["wages"]["sources"][0]["sourceType"], "document_extract")
|
|
|
|
def test_conflicting_facts_raise_structured_issue(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
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={"filingStatus": "single"},
|
|
document_paths=[],
|
|
)
|
|
|
|
with self.assertRaises(CaseConflictError) as context:
|
|
manager.intake(
|
|
tax_year=2025,
|
|
user_facts={"filingStatus": "married_filing_jointly"},
|
|
document_paths=[],
|
|
)
|
|
|
|
issue = context.exception.issue
|
|
self.assertEqual(issue["status"], "needs_resolution")
|
|
self.assertEqual(issue["issueType"], "fact_conflict")
|
|
self.assertEqual(issue["field"], "filingStatus")
|
|
self.assertTrue((case_dir / "issues" / "open-issues.json").exists())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|