258 lines
8.5 KiB
Python
258 lines
8.5 KiB
Python
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)
|
|
|
|
|