import os import io import uuid from datetime import date import pytest from fastapi.testclient import TestClient # Ensure required env vars before importing app os.environ.setdefault("SECRET_KEY", "x" * 32) os.environ.setdefault("DATABASE_URL", "sqlite:////tmp/delphi_test.sqlite") from app.main import app # noqa: E402 from app.auth.security import get_current_user # noqa: E402 from tests.helpers import assert_http_error, assert_validation_error # noqa: E402 class _User: def __init__(self): self.id = 1 self.username = "tester" self.is_admin = True self.is_active = True self.is_approver = True @pytest.fixture() def client(): app.dependency_overrides[get_current_user] = lambda: _User() try: yield TestClient(app) finally: app.dependency_overrides.pop(get_current_user, None) def _create_customer_and_file(client: TestClient) -> tuple[str, str]: cust_id = f"Q-{uuid.uuid4().hex[:8]}" resp = client.post("/api/customers/", json={"id": cust_id, "last": "QDRO", "email": "q@example.com"}) assert resp.status_code == 200, resp.text file_no = f"QF-{uuid.uuid4().hex[:6]}" payload = { "file_no": file_no, "id": cust_id, "regarding": "QDRO matter", "empl_num": "E01", "file_type": "CIVIL", "opened": date.today().isoformat(), "status": "ACTIVE", "rate_per_hour": 150.0, } resp = client.post("/api/files/", json=payload) assert resp.status_code == 200, resp.text return cust_id, file_no def _dummy_docx_bytes() -> bytes: try: from docx import Document except Exception: return b"PK\x03\x04" d = Document() p = d.add_paragraph() p.add_run("QDRO for ") p.add_run("{{CLIENT_FULL}}") p = d.add_paragraph() p.add_run("File ") p.add_run("{{FILE_NO}}") buf = io.BytesIO() d.save(buf) return buf.getvalue() def test_qdro_crud_and_list_by_file(client: TestClient): _, file_no = _create_customer_and_file(client) # 404 when file missing on create resp = client.post("/api/qdros", json={"file_no": "NOFILE"}) assert_http_error(resp, 404, "File not found") # Create create = { "file_no": file_no, "version": "01", "status": "DRAFT", "case_number": "2025-1234", "pet": "Alice", "res": "Bob", "percent_awarded": "50%", } resp = client.post("/api/qdros", json=create) assert resp.status_code == 200, resp.text q = resp.json() qid = q["id"] assert q["file_no"] == file_no # List by file resp = client.get(f"/api/qdros/{file_no}") assert resp.status_code == 200 assert any(item["id"] == qid for item in resp.json()) # Get by id resp = client.get(f"/api/qdros/item/{qid}") assert resp.status_code == 200 assert resp.json()["id"] == qid # Update resp = client.put(f"/api/qdros/{qid}", json={"status": "APPROVED"}) assert resp.status_code == 200 assert resp.json()["status"] == "APPROVED" # Delete resp = client.delete(f"/api/qdros/{qid}") assert resp.status_code == 200 # Now 404 resp = client.get(f"/api/qdros/item/{qid}") assert_http_error(resp, 404, "QDRO not found") def test_qdro_division_calculation_and_persist_percent(client: TestClient): _, file_no = _create_customer_and_file(client) resp = client.post("/api/qdros", json={"file_no": file_no}) qid = resp.json()["id"] # percent -> amount resp = client.post(f"/api/qdros/{qid}/calculate-division", json={"account_balance": 10000.0, "percent": 25.0}) assert resp.status_code == 200, resp.text assert resp.json()["amount"] == 2500.0 # amount -> percent and save resp = client.post( f"/api/qdros/{qid}/calculate-division", json={"account_balance": 20000.0, "amount": 5000.0, "save_percent_string": True}, ) assert resp.status_code == 200 pct = resp.json()["percent"] assert round(pct, 2) == 25.0 # Verify saved percent_awarded string resp = client.get(f"/api/qdros/item/{qid}") assert resp.status_code == 200 assert "%" in (resp.json().get("percent_awarded") or "") def test_qdro_versions_and_communications(client: TestClient): _, file_no = _create_customer_and_file(client) resp = client.post("/api/qdros", json={"file_no": file_no, "content": "Initial"}) qid = resp.json()["id"] # Create version resp = client.post(f"/api/qdros/{qid}/versions", json={"version_label": "02", "status": "DRAFT"}) assert resp.status_code == 200 ver = resp.json() assert ver["version_label"] == "02" # List versions resp = client.get(f"/api/qdros/{qid}/versions") assert resp.status_code == 200 assert any(v["id"] == ver["id"] for v in resp.json()) # Communications comm = { "channel": "email", "subject": "Request Info", "message": "Please provide latest statements", "contact_name": "Plan Admin", "status": "sent", } resp = client.post(f"/api/qdros/{qid}/communications", json=comm) assert resp.status_code == 200 comm_id = resp.json()["id"] resp = client.get(f"/api/qdros/{qid}/communications") assert resp.status_code == 200 assert any(c["id"] == comm_id for c in resp.json()) def test_plan_info_create_and_list(client: TestClient): plan_id = f"PL-{uuid.uuid4().hex[:6]}" payload = {"plan_id": plan_id, "plan_name": "Acme 401k", "plan_type": "401k"} resp = client.post("/api/plan-info", json=payload) assert resp.status_code == 200, resp.text resp = client.get("/api/plan-info") assert resp.status_code == 200 ids = {row["plan_id"] for row in resp.json()} assert plan_id in ids def test_qdro_document_generation_uses_template_system(client: TestClient): # Create file + qdro _, file_no = _create_customer_and_file(client) resp = client.post("/api/qdros", json={"file_no": file_no, "pet": "Alice", "res": "Bob"}) qid = resp.json()["id"] # Upload a template through templates API files = { "file": ( "qdro.docx", _dummy_docx_bytes(), "application/vnd.openxmlformats-officedocument.wordprocessingml.document", ) } resp = client.post( "/api/templates/upload", data={"name": f"QDRO Template {uuid.uuid4().hex[:6]}", "semantic_version": "1.0.0"}, files=files, ) assert resp.status_code == 200, resp.text tpl_id = resp.json()["id"] # Generate using our endpoint resp = client.post(f"/api/qdros/{qid}/generate-document", json={"template_id": tpl_id, "context": {}}) assert resp.status_code == 200, resp.text body = resp.json() assert isinstance(body.get("resolved"), dict) assert isinstance(body.get("unresolved"), list) assert body.get("output_size", 0) >= 0 def test_qdro_transition_authorization(client: TestClient): # Override to non-admin, non-approver user class _LimitedUser: def __init__(self): self.id = 2 self.username = "limited" self.is_admin = False self.is_active = True self.is_approver = False app.dependency_overrides[get_current_user] = lambda: _LimitedUser() try: # Set up file and qdro # Need a client with current override; use existing fixture client indirectly via app local_client = TestClient(app) _, file_no = _create_customer_and_file(local_client) resp = local_client.post("/api/qdros", json={"file_no": file_no}) assert resp.status_code == 200, resp.text qid = resp.json()["id"] # DRAFT -> APPROVAL_PENDING allowed resp = local_client.post(f"/api/qdros/{qid}/submit-for-approval", json={"reason": "send"}) assert resp.status_code == 200, resp.text assert resp.json()["status"] == "APPROVAL_PENDING" # APPROVAL_PENDING -> APPROVED forbidden for non-approver resp = local_client.post(f"/api/qdros/{qid}/approve", json={"reason": "ok"}) assert_http_error(resp, 403, "Not enough permissions") # APPROVAL_PENDING -> FILED forbidden for non-approver resp = local_client.post(f"/api/qdros/{qid}/file", json={"reason": "file"}) assert_http_error(resp, 403, "Not enough permissions") # Generic transition to APPROVED forbidden resp = local_client.post(f"/api/qdros/{qid}/transition", json={"target_status": "APPROVED"}) assert_http_error(resp, 403, "Not enough permissions") finally: app.dependency_overrides.pop(get_current_user, None)