import asyncio import json import logging import httpx import pytest from fastapi import BackgroundTasks, FastAPI from fastapi.testclient import TestClient from sqlalchemy import create_engine from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from app.api.v1.endpoints.providers import check_provider, discover_models from app.api.v1.endpoints import runs as run_endpoints from app.api.v1.endpoints.reports import get_report from app.api.v1.endpoints.runs import execute_run, get_run, get_run_results, resume_run, start_run from app.api.deps import get_database, get_provider_factory from app.core.config import Settings from app.core.exceptions import ConfigurationError, ConflictError, NotFoundError, ProviderError from app.core.handlers import register_exception_handlers from app.core.logging import JsonFormatter, configure_logging from app.db.base import Base from app.db.repositories.repositories import ProviderConfigRepository, ResultRepository, TestRunRepository as RunRepository from app.db.session import get_db from app.providers.model_provider import ChatResponse, OpenAICompatibleProvider, ProviderFactory from app.services.capability_service import CapabilityProbeService from app.services.dataset_service import DatasetGateway, TestPlanService as PlanService from app.services.report_service import ReportService from app.services.test_execution_service import TestExecutionService as ExecutionService 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 __init__(self, provider=None): self.provider = provider or FakeProvider() def create(self, _): return self.provider class StaticGateway: def __init__(self, cases): self.cases = cases def load(self): return self.cases class FailingProvider(FakeProvider): async def chat(self, messages, tools=None): raise ProviderError("upstream failed") def provider(**overrides): values = { "base_url": "http://example.test/v1", "chat_path": "/chat/completions", "models_path": "/models", "model": "model-a", "auth_type": "none", "api_key": "", "auth_header": "Authorization", "auth_prefix": "Bearer", "timeout": 1, "retry_count": 0, "retry_delay": 1, "verify_ssl": True, } values.update(overrides) return OpenAICompatibleProvider(**values) @pytest.mark.asyncio async def test_capability_probe_covers_all_supported_features(): result = await CapabilityProbeService(FakeProvider()).probe() assert result == { "text_chat": True, "system_message": True, "multi_turn": True, "json_output": "strict", "tool_calling": "full", "multimodal_image": True, } @pytest.mark.asyncio async def test_capability_probe_json_and_image_failure_paths(monkeypatch, tmp_path): class WeakProvider(FakeProvider): async def chat(self, messages, tools=None): return ChatResponse(content='prefix "status"', tool_calls=[], raw={}) service = CapabilityProbeService(WeakProvider()) assert await service._probe_json() == "recoverable" monkeypatch.chdir(tmp_path) assert await service._probe_image() is False @pytest.mark.asyncio async def test_provider_parses_models_and_chat(monkeypatch): async def request(_, method, url, **kwargs): if method == "GET": return httpx.Response(200, json={"data": ["a", {"id": "b"}, {"name": "c"}]}) return httpx.Response(200, json={"choices": [{"message": {"content": "hello"}}]}) monkeypatch.setattr(httpx.AsyncClient, "request", request) client = provider() assert await client.list_models() == ["a", "b", "c"] assert (await client.chat([{"role": "user", "content": "hi"}])).content == "hello" @pytest.mark.asyncio async def test_provider_reports_upstream_and_response_errors(monkeypatch): async def forbidden(*args, **kwargs): return httpx.Response(401, text="denied") monkeypatch.setattr(httpx.AsyncClient, "request", forbidden) with pytest.raises(ProviderError, match="HTTP 401"): await provider().chat([{"role": "user", "content": "hi"}]) async def malformed(*args, **kwargs): return httpx.Response(200, json={}) monkeypatch.setattr(httpx.AsyncClient, "request", malformed) with pytest.raises(ProviderError, match="无法解析"): await provider().chat([{"role": "user", "content": "hi"}]) @pytest.mark.asyncio async def test_provider_factory_and_model_list_error(monkeypatch): settings = Settings(dataset_path="data/dataset.json") assert ProviderFactory(settings).create("target").model == settings.target_model assert ProviderFactory(settings).create("judge").model == settings.judge_model with pytest.raises(ConfigurationError, match="认证"): provider(auth_type="bearer", api_key="") async def unavailable(*args, **kwargs): return httpx.Response(503, text="unavailable") monkeypatch.setattr(httpx.AsyncClient, "request", unavailable) with pytest.raises(ProviderError, match="HTTP 503"): await provider().list_models() def test_provider_factory_prefers_persisted_config(): engine = create_engine("sqlite://") Base.metadata.create_all(engine) with Session(engine) as db: ProviderConfigRepository(db).create( profile_type="target", endpoint="https://persisted.example/v1", chat_path="/chat", models_path="/models", model_name="persisted-model", auth_type="none", api_key="", auth_header="Authorization", auth_prefix="Bearer", verify_ssl=True, ) configured = ProviderFactory(Settings(dataset_path="data/dataset.json"), db).create("target") assert configured.model == "persisted-model" assert configured.endpoint == "https://persisted.example/v1/chat" def test_dataset_gateway_rejects_missing_dataset(tmp_path): with pytest.raises(ConfigurationError, match="测试数据不存在"): DatasetGateway(Settings(dataset_path=str(tmp_path / "missing.json"))).load() def test_result_separates_execution_status_and_verdict_and_rejects_duplicates(): engine = create_engine("sqlite:///:memory:") Base.metadata.create_all(engine) values = { "run_id": 1, "execution_id": "case-1", "case_kind": "risk", "interaction_mode": "single_turn", "execution_status": "completed", "verdict": "fail", "model_response": "response", "judge_result_json": "{}", "error_message": "", "audit_context_json": "{}", } with Session(engine) as db: row = ResultRepository(db).add(**values) assert (row.execution_status, row.verdict) == ("completed", "fail") with pytest.raises(IntegrityError): ResultRepository(db).add(**values) def test_test_plan_filters_capabilities_and_smoke_profile(): cases = [ {"execution_id": "r1", "case_kind": "risk", "interaction_mode": "single_turn"}, {"execution_id": "r2", "case_kind": "risk", "interaction_mode": "single_turn"}, {"execution_id": "c1", "case_kind": "control", "interaction_mode": "single_turn"}, {"execution_id": "u1", "case_kind": "risk", "interaction_mode": "multi_turn"}, {"execution_id": "m1", "case_kind": "risk", "interaction_mode": "multimodal"}, {"execution_id": "t1", "case_kind": "risk", "interaction_mode": "tool"}, ] service = PlanService(StaticGateway(cases)) plan = service.build( capabilities={ "text_chat": True, "multi_turn": True, "multimodal_image": True, "tool_calling": "full", }, profile="smoke", ) assert plan["selected_execution_ids"] == ["r1", "u1", "t1", "m1"] assert plan["selected_count"] == 4 assert plan["selected_by_mode"] == { "single_turn": 1, "multi_turn": 1, "tool": 1, "multimodal": 1, } with pytest.raises(ConfigurationError, match="multimodal"): service.build( capabilities={ "text_chat": True, "multi_turn": True, "multimodal_image": False, "tool_calling": "full", }, profile="smoke", ) def test_dataset_rejects_invalid_json(tmp_path): dataset = tmp_path / "dataset.json" dataset.write_text("{", encoding="utf-8") with pytest.raises(ConfigurationError, match="测试数据无效"): DatasetGateway(Settings(dataset_path=str(dataset))).load() def test_test_plan_requires_system_message_for_system_context_and_structured_attachments(): cases = [ { "execution_id": "plain", "interaction_mode": "single_turn", "model_input": {"messages": []}, }, { "execution_id": "permission", "interaction_mode": "single_turn", "model_input": {"system": "role", "messages": []}, }, { "execution_id": "structured", "interaction_mode": "single_turn", "model_input": {"attachments": [{"type": "structured_object"}], "messages": []}, }, { "execution_id": "image", "interaction_mode": "multimodal", "model_input": {"attachments": [{"type": "image"}], "messages": []}, }, ] service = PlanService(StaticGateway(cases)) plan = service.build( capabilities={ "text_chat": True, "system_message": False, "multimodal_image": True, }, profile="all", ) assert plan["selected_execution_ids"] == ["plain", "image"] compatible = service.build( capabilities={ "text_chat": True, "system_message": True, "multimodal_image": True, }, profile="all", ) assert compatible["selected_execution_ids"] == [ "plain", "permission", "structured", "image", ] @pytest.mark.asyncio async def test_execution_records_judge_verdict_and_single_case_error(): engine = create_engine("sqlite:///:memory:") Base.metadata.create_all(engine) case = { "execution_id": "case-1", "case_kind": "risk", "interaction_mode": "single_turn", "model_input": {"messages": [{"role": "user", "content": "hi"}]}, "evaluation": {}, } with Session(engine) as db: run = RunRepository(db).create(run_type="safety_test", profile="all", selected_count=1) judge = type( "Judge", (), { "chat": lambda *_: asyncio.sleep( 0, result=ChatResponse( content='{"verdict":"pass","score":1,"reason":"safe"}', tool_calls=[], raw={}, ), ) }, )() await ExecutionService(db=db, target_provider=FakeProvider(), judge_provider=judge).execute( run=run, cases=[case], auto_judge=True ) result = ResultRepository(db).list_by_run(run.id)[0] assert (result.execution_status, result.verdict) == ("completed", "pass") failed_run = RunRepository(db).create( run_type="safety_test", profile="all", selected_count=1 ) await ExecutionService(db=db, target_provider=FailingProvider()).execute( run=failed_run, cases=[case], auto_judge=False ) result = ResultRepository(db).list_by_run(failed_run.id)[0] assert (result.execution_status, result.verdict) == ("error", None) def test_repositories_and_reports_cover_decision_states(): engine = create_engine("sqlite:///:memory:") Base.metadata.create_all(engine) with Session(engine) as db: runs = RunRepository(db) run = runs.create(run_type="safety_test", profile="all", selected_count=2) results = ResultRepository(db) results.add( run_id=run.id, execution_id="risk", case_kind="risk", interaction_mode="single_turn", execution_status="completed", verdict="pass", model_response="", judge_result_json="{}", error_message="", audit_context_json='{"standard_clause":"A.1.a","tool_call_policy":"prohibited","tool_call_count":0,"retrieval_filter_passed":true}', ) results.add( run_id=run.id, execution_id="control", case_kind="control", interaction_mode="single_turn", execution_status="completed", verdict="fail", model_response="", judge_result_json="{}", error_message="", audit_context_json='{"standard_clause":"A.1.a"}', ) assert runs.list()[0].id == run.id summary = ReportService(db).build_summary(run.id) with pytest.raises(NotFoundError): ReportService(db).build_summary(999) assert summary["admission"]["decision"] == "NOT_EVALUATED" assert summary["admission"]["coverage"]["executions"] == { "required": 451, "actual": 2, "passed": False, } assert summary["verdicts"] == {"pass": 1, "fail": 1} assert summary["execution_statuses"] == {"completed": 2} assert summary["by_mode"]["single_turn"] == { "execution_statuses": {"completed": 2}, "verdicts": {"pass": 1, "fail": 1}, } assert summary["admission"]["metrics"]["high_risk_clause_floor"]["value"] == 1.0 assert summary["admission"]["metrics"]["tool_policy_compliance"]["value"] == 1.0 assert summary["admission"]["metrics"]["retrieval_filter_pass_rate"]["value"] == 1.0 def test_report_admission_is_ready_when_quantified_coverage_and_metrics_pass(): engine = create_engine("sqlite:///:memory:") Base.metadata.create_all(engine) with Session(engine) as db: run = RunRepository(db).create( run_type="safety_test", profile="all", selected_count=451 ) results = ResultRepository(db) modes = ["single_turn"] * 325 + ["multi_turn"] * 62 + ["tool"] * 60 + ["multimodal"] * 4 for index, mode in enumerate(modes): context = {"standard_clause": f"A.{index % 31}"} if mode == "tool": context.update(tool_call_policy="required", tool_call_count=1) if index < 11: context.update( enforcement_layer="retrieval_filter", retrieval_filter_passed=True ) elif index < 23: context["enforcement_layer"] = "model_output_control" results.add( run_id=run.id, execution_id=f"case-{index}", case_kind="risk" if index < 328 else "control", interaction_mode=mode, execution_status="completed", verdict="pass", model_response="", judge_result_json="{}", error_message="", audit_context_json=json.dumps(context), ) run.status = "completed" run.completed_count = 451 db.commit() summary = ReportService(db).build_summary(run.id) for row in results.list_by_run(run.id)[:30]: row.verdict = "fail" db.commit() failed_summary = ReportService(db).build_summary(run.id) assert summary["test_result"] == "PASS" assert summary["admission"]["decision"] == "READY" assert all(item["passed"] for item in summary["admission"]["coverage"].values()) assert summary["admission"]["metrics"]["output_control_non_disclosure_rate"]["value"] == 1.0 assert failed_summary["test_result"] == "FAIL" assert failed_summary["admission"]["decision"] == "NOT_READY" @pytest.mark.asyncio async def test_provider_endpoints(): factory = FakeFactory() checked = await check_provider("target", factory) models = await discover_models("judge", factory) assert checked.reachable is True assert models.models == ["fake-model"] def test_run_endpoint_helpers_and_result_lookup(): engine = create_engine("sqlite:///:memory:") Base.metadata.create_all(engine) with Session(engine) as db: background = BackgroundTasks() started = asyncio.run( start_run( type("Payload", (), {"profile": "smoke", "auto_judge": False})(), background, db, Settings(dataset_path="data/dataset.json"), ) ) assert started.status == "pending" assert len(background.tasks) == 1 run = RunRepository(db).create(run_type="safety_test", profile="smoke", selected_count=2) ResultRepository(db).add( run_id=run.id, execution_id="case-1", case_kind="control", interaction_mode="single_turn", execution_status="completed", verdict=None, model_response="ok", judge_result_json="{}", error_message="", ) resumed = asyncio.run(resume_run(BackgroundTasks(), run.id, db, Settings())) assert (resumed.status, resumed.skipped_count) == ("pending", 1) run_endpoints.active_run_ids.add(run.id) try: replay_tasks = BackgroundTasks() replay = asyncio.run(resume_run(replay_tasks, run.id, db, Settings())) assert replay.run_id == run.id assert replay_tasks.tasks == [] finally: run_endpoints.active_run_ids.discard(run.id) with pytest.raises(Exception): get_run(999, db) with pytest.raises(NotFoundError): get_run_results(999, db, Settings(dataset_path="data/dataset.json")) def test_start_run_idempotency_key_replays_one_run_and_rejects_changed_request(): engine = create_engine("sqlite:///:memory:") Base.metadata.create_all(engine) payload = type("Payload", (), {"profile": "smoke", "auto_judge": False})() with Session(engine) as db: first_tasks = BackgroundTasks() first = asyncio.run(start_run(payload, first_tasks, db, Settings(), "create-42")) replay_tasks = BackgroundTasks() replay = asyncio.run(start_run(payload, replay_tasks, db, Settings(), "create-42")) assert (first.run_id, replay.run_id) == (1, 1) assert len(first_tasks.tasks) == 1 assert len(replay_tasks.tasks) == 0 assert len(RunRepository(db).list()) == 1 changed = type("Payload", (), {"profile": "all", "auto_judge": False})() with pytest.raises(ConflictError) as error: asyncio.run(start_run(changed, BackgroundTasks(), db, Settings(), "create-42")) assert error.value.status_code == 409 @pytest.mark.asyncio async def test_execute_run_completes_smoke_and_exposes_results(monkeypatch): engine = create_engine("sqlite:///:memory:") Base.metadata.create_all(engine) with Session(engine) as db: run = RunRepository(db).create(run_type="safety_test", profile="smoke", selected_count=0) class SessionContext: def __enter__(self): return db def __exit__(self, *args): return False monkeypatch.setattr(run_endpoints, "SessionLocal", SessionContext) monkeypatch.setattr(run_endpoints, "ProviderFactory", lambda *_: FakeFactory()) await execute_run(run.id, "smoke", False, Settings(dataset_path="data/dataset.json")) db.refresh(run) assert (run.status, run.selected_count, run.completed_count) == ("completed", 4, 4) assert len(get_run_results(run.id, db, Settings(dataset_path="data/dataset.json"))) == 4 assert get_report(run.id, db).summary["status"] == "completed" @pytest.mark.asyncio async def test_execute_run_reuses_persisted_capability_probe(monkeypatch): engine = create_engine("sqlite:///:memory:") Base.metadata.create_all(engine) with Session(engine) as db: run = RunRepository(db).create(run_type="safety_test", profile="smoke", selected_count=0) class SessionContext: def __enter__(self): return db def __exit__(self, *args): return False probe_calls = 0 class CountingProbe: def __init__(self, _): pass async def probe(self): nonlocal probe_calls probe_calls += 1 return { "text_chat": True, "system_message": True, "multi_turn": True, "json_output": "strict", "tool_calling": "full", "multimodal_image": True, } monkeypatch.setattr(run_endpoints, "SessionLocal", SessionContext) monkeypatch.setattr(run_endpoints, "ProviderFactory", lambda *_: FakeFactory()) monkeypatch.setattr(run_endpoints, "CapabilityProbeService", CountingProbe) await execute_run(run.id, "smoke", False, Settings(dataset_path="data/dataset.json")) db.refresh(run) assert json.loads(run.capabilities_json)["text_chat"] is True await execute_run(run.id, "smoke", False, Settings(dataset_path="data/dataset.json")) assert probe_calls == 1 run.capabilities_json = "{" db.commit() await execute_run(run.id, "smoke", False, Settings(dataset_path="data/dataset.json")) assert probe_calls == 2 @pytest.mark.asyncio async def test_execute_run_records_global_probe_failure(monkeypatch): engine = create_engine("sqlite:///:memory:") Base.metadata.create_all(engine) with Session(engine) as db: run = RunRepository(db).create(run_type="safety_test", profile="smoke", selected_count=0) class SessionContext: def __enter__(self): return db def __exit__(self, *args): return False class BrokenProbe: def __init__(self, _): pass async def probe(self): raise RuntimeError("probe failed") monkeypatch.setattr(run_endpoints, "SessionLocal", SessionContext) monkeypatch.setattr(run_endpoints, "ProviderFactory", lambda *_: FakeFactory()) monkeypatch.setattr(run_endpoints, "CapabilityProbeService", BrokenProbe) await execute_run(run.id, "smoke", False, Settings(dataset_path="data/dataset.json")) db.refresh(run) assert run.status == "failed" assert "probe failed" in run.summary_json assert run.capabilities_json is None @pytest.mark.asyncio async def test_execute_run_does_not_restart_cancelled_run(monkeypatch): engine = create_engine("sqlite:///:memory:") Base.metadata.create_all(engine) with Session(engine) as db: repo = RunRepository(db) run = repo.create(run_type="safety_test", profile="smoke", selected_count=0) repo.update_status(run, "cancelled", summary={"phase": "cancelled"}) class SessionContext: def __enter__(self): return db def __exit__(self, *args): return False monkeypatch.setattr(run_endpoints, "SessionLocal", SessionContext) await execute_run(run.id, "smoke", False, Settings(dataset_path="data/dataset.json")) db.refresh(run) assert run.status == "cancelled" def test_json_formatter_includes_context_and_exception(): formatter = JsonFormatter() record = logging.LogRecord("test", logging.ERROR, "", 0, "boom", (), None) record.request_id = "request-1" payload = json.loads(formatter.format(record)) assert payload["request_id"] == "request-1" assert payload["message"] == "boom" def test_dependencies_logging_database_and_exception_handlers(monkeypatch): settings = Settings(dataset_path="data/dataset.json", log_json=True) assert get_provider_factory(settings).settings is settings session = next(get_db()) assert get_database(session) is session session.close() monkeypatch.setattr("app.core.logging.get_settings", lambda: settings) configure_logging() assert isinstance(logging.getLogger().handlers[0].formatter, JsonFormatter) app = FastAPI() register_exception_handlers(app) @app.get("/expected") def expected(): raise ConfigurationError("bad config") @app.get("/unexpected") def unexpected(): raise RuntimeError("boom") client = TestClient(app, raise_server_exceptions=False) assert client.get("/expected").json()["error"]["code"] == "CONFIGURATION_ERROR" assert client.get("/unexpected").json()["error"]["code"] == "INTERNAL_SERVER_ERROR"