Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| """ | |
| API smoke tests for LedgerShield. | |
| Run: | |
| python -m pytest tests/test_api_smoke.py -q | |
| """ | |
| from fastapi.testclient import TestClient | |
| import server.app as app_module | |
| from server.app import app | |
| client = TestClient(app) | |
| def _run_task_d_investigation_steps() -> None: | |
| """ | |
| Do a short realistic investigation before submitting task D. | |
| This is important because the new grader rewards trajectory quality, | |
| not just the final answer. | |
| """ | |
| steps = [ | |
| { | |
| "action_type": "inspect_email_thread", | |
| "payload": {"thread_id": "THR-100"}, | |
| }, | |
| { | |
| "action_type": "lookup_vendor_history", | |
| "payload": {"vendor_key": "northwind-industrial"}, | |
| }, | |
| { | |
| "action_type": "lookup_policy", | |
| "payload": {}, | |
| }, | |
| { | |
| "action_type": "compare_bank_account", | |
| "payload": { | |
| "vendor_key": "northwind-industrial", | |
| "proposed_bank_account": "IN99FAKE000999888", | |
| }, | |
| }, | |
| { | |
| "action_type": "request_callback_verification", | |
| "payload": {}, | |
| }, | |
| { | |
| "action_type": "lookup_policy", | |
| "payload": {}, | |
| }, | |
| { | |
| "action_type": "lookup_policy", | |
| "payload": {}, | |
| }, | |
| { | |
| "action_type": "lookup_policy", | |
| "payload": {}, | |
| }, | |
| ] | |
| for step in steps: | |
| response = client.post("/step", json=step) | |
| assert response.status_code == 200, response.text | |
| def test_health(): | |
| response = client.get("/health") | |
| assert response.status_code == 200 | |
| assert response.json()["status"] == "ok" | |
| def test_leaderboard_endpoint(): | |
| response = client.get("/leaderboard") | |
| assert response.status_code == 200 | |
| payload = response.json() | |
| assert payload["benchmark"] == "ledgershield-v3" | |
| assert "entries" in payload | |
| def test_leaderboard_endpoint_degrades_gracefully_without_benchmark_report(monkeypatch): | |
| monkeypatch.setattr(app_module, "_load_benchmark_report_module", lambda: None) | |
| fallback_client = TestClient(app_module.build_app()) | |
| response = fallback_client.get("/leaderboard") | |
| assert response.status_code == 200 | |
| payload = response.json() | |
| assert payload["benchmark"] == "ledgershield-v3" | |
| assert payload["entries"] == [] | |
| assert "unavailable" in payload["note"] | |
| def test_reset_endpoint(): | |
| response = client.post("/reset", json={"case_id": "CASE-A-001"}) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert data["observation"]["case_id"] == "CASE-A-001" | |
| assert data["done"] is False | |
| assert data["observation"]["task_type"] == "task_a" | |
| assert len(data["observation"]["visible_documents"]) >= 1 | |
| assert "benchmark_split" not in data["observation"]["case_metadata"] | |
| def test_state_endpoint(): | |
| client.post("/reset", json={"case_id": "CASE-D-001"}) | |
| response = client.get("/state") | |
| assert response.status_code == 200 | |
| data = response.json() | |
| assert data["case_id"] == "CASE-D-001" | |
| assert "gold" not in str(data).lower() | |
| assert "hidden_risk_signals" not in data | |
| assert "observed_risk_signals" in data | |
| assert "pending_events" in data | |
| def test_step_ocr_endpoint(): | |
| client.post("/reset", json={"case_id": "CASE-A-001"}) | |
| response = client.post( | |
| "/step", | |
| json={ | |
| "action_type": "ocr", | |
| "payload": {"doc_id": "INV-A-001", "mode": "accurate"}, | |
| }, | |
| ) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| last = data["observation"]["last_tool_result"] | |
| assert last["tool_name"] == "ocr" | |
| assert last["success"] is True | |
| assert len(last["tokens"]) > 0 | |
| assert "reward_model" in data["info"] | |
| assert data["info"]["reward_model"]["terminal"] is False | |
| assert data["done"] is False | |
| def test_submit_decision_endpoint(): | |
| client.post("/reset", json={"case_id": "CASE-D-001"}) | |
| _run_task_d_investigation_steps() | |
| response = client.post( | |
| "/step", | |
| json={ | |
| "action_type": "submit_decision", | |
| "payload": { | |
| "decision": "ESCALATE_FRAUD", | |
| "confidence": 0.95, | |
| "reason_codes": [ | |
| "bank_override_attempt", | |
| "sender_domain_spoof", | |
| "policy_bypass_attempt", | |
| "duplicate_near_match", | |
| ], | |
| "policy_checks": { | |
| "three_way_match": "pass", | |
| "bank_change_verification": "fail", | |
| "duplicate_check": "fail", | |
| "approval_threshold_check": "pass", | |
| }, | |
| "evidence_map": { | |
| "bank_override_attempt": { | |
| "doc_id": "INV-D-001", | |
| "page": 1, | |
| "bbox": [10, 110, 170, 120], | |
| "token_ids": ["d6"], | |
| }, | |
| "sender_domain_spoof": { | |
| "doc_id": "THR-100", | |
| "page": 1, | |
| "bbox": [10, 10, 250, 20], | |
| "token_ids": ["ed1"], | |
| }, | |
| "policy_bypass_attempt": { | |
| "doc_id": "THR-100", | |
| "page": 1, | |
| "bbox": [10, 70, 250, 80], | |
| "token_ids": ["ed4"], | |
| }, | |
| "duplicate_near_match": { | |
| "doc_id": "INV-D-001", | |
| "page": 1, | |
| "bbox": [10, 30, 150, 40], | |
| "token_ids": ["d2"], | |
| }, | |
| }, | |
| "counterfactual": ( | |
| "Would PAY if the sender domain matched approved vendor records " | |
| "and the bank account matched vendor master." | |
| ), | |
| }, | |
| }, | |
| ) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| result = data["observation"]["last_tool_result"] | |
| assert data["done"] is True | |
| assert result["tool_name"] == "submit_decision" | |
| assert result["success"] is True | |
| assert result["decision"] == "ESCALATE_FRAUD" | |
| assert result["final_score"] > 0.85 | |
| assert data["info"]["reward_model"]["terminal"] is True | |
| def test_clean_duplicate_screening_case_returns_no_hits(): | |
| client.post("/reset", json={"case_id": "CASE-C-002"}) | |
| response = client.post( | |
| "/step", | |
| json={ | |
| "action_type": "search_ledger", | |
| "payload": { | |
| "vendor_key": "eurocaps-components", | |
| "invoice_number": "EC-4402-26", | |
| "amount": 845.0, | |
| }, | |
| }, | |
| ) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| last = data["observation"]["last_tool_result"] | |
| assert last["tool_name"] == "search_ledger" | |
| assert last["count"] == 0 | |
| def test_clean_task_d_pay_submission_endpoint(): | |
| client.post("/reset", json={"case_id": "CASE-D-002"}) | |
| steps = [ | |
| {"action_type": "ocr", "payload": {"doc_id": "THR-130", "mode": "accurate"}}, | |
| {"action_type": "inspect_email_thread", "payload": {"thread_id": "THR-130"}}, | |
| {"action_type": "lookup_vendor_history", "payload": {"vendor_key": "bluepeak-logistics"}}, | |
| {"action_type": "lookup_policy", "payload": {}}, | |
| { | |
| "action_type": "compare_bank_account", | |
| "payload": { | |
| "vendor_key": "bluepeak-logistics", | |
| "proposed_bank_account": "IN77BP555666777", | |
| }, | |
| }, | |
| { | |
| "action_type": "search_ledger", | |
| "payload": { | |
| "vendor_key": "bluepeak-logistics", | |
| "invoice_number": "BLP-8891-MAY", | |
| "amount": 8850.0, | |
| }, | |
| }, | |
| ] | |
| for step in steps: | |
| response = client.post("/step", json=step) | |
| assert response.status_code == 200, response.text | |
| response = client.post( | |
| "/step", | |
| json={ | |
| "action_type": "submit_decision", | |
| "payload": { | |
| "decision": "PAY", | |
| "confidence": 0.88, | |
| "reason_codes": [], | |
| "policy_checks": { | |
| "three_way_match": "pass", | |
| "bank_change_verification": "pass", | |
| "duplicate_check": "pass", | |
| "approval_threshold_check": "pass", | |
| }, | |
| "evidence_map": {}, | |
| "counterfactual": ( | |
| "Would HOLD if the sender domain changed, the bank account mismatched " | |
| "vendor master, or a duplicate cluster appeared in ledger history." | |
| ), | |
| }, | |
| }, | |
| ) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| result = data["observation"]["last_tool_result"] | |
| assert data["done"] is True | |
| assert result["decision"] == "PAY" | |
| assert result["final_score"] > 0.80 | |
| def test_campaign_task_d_reset_exposes_portfolio_context(): | |
| response = client.post("/reset", json={"case_id": "CASE-D-003"}) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| observation = data["observation"] | |
| invoice_docs = [doc for doc in observation["visible_documents"] if doc["doc_type"] == "invoice"] | |
| assert len(invoice_docs) == 2 | |
| assert observation["portfolio_context"]["linked_invoice_count"] == 2 | |
| assert observation["portfolio_context"]["queue_pressure"] == "campaign" | |
| def test_task_e_reset_exposes_multi_invoice_campaign(): | |
| response = client.post("/reset", json={"case_id": "CASE-E-001"}) | |
| assert response.status_code == 200 | |
| data = response.json() | |
| observation = data["observation"] | |
| invoice_docs = [doc for doc in observation["visible_documents"] if doc["doc_type"] == "invoice"] | |
| assert len(invoice_docs) == 3 | |
| assert observation["task_type"] == "task_e" | |
| assert observation["portfolio_context"]["linked_invoice_count"] == 3 | |
| assert observation["portfolio_context"]["queue_pressure"] == "campaign" | |