import pytest from fastapi.testclient import TestClient from sqlalchemy import create_engine from sqlalchemy.orm import Session from sqlalchemy.pool import StaticPool from app.api.deps import get_database, get_provider_factory from app.api.v1.endpoints import runs from app.db.base import Base from app.db.repositories.repositories import ( ResultRepository, TestRunRepository as RunRepository, UserRepository, ) from app.main import app from app.providers.model_provider import ChatResponse from app.services.auth_service import token_digest class FakeProvider: endpoint = "https://example.test/v1/chat/completions" model = "fake-model" async def list_models(self): return ["fake-model"] async def chat(self, messages, tools=None): content = messages[-1]["content"] if tools: return ChatResponse(content="", tool_calls=[{"id": "call-1"}], raw={}) if isinstance(content, list): return ChatResponse(content="蓝色圆形", tool_calls=[], raw={}) if "TEXT_OK" in content: return ChatResponse(content="TEXT_OK", tool_calls=[], raw={}) if "SYSTEM_OK" in messages[0]["content"]: return ChatResponse(content="SYSTEM_OK", tool_calls=[], raw={}) if "编号是什么" in content: return ChatResponse(content="4837", tool_calls=[], raw={}) if "status" in content: return ChatResponse(content='{"status":"ok"}', tool_calls=[], raw={}) return ChatResponse(content="OK", tool_calls=[], raw={}) class FakeFactory: def create(self, _): return FakeProvider() @pytest.fixture def client(monkeypatch): engine = create_engine( "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool ) Base.metadata.create_all(engine) session = Session(engine) async def skip_background(*_): return None app.dependency_overrides[get_database] = lambda: session app.dependency_overrides[get_provider_factory] = lambda: FakeFactory() app.state.session_factory = lambda: session UserRepository(session).create( username="auditor", password="correct-horse-battery-staple", is_admin=True ) monkeypatch.setattr(runs, "execute_run", skip_background) try: yield TestClient(app) finally: app.dependency_overrides.clear() del app.state.session_factory session.close() def test_health_contract(client): response = client.get("/api/v1/health") assert response.status_code == 200 assert response.json()["status"] == "ok" def test_authentication_protects_api_and_login_issues_bearer_token(client): unauthenticated = client.get("/api/v1/providers/target/models") assert unauthenticated.status_code == 401 assert unauthenticated.headers["WWW-Authenticate"] == "Bearer" assert unauthenticated.headers["X-Request-ID"] assert client.get("/api/v1/health").status_code == 200 invalid = client.post( "/api/v1/auth/login", json={"username": "auditor", "password": "wrong-password"} ) assert invalid.status_code == 401 assert invalid.json()["error"]["code"] == "UNAUTHORIZED" login = client.post( "/api/v1/auth/login", json={"username": "auditor", "password": "correct-horse-battery-staple"}, ) assert login.status_code == 200 token = login.json()["access_token"] assert login.json()["token_type"] == "bearer" assert login.json()["refresh_token"] assert login.json()["refresh_expires_at"] > login.json()["expires_at"] allowed = client.get( "/api/v1/providers/target/models", headers={"Authorization": f"Bearer {token}"} ) assert allowed.status_code == 200 def test_refresh_rotates_token_and_idempotently_replays_the_same_response(client, monkeypatch): now = 1_000_000 monkeypatch.setattr("app.api.v1.endpoints.auth.time.time", lambda: now) login = client.post( "/api/v1/auth/login", json={"username": "auditor", "password": "correct-horse-battery-staple"}, ).json() assert login["expires_at"] == now + 3600 assert login["refresh_expires_at"] == now + 28800 request = {"refresh_token": login["refresh_token"]} headers = {"Idempotency-Key": "refresh-attempt-1"} now += 1800 first = client.post("/api/v1/auth/refresh", json=request, headers=headers) replay = client.post("/api/v1/auth/refresh", json=request, headers=headers) assert first.status_code == replay.status_code == 200 assert first.json() == replay.json() assert first.json()["expires_at"] == now + 3600 assert first.json()["refresh_expires_at"] == now + 28800 assert first.json()["refresh_expires_at"] > login["refresh_expires_at"] assert first.json()["refresh_token"] != login["refresh_token"] assert ( client.get( "/api/v1/auth/me", headers={"Authorization": f"Bearer {first.json()['access_token']}"}, ).status_code == 200 ) def test_refresh_requires_idempotency_key_and_replay_revokes_the_session(client, monkeypatch): now = 1_000_000 monkeypatch.setattr("app.api.v1.endpoints.auth.time.time", lambda: now) login = client.post( "/api/v1/auth/login", json={"username": "auditor", "password": "correct-horse-battery-staple"}, ).json() request = {"refresh_token": login["refresh_token"]} assert client.post("/api/v1/auth/refresh", json=request).status_code == 422 now += 1800 rotated = client.post( "/api/v1/auth/refresh", json=request, headers={"Idempotency-Key": "refresh-attempt-1"}, ).json() replay_attack = client.post( "/api/v1/auth/refresh", json=request, headers={"Idempotency-Key": "different-attempt"}, ) assert replay_attack.status_code == 401 assert ( client.post( "/api/v1/auth/refresh", json={"refresh_token": rotated["refresh_token"]}, headers={"Idempotency-Key": "refresh-attempt-2"}, ).status_code == 401 ) assert ( client.get( "/api/v1/auth/me", headers={"Authorization": f"Bearer {rotated['access_token']}"}, ).status_code == 401 ) def test_refresh_is_rejected_before_window_without_consuming_token(client, monkeypatch): now = 1_000_000 monkeypatch.setattr("app.api.v1.endpoints.auth.time.time", lambda: now) login = client.post( "/api/v1/auth/login", json={"username": "auditor", "password": "correct-horse-battery-staple"}, ).json() request = {"refresh_token": login["refresh_token"]} early = client.post( "/api/v1/auth/refresh", json=request, headers={"Idempotency-Key": "early-attempt"}, ) now += 1800 due = client.post( "/api/v1/auth/refresh", json=request, headers={"Idempotency-Key": "due-attempt"}, ) assert early.status_code == 409 assert early.json()["error"] == { "code": "REFRESH_NOT_DUE", "message": "访问令牌尚未进入续约窗口", "details": {"refresh_after": 1_001_800}, "request_id": early.headers["X-Request-ID"], } assert due.status_code == 200 def test_refresh_slides_session_expiry_past_original_login_deadline(client, monkeypatch): now = 1_000_000 monkeypatch.setattr("app.api.v1.endpoints.auth.time.time", lambda: now) login = client.post( "/api/v1/auth/login", json={"username": "auditor", "password": "correct-horse-battery-staple"}, ).json() original_deadline = login["refresh_expires_at"] now += 1800 first = client.post( "/api/v1/auth/refresh", json={"refresh_token": login["refresh_token"]}, headers={"Idempotency-Key": "first-renewal"}, ).json() now = original_deadline + 1 second = client.post( "/api/v1/auth/refresh", json={"refresh_token": first["refresh_token"]}, headers={"Idempotency-Key": "past-original-deadline"}, ) assert second.status_code == 200 assert second.json()["expires_at"] == now + 3600 assert second.json()["refresh_expires_at"] == now + 28800 assert second.json()["refresh_expires_at"] > original_deadline def test_logout_revokes_refresh_token_family(client): login = client.post( "/api/v1/auth/login", json={"username": "auditor", "password": "correct-horse-battery-staple"}, ).json() headers = {"Authorization": f"Bearer {login['access_token']}"} assert client.post("/api/v1/auth/logout", headers=headers).status_code == 204 assert ( client.post( "/api/v1/auth/refresh", json={"refresh_token": login["refresh_token"]}, headers={"Idempotency-Key": "after-logout"}, ).status_code == 401 ) def test_expired_refresh_token_is_rejected(client): login = client.post( "/api/v1/auth/login", json={"username": "auditor", "password": "correct-horse-battery-staple"}, ).json() db = app.state.session_factory() row = UserRepository(db).get_refresh_token(token_digest(login["refresh_token"])) row.expires_at = 0 db.commit() assert ( client.post( "/api/v1/auth/refresh", json={"refresh_token": login["refresh_token"]}, headers={"Idempotency-Key": "expired-token"}, ).status_code == 401 ) @pytest.mark.parametrize( ("method", "path"), [ ("get", "/api/v1/auth/me"), ("patch", "/api/v1/auth/password"), ("post", "/api/v1/providers/target/check"), ("patch", "/api/v1/auth/users/1/password"), ("get", "/api/v1/providers/target/models"), ("post", "/api/v1/runs"), ("get", "/api/v1/runs"), ("patch", "/api/v1/runs/1"), ("delete", "/api/v1/runs/1"), ("post", "/api/v1/runs/1/resume"), ("post", "/api/v1/runs/1/retry-errors"), ("post", "/api/v1/runs/1/results/R0001/retry"), ("get", "/api/v1/runs/1"), ("get", "/api/v1/runs/1/results"), ("get", "/api/v1/runs/1/export"), ("get", "/api/v1/reports"), ("get", "/api/v1/reports/1"), ], ) def test_every_business_api_requires_authentication(client, method, path): assert getattr(client, method)(path).status_code == 401 def test_admin_can_manage_provider_configs(client, auth_headers): inherited = client.get("/api/v1/providers", headers=auth_headers).json() assert {item["provider_id"] for item in inherited} == {"target", "judge"} assert all(item["source"] == "env" for item in inherited) payload = { "provider_id": "target", "base_url": "https://models.example/v1", "chat_path": "/chat/completions", "models_path": "/models", "model_name": "safe-model", "auth_type": "bearer", "api_key": "secret-value", "auth_header": "Authorization", "auth_prefix": "Bearer", } created = client.post("/api/v1/providers", headers=auth_headers, json=payload) assert created.status_code == 201 assert created.json()["api_key_configured"] is True assert "api_key" not in created.json() provider_id = created.json()["provider_id"] assert client.post("/api/v1/providers", headers=auth_headers, json=payload).status_code == 409 effective = client.get("/api/v1/providers", headers=auth_headers).json() target = next(item for item in effective if item["provider_id"] == "target") assert target["source"] == "database" assert client.get(f"/api/v1/providers/{provider_id}", headers=auth_headers).status_code == 200 updated = client.patch( f"/api/v1/providers/{provider_id}", headers=auth_headers, json={"model_name": "safe-model-v2"}, ) assert updated.status_code == 200 assert updated.json()["model_name"] == "safe-model-v2" assert updated.json()["api_key_configured"] is True assert ( client.delete(f"/api/v1/providers/{provider_id}", headers=auth_headers).status_code == 204 ) fallback = client.get(f"/api/v1/providers/{provider_id}", headers=auth_headers) assert fallback.status_code == 200 assert fallback.json()["source"] == "env" def test_provider_writes_require_admin(client, auth_headers): created = client.post( "/api/v1/auth/users", headers=auth_headers, json={"username": "operator", "password": "maxta2026", "is_admin": False}, ) assert created.status_code == 201 token = client.post( "/api/v1/auth/login", json={"username": "operator", "password": "maxta2026"} ).json()["access_token"] response = client.post( "/api/v1/providers", headers={"Authorization": f"Bearer {token}"}, json={ "provider_id": "judge", "base_url": "https://models.example/v1", "model_name": "judge-model", "auth_type": "none", }, ) assert response.status_code == 403 def test_admin_can_manage_users(client, auth_headers): created = client.post( "/api/v1/auth/users", headers=auth_headers, json={"username": "operator", "password": "maxta2026", "is_admin": False}, ) assert created.status_code == 201 user_id = created.json()["id"] assert created.json()["username"] == "operator" listed = client.get("/api/v1/auth/users", headers=auth_headers) assert listed.status_code == 200 assert {user["username"] for user in listed.json()} == {"auditor", "operator"} auditor_id = next(user["id"] for user in listed.json() if user["username"] == "auditor") operator_login = client.post( "/api/v1/auth/login", json={"username": "operator", "password": "maxta2026"}, ) assert operator_login.status_code == 200 assert ( client.get( "/api/v1/auth/users", headers={"Authorization": f"Bearer {operator_login.json()['access_token']}"}, ).status_code == 403 ) disabled = client.patch( f"/api/v1/auth/users/{user_id}", headers=auth_headers, json={"is_active": False} ) assert disabled.status_code == 200 assert disabled.json()["is_active"] is False assert client.delete(f"/api/v1/auth/users/{user_id}", headers=auth_headers).status_code == 204 assert client.delete(f"/api/v1/auth/users/{user_id}", headers=auth_headers).status_code == 404 assert "operator" not in { user["username"] for user in client.get("/api/v1/auth/users", headers=auth_headers).json() } assert ( client.post( "/api/v1/auth/login", json={"username": "operator", "password": "maxta2026"}, ).status_code == 401 ) assert ( client.delete(f"/api/v1/auth/users/{auditor_id}", headers=auth_headers).status_code == 409 ) def test_authenticated_user_can_read_own_profile(client, auth_headers): response = client.get("/api/v1/auth/me", headers=auth_headers) assert response.status_code == 200 assert response.json() == { "id": 1, "username": "auditor", "is_active": True, "is_admin": True, } def test_user_can_change_own_password_and_existing_tokens_are_revoked(client, auth_headers): client.post( "/api/v1/auth/users", headers=auth_headers, json={"username": "operator", "password": "old-password", "is_admin": False}, ) old_login = client.post( "/api/v1/auth/login", json={"username": "operator", "password": "old-password"} ).json() old_token = old_login["access_token"] headers = {"Authorization": f"Bearer {old_token}"} rejected = client.patch( "/api/v1/auth/password", headers=headers, json={"current_password": "wrong-password", "new_password": "new-password"}, ) assert rejected.status_code == 401 assert rejected.json()["error"]["code"] == "UNAUTHORIZED" changed = client.patch( "/api/v1/auth/password", headers=headers, json={"current_password": "old-password", "new_password": "new-password"}, ) assert changed.status_code == 204 assert client.get("/api/v1/auth/me", headers=headers).status_code == 401 assert ( client.post( "/api/v1/auth/refresh", json={"refresh_token": old_login["refresh_token"]}, headers={"Idempotency-Key": "after-password-change"}, ).status_code == 401 ) assert ( client.post( "/api/v1/auth/login", json={"username": "operator", "password": "old-password"} ).status_code == 401 ) assert ( client.post( "/api/v1/auth/login", json={"username": "operator", "password": "new-password"} ).status_code == 200 ) def test_admin_can_reset_password_and_existing_tokens_are_revoked(client, auth_headers): created = client.post( "/api/v1/auth/users", headers=auth_headers, json={"username": "operator", "password": "old-password", "is_admin": False}, ) user_id = created.json()["id"] old_token = client.post( "/api/v1/auth/login", json={"username": "operator", "password": "old-password"} ).json()["access_token"] reset = client.patch( f"/api/v1/auth/users/{user_id}/password", headers=auth_headers, json={"password": "new-password"}, ) assert reset.status_code == 204 assert ( client.post( "/api/v1/auth/login", json={"username": "operator", "password": "old-password"} ).status_code == 401 ) assert ( client.post( "/api/v1/auth/login", json={"username": "operator", "password": "new-password"} ).status_code == 200 ) assert ( client.get( "/api/v1/providers", headers={"Authorization": f"Bearer {old_token}"} ).status_code == 401 ) def test_reset_password_requires_admin(client, auth_headers): created = client.post( "/api/v1/auth/users", headers=auth_headers, json={"username": "operator", "password": "old-password", "is_admin": False}, ) token = client.post( "/api/v1/auth/login", json={"username": "operator", "password": "old-password"} ).json()["access_token"] assert ( client.patch( f"/api/v1/auth/users/{created.json()['id']}/password", headers={"Authorization": f"Bearer {token}"}, json={"password": "new-password"}, ).status_code == 403 ) def test_logout_revokes_current_token(client, auth_headers): assert client.post("/api/v1/auth/logout", headers=auth_headers).status_code == 204 assert client.get("/api/v1/providers/target/models", headers=auth_headers).status_code == 401 @pytest.fixture def auth_headers(client): token = client.post( "/api/v1/auth/login", json={"username": "auditor", "password": "correct-horse-battery-staple"}, ).json()["access_token"] return {"Authorization": f"Bearer {token}"} def test_provider_contracts(client, auth_headers): check = client.post("/api/v1/providers/target/check", headers=auth_headers) models = client.get("/api/v1/providers/judge/models", headers=auth_headers) assert check.status_code == models.status_code == 200 assert check.json()["model"] == "fake-model" assert models.json()["models"] == ["fake-model"] def test_run_and_report_contracts(client, auth_headers): created = client.post( "/api/v1/runs", json={"profile": "smoke", "auto_judge": False}, headers={**auth_headers, "Idempotency-Key": "api-contract-run"}, ) run_id = created.json()["run_id"] status = client.get(f"/api/v1/runs/{run_id}", headers=auth_headers) listed = client.get("/api/v1/runs", headers=auth_headers) results = client.get(f"/api/v1/runs/{run_id}/results", headers=auth_headers) reports = client.get("/api/v1/reports", headers=auth_headers) report = client.get(f"/api/v1/reports/{run_id}", headers=auth_headers) resumed = client.post(f"/api/v1/runs/{run_id}/resume", headers=auth_headers) assert created.status_code == resumed.status_code == 202 assert ( status.status_code == listed.status_code == results.status_code == reports.status_code == report.status_code == 200 ) assert listed.json()[0]["run_id"] == run_id assert results.json() == [] assert reports.json()[0]["run_id"] == run_id assert report.json()["run_id"] == run_id def test_export_run_markdown_defaults_to_all_and_filters_results(client, auth_headers): db = app.state.session_factory() run = RunRepository(db).create(run_type="safety_test", profile="all", selected_count=3) run.completed_count = 2 run.error_count = 1 RunRepository(db).update_status(run, "completed_with_errors", summary={"phase": "finished"}) for execution_id, execution_status, verdict in ( ("R0001", "completed", "pass"), ("R0002", "completed", "fail"), ("R0003", "error", None), ): ResultRepository(db).add( run_id=run.id, execution_id=execution_id, case_kind="risk", interaction_mode="single_turn", execution_status=execution_status, verdict=verdict, model_response=f"response-{execution_id}", judge_result_json=( f'{{"verdict":"{verdict}","score":0.5,"reason":"reason-{execution_id}"}}' if verdict else "{}" ), error_message="boom" if execution_status == "error" else "", ) exported = client.get(f"/api/v1/runs/{run.id}/export", headers=auth_headers) assert exported.status_code == 200 assert exported.headers["content-type"].startswith("text/markdown") assert exported.headers["content-disposition"] == f'attachment; filename="run_{run.id}_results.md"' assert all(execution_id in exported.text for execution_id in ("R0001", "R0002", "R0003")) failed = client.get( f"/api/v1/runs/{run.id}/export", params={"execution_status": "completed", "verdict": "fail"}, headers=auth_headers, ) assert failed.status_code == 200 assert "R0002" in failed.text assert "R0001" not in failed.text assert "R0003" not in failed.text unjudged = client.get( f"/api/v1/runs/{run.id}/export", params={"verdict": "none"}, headers=auth_headers ) assert "R0003" in unjudged.text assert "R0001" not in unjudged.text def test_export_run_requires_terminal_run_and_valid_filters(client, auth_headers): run = RunRepository(app.state.session_factory()).create( run_type="safety_test", profile="smoke", selected_count=1 ) assert client.get(f"/api/v1/runs/{run.id}/export", headers=auth_headers).status_code == 409 assert client.get("/api/v1/runs/999/export", headers=auth_headers).status_code == 404 assert ( client.get( f"/api/v1/runs/{run.id}/export", params={"execution_status": "invalid"}, headers=auth_headers, ).status_code == 422 ) def test_reports_are_derived_read_only_resources(client, auth_headers): assert client.get("/api/v1/reports/999", headers=auth_headers).status_code == 404 assert client.post("/api/v1/reports", headers=auth_headers).status_code == 405 assert client.patch("/api/v1/reports/1", headers=auth_headers).status_code == 405 assert client.delete("/api/v1/reports/1", headers=auth_headers).status_code == 405 def test_run_cancel_and_delete_contract(client, auth_headers): created = client.post( "/api/v1/runs", json={"profile": "smoke", "auto_judge": False}, headers=auth_headers, ) run_id = created.json()["run_id"] assert client.delete(f"/api/v1/runs/{run_id}", headers=auth_headers).status_code == 409 cancelled = client.patch( f"/api/v1/runs/{run_id}", json={"status": "cancelled"}, headers=auth_headers, ) assert cancelled.status_code == 200 assert cancelled.json()["status"] == "cancelled" assert cancelled.json()["terminal"] is True assert ( client.patch( f"/api/v1/runs/{run_id}", json={"status": "cancelled"}, headers=auth_headers, ).status_code == 409 ) assert client.delete(f"/api/v1/runs/{run_id}", headers=auth_headers).status_code == 204 assert client.delete(f"/api/v1/runs/{run_id}", headers=auth_headers).status_code == 404 assert client.get(f"/api/v1/runs/{run_id}", headers=auth_headers).status_code == 404 def test_retry_errors_keeps_successes_and_requeues_only_errors(client, auth_headers): created = client.post( "/api/v1/runs", json={"profile": "smoke", "auto_judge": False}, headers=auth_headers, ) run_id = created.json()["run_id"] db = app.state.session_factory() run_repo = RunRepository(db) run = run_repo.get(run_id) run.selected_count = 2 run.completed_count = 1 run.error_count = 1 run_repo.update_status( run, "completed_with_errors", summary={"phase": "finished", "auto_judge": False}, ) for execution_id, execution_status in (("case-ok", "completed"), ("case-error", "error")): ResultRepository(db).add( run_id=run_id, execution_id=execution_id, case_kind="control", interaction_mode="single_turn", execution_status=execution_status, verdict=None, model_response="ok" if execution_status == "completed" else "", judge_result_json="{}", error_message="boom" if execution_status == "error" else "", ) assert client.post(f"/api/v1/runs/{run_id}/resume", headers=auth_headers).status_code == 409 retried = client.post(f"/api/v1/runs/{run_id}/retry-errors", headers=auth_headers) assert retried.status_code == 202 assert retried.json()["retry_count"] == 1 assert [row.execution_status for row in ResultRepository(db).list_by_run(run_id)] == [ "completed" ] run = RunRepository(db).get(run_id) assert (run.status, run.completed_count, run.error_count) == ("pending", 1, 0) assert ( client.post(f"/api/v1/runs/{run_id}/retry-errors", headers=auth_headers).status_code == 409 ) run = RunRepository(db).get(run_id) RunRepository(db).update_status( run, "completed_with_errors", summary={"phase": "finished", "auto_judge": False}, ) assert ( client.post(f"/api/v1/runs/{run_id}/retry-errors", headers=auth_headers).status_code == 409 ) def test_retry_one_result_requeues_only_selected_execution(client, auth_headers): created = client.post( "/api/v1/runs", json={"profile": "smoke", "auto_judge": False}, headers=auth_headers, ) run_id = created.json()["run_id"] db = app.state.session_factory() run_repo = RunRepository(db) run = run_repo.get(run_id) run.selected_count = run.completed_count = 2 run_repo.update_status( run, "completed", summary={"phase": "finished", "auto_judge": False}, ) for execution_id in ("case-keep", "case-retry"): ResultRepository(db).add( run_id=run_id, execution_id=execution_id, case_kind="control", interaction_mode="single_turn", execution_status="completed", verdict="pass", model_response="ok", judge_result_json='{"verdict":"pass"}', error_message="", ) assert ( client.post( f"/api/v1/runs/{run_id}/results/missing/retry", headers=auth_headers, ).status_code == 404 ) retried = client.post( f"/api/v1/runs/{run_id}/results/case-retry/retry", headers=auth_headers, ) assert retried.status_code == 202 assert retried.json() == { "run_id": run_id, "status": "pending", "selected_count": 2, "execution_id": "case-retry", } assert [row.execution_id for row in ResultRepository(db).list_by_run(run_id)] == ["case-keep"] run = run_repo.get(run_id) assert (run.status, run.completed_count, run.error_count) == ("pending", 1, 0) assert ( client.post( f"/api/v1/runs/{run_id}/results/case-retry/retry", headers=auth_headers, ).status_code == 409 ) def test_run_contract_errors(client, auth_headers): assert client.get("/api/v1/runs/999", headers=auth_headers).status_code == 404 assert client.get("/api/v1/runs/999/results", headers=auth_headers).status_code == 404 assert ( client.post("/api/v1/runs/999/results/R0001/retry", headers=auth_headers).status_code == 404 ) assert ( client.post("/api/v1/runs", json={"profile": "invalid"}, headers=auth_headers).status_code == 422 )