Spaces:
Configuration error
Configuration error
| """Integration tests: FastAPI TestClient against every endpoint (TEST-02), | |
| covering auth, role enforcement, and client data isolation. | |
| Uses the transactional-rollback fixtures in conftest.py so nothing written | |
| here is ever actually persisted to the database (except the seed officer | |
| row, which is idempotent and expected to persist -- see conftest.py). | |
| """ | |
| from __future__ import annotations | |
| from app.config import settings | |
| from tests.integration.conftest import ( | |
| PAYMENT_LEGIT_PATTERN, | |
| TRANSFER_FRAUD_PATTERN, | |
| create_client, | |
| login, | |
| ) | |
| def test_health(api_client): | |
| res = api_client.get("/health") | |
| assert res.status_code == 200 | |
| assert res.json()["status"] == "ok" | |
| def test_root_redirects_to_login(api_client): | |
| res = api_client.get("/", follow_redirects=False) | |
| assert res.status_code in (302, 307) | |
| assert res.headers["location"] == "/login" | |
| # ------------------------------------------------------------------ auth -- | |
| def test_login_with_correct_credentials_succeeds(api_client): | |
| res = api_client.post( | |
| "/api/auth/login", | |
| json={"email": settings.SEED_OFFICER_EMAIL, "password": settings.SEED_OFFICER_PASSWORD}, | |
| ) | |
| assert res.status_code == 200 | |
| body = res.json() | |
| assert body["role"] == "officer" | |
| assert "session_token" in res.cookies | |
| def test_login_with_wrong_password_fails_generically(api_client): | |
| res = api_client.post( | |
| "/api/auth/login", | |
| json={"email": settings.SEED_OFFICER_EMAIL, "password": "definitely-wrong"}, | |
| ) | |
| assert res.status_code == 401 | |
| assert res.json()["detail"] == "Incorrect email or password." | |
| def test_login_with_unknown_email_fails_with_same_generic_error(api_client): | |
| res = api_client.post( | |
| "/api/auth/login", json={"email": "nobody@fakebankmail.com", "password": "whatever123"} | |
| ) | |
| assert res.status_code == 401 | |
| assert res.json()["detail"] == "Incorrect email or password." | |
| def test_me_without_session_is_401(api_client): | |
| res = api_client.get("/api/auth/me") | |
| assert res.status_code == 401 | |
| def test_me_with_valid_session_returns_user(api_client, officer_identity): | |
| res = api_client.get("/api/auth/me", cookies=officer_identity["cookies"]) | |
| assert res.status_code == 200 | |
| assert res.json()["email"] == settings.SEED_OFFICER_EMAIL | |
| def test_logout_invalidates_the_session(api_client, officer_identity): | |
| res = api_client.post("/api/auth/logout", cookies=officer_identity["cookies"]) | |
| assert res.status_code == 200 | |
| res = api_client.get("/api/auth/me", cookies=officer_identity["cookies"]) | |
| assert res.status_code == 401 | |
| # --------------------------------------------------------- role boundaries -- | |
| def test_officer_route_rejects_anonymous(api_client): | |
| res = api_client.get("/api/officer/clients") | |
| assert res.status_code == 401 | |
| def test_officer_route_rejects_client_role(api_client, officer_identity): | |
| client_identity = create_client( | |
| api_client, officer_identity, email="roletest.client@fakebankmail.com", name="Role Test" | |
| ) | |
| res = api_client.get("/api/officer/clients", cookies=client_identity["cookies"]) | |
| assert res.status_code == 403 | |
| def test_client_route_rejects_officer_role(api_client, officer_identity): | |
| res = api_client.get("/api/client/profile", cookies=officer_identity["cookies"]) | |
| assert res.status_code == 403 | |
| def test_client_route_rejects_anonymous(api_client): | |
| res = api_client.get("/api/client/profile") | |
| assert res.status_code == 401 | |
| # --------------------------------------------------------------- officer -- | |
| def test_officer_creates_client_and_lists_it(api_client, officer_identity): | |
| identity = create_client( | |
| api_client, officer_identity, email="create.test@fakebankmail.com", name="Create Test" | |
| ) | |
| assert identity["user"]["role"] == "client" | |
| res = api_client.get("/api/officer/clients", cookies=officer_identity["cookies"]) | |
| assert res.status_code == 200 | |
| assert any(c["email"] == "create.test@fakebankmail.com" for c in res.json()) | |
| def test_officer_create_client_duplicate_email_conflicts(api_client, officer_identity): | |
| create_client(api_client, officer_identity, email="dupe@fakebankmail.com", name="First") | |
| res = api_client.post( | |
| "/api/officer/clients", | |
| json={ | |
| "name": "Second", | |
| "email": "dupe@fakebankmail.com", | |
| "temp_password": "TempPass123!", | |
| "starting_balance": 100, | |
| "account_type": "checking", | |
| }, | |
| cookies=officer_identity["cookies"], | |
| ) | |
| assert res.status_code == 409 | |
| def test_officer_client_detail_shows_transactions_and_alerts(api_client, officer_identity, two_clients): | |
| client_a, _ = two_clients | |
| res = api_client.post( | |
| "/api/client/transactions", | |
| json={ | |
| "type": "TRANSFER", | |
| "amount": 5000, | |
| "name_dest": "C_MULE", | |
| "simulated_at": "2026-03-01T10:00:00Z", | |
| }, | |
| cookies=client_a["cookies"], | |
| ) | |
| assert res.status_code == 200 | |
| assert res.json()["risk_tier"] == "high" | |
| res = api_client.get(f"/api/officer/clients/{client_a['client_id']}", cookies=officer_identity["cookies"]) | |
| assert res.status_code == 200 | |
| detail = res.json() | |
| assert len(detail["transactions"]) == 1 | |
| assert len(detail["alerts"]) == 1 | |
| assert detail["current_balance"] == 0.0 | |
| def test_officer_sees_alerts_across_all_clients(api_client, officer_identity, two_clients): | |
| client_a, client_b = two_clients | |
| api_client.post( | |
| "/api/client/transactions", | |
| json={"type": "TRANSFER", "amount": 5000, "name_dest": "C_MULE_A", "simulated_at": "2026-03-01T10:00:00Z"}, | |
| cookies=client_a["cookies"], | |
| ) | |
| api_client.post( | |
| "/api/client/transactions", | |
| json={"type": "TRANSFER", "amount": 3000, "name_dest": "C_MULE_B", "simulated_at": "2026-03-01T11:00:00Z"}, | |
| cookies=client_b["cookies"], | |
| ) | |
| res = api_client.get("/api/officer/alerts", params={"page_size": 200}, cookies=officer_identity["cookies"]) | |
| assert res.status_code == 200 | |
| body = res.json() | |
| # Don't assert an exact global total -- the queue is cross-client by | |
| # design and may include real, already-committed alerts outside this | |
| # test's rolled-back transaction. Just confirm both fixture clients' | |
| # alerts are present in it. | |
| client_names = {item["client_name"] for item in body["items"]} | |
| assert {"Fixture Alpha", "Fixture Beta"}.issubset(client_names) | |
| def test_officer_can_review_and_dismiss_alert(api_client, officer_identity, two_clients): | |
| client_a, _ = two_clients | |
| api_client.post( | |
| "/api/client/transactions", | |
| json={"type": "TRANSFER", "amount": 5000, "name_dest": "C_MULE", "simulated_at": "2026-03-01T10:00:00Z"}, | |
| cookies=client_a["cookies"], | |
| ) | |
| alerts = api_client.get( | |
| "/api/officer/alerts", params={"disposition": "open"}, cookies=officer_identity["cookies"] | |
| ).json()["items"] | |
| alert_id = alerts[0]["id"] | |
| res = api_client.patch( | |
| f"/api/officer/alerts/{alert_id}", json={"disposition": "reviewed"}, cookies=officer_identity["cookies"] | |
| ) | |
| assert res.status_code == 200 | |
| assert res.json()["disposition"] == "reviewed" | |
| assert res.json()["reviewed_at"] is not None | |
| def test_officer_can_reject_alert_and_reverse_transaction(api_client, officer_identity, two_clients): | |
| client_a, _ = two_clients | |
| res = api_client.post( | |
| "/api/client/transactions", | |
| json={"type": "TRANSFER", "amount": 5000, "name_dest": "C_MULE", "simulated_at": "2026-03-01T10:00:00Z"}, | |
| cookies=client_a["cookies"], | |
| ) | |
| assert res.status_code == 200 | |
| assert api_client.get("/api/client/profile", cookies=client_a["cookies"]).json()["current_balance"] == 0.0 | |
| alerts = api_client.get( | |
| "/api/officer/alerts", params={"disposition": "open"}, cookies=officer_identity["cookies"] | |
| ).json()["items"] | |
| alert_id = next(a["id"] for a in alerts if a["client_id"] == client_a["client_id"]) | |
| res = api_client.patch( | |
| f"/api/officer/alerts/{alert_id}", json={"disposition": "rejected"}, cookies=officer_identity["cookies"] | |
| ) | |
| assert res.status_code == 200 | |
| body = res.json() | |
| assert body["disposition"] == "rejected" | |
| assert body["reversed"] is True | |
| # the amount is back on the client's balance | |
| profile = api_client.get("/api/client/profile", cookies=client_a["cookies"]).json() | |
| assert profile["current_balance"] == 5000.0 | |
| # the client's own alert view reflects the reversal in plain language | |
| client_alerts = api_client.get("/api/client/alerts", cookies=client_a["cookies"]).json() | |
| assert client_alerts[0]["reversed"] is True | |
| assert "reversed" in client_alerts[0]["status_message"].lower() | |
| # officer's client detail view shows the transaction as reversed too | |
| detail = api_client.get( | |
| f"/api/officer/clients/{client_a['client_id']}", cookies=officer_identity["cookies"] | |
| ).json() | |
| assert detail["current_balance"] == 5000.0 | |
| assert detail["transactions"][0]["reversed"] is True | |
| def test_officer_patch_nonexistent_alert_404s(api_client, officer_identity): | |
| res = api_client.patch( | |
| "/api/officer/alerts/999999999", json={"disposition": "dismissed"}, cookies=officer_identity["cookies"] | |
| ) | |
| assert res.status_code == 404 | |
| def test_officer_stats_endpoints(api_client, officer_identity): | |
| for path in ["/api/officer/stats/overview", "/api/officer/stats/eda", "/api/officer/stats/model-performance"]: | |
| res = api_client.get(path, cookies=officer_identity["cookies"]) | |
| assert res.status_code == 200, path | |
| def test_officer_predict_and_batch(api_client, officer_identity): | |
| res = api_client.post("/api/predict", json=TRANSFER_FRAUD_PATTERN, cookies=officer_identity["cookies"]) | |
| assert res.status_code == 200 | |
| assert res.json()["risk_tier"] == "high" | |
| res = api_client.post( | |
| "/api/transactions/batch", | |
| json={"transactions": [PAYMENT_LEGIT_PATTERN]}, | |
| cookies=officer_identity["cookies"], | |
| ) | |
| assert res.status_code == 200 | |
| assert res.json()["results"][0]["risk_tier"] == "low" | |
| def test_predict_rejects_client_role(api_client, officer_identity): | |
| client_identity = create_client( | |
| api_client, officer_identity, email="predicttest@fakebankmail.com", name="Predict Test" | |
| ) | |
| res = api_client.post("/api/predict", json=TRANSFER_FRAUD_PATTERN, cookies=client_identity["cookies"]) | |
| assert res.status_code == 403 | |
| # ---------------------------------------------------------------- client -- | |
| def test_client_profile_reflects_balance_and_alerts(api_client, two_clients): | |
| client_a, _ = two_clients | |
| res = api_client.get("/api/client/profile", cookies=client_a["cookies"]) | |
| assert res.status_code == 200 | |
| body = res.json() | |
| assert body["current_balance"] == 5000.0 | |
| assert body["open_alerts"] == [] | |
| def test_client_submits_legit_transaction(api_client, two_clients): | |
| client_a, _ = two_clients | |
| res = api_client.post( | |
| "/api/client/transactions", | |
| json={"type": "PAYMENT", "amount": 50, "name_dest": "MSHOP", "simulated_at": "2026-03-01T09:00:00Z"}, | |
| cookies=client_a["cookies"], | |
| ) | |
| assert res.status_code == 200 | |
| body = res.json() | |
| assert body["risk_tier"] == "low" | |
| assert body["top_features"] is None # SHAP internals are officer-only | |
| def test_client_cannot_overdraw(api_client, two_clients): | |
| client_a, _ = two_clients | |
| res = api_client.post( | |
| "/api/client/transactions", | |
| json={"type": "TRANSFER", "amount": 999999, "name_dest": "C_X", "simulated_at": "2026-03-01T09:00:00Z"}, | |
| cookies=client_a["cookies"], | |
| ) | |
| assert res.status_code == 422 | |
| def test_client_sees_only_own_transactions_and_alerts(api_client, two_clients): | |
| client_a, client_b = two_clients | |
| api_client.post( | |
| "/api/client/transactions", | |
| json={"type": "TRANSFER", "amount": 5000, "name_dest": "C_MULE_A", "simulated_at": "2026-03-01T10:00:00Z"}, | |
| cookies=client_a["cookies"], | |
| ) | |
| api_client.post( | |
| "/api/client/transactions", | |
| json={"type": "PAYMENT", "amount": 20, "name_dest": "MSHOP", "simulated_at": "2026-03-01T09:00:00Z"}, | |
| cookies=client_b["cookies"], | |
| ) | |
| res_a = api_client.get("/api/client/transactions", cookies=client_a["cookies"]) | |
| res_b = api_client.get("/api/client/transactions", cookies=client_b["cookies"]) | |
| assert len(res_a.json()) == 1 | |
| assert len(res_b.json()) == 1 | |
| assert res_a.json()[0]["name_dest"] == "C_MULE_A" | |
| assert res_b.json()[0]["name_dest"] == "MSHOP" | |
| alerts_a = api_client.get("/api/client/alerts", cookies=client_a["cookies"]).json() | |
| alerts_b = api_client.get("/api/client/alerts", cookies=client_b["cookies"]).json() | |
| assert len(alerts_a) == 1 # the fraud-pattern transfer | |
| assert len(alerts_b) == 0 # the legit payment | |
| def test_client_cannot_access_another_clients_data_via_officer_routes(api_client, officer_identity, two_clients): | |
| """A client has no client_id parameter anywhere in their own routes, so | |
| the only way to attempt cross-client access is hitting an officer | |
| route directly -- confirm that's blocked by role, not by omission.""" | |
| client_a, _ = two_clients | |
| res = api_client.get(f"/api/officer/clients/{client_a['client_id']}", cookies=client_a["cookies"]) | |
| assert res.status_code == 403 | |
| # ---------------------------------------------------------------- pages -- | |
| def test_html_pages_render(api_client): | |
| for path in [ | |
| "/login", | |
| "/officer", | |
| "/officer/clients", | |
| "/officer/alerts", | |
| "/officer/insights", | |
| "/officer/model", | |
| "/client", | |
| "/client/transactions/new", | |
| "/client/transactions", | |
| "/client/alerts", | |
| ]: | |
| res = api_client.get(path) | |
| assert res.status_code == 200, path | |
| assert "text/html" in res.headers["content-type"] | |
| def test_openapi_docs_available(api_client): | |
| assert api_client.get("/docs").status_code == 200 | |
| assert api_client.get("/openapi.json").status_code == 200 | |