import unittest from types import SimpleNamespace from fastapi.testclient import TestClient from app.api.system import ( get_audit_log_service, get_governance_service, get_shared_ai_platform_client, get_template_service, require_current_user, ) from app.core.config import settings from app.core.security import get_password_hash from app.db import get_db_session from app.main import app class StubScalarResult: def __init__(self, user=None): self.user = user def scalar(self): return 1 def scalar_one_or_none(self): return self.user class StubReadyDb: def __init__(self, password="rotated-secret"): self.user = SimpleNamespace( username="admin@example.com", password_hash=get_password_hash(password), ) def execute(self, _statement): return StubScalarResult(self.user) class StubBrokenDb: def execute(self, _statement): raise RuntimeError("database unavailable") class StubSharedAIPlatformClient: def __init__(self): self.knowledge_bases = [] self.documents_by_kb = {} def list_knowledge_bases(self, current_user, page, page_size): if self.knowledge_bases: return { "items": self.knowledge_bases, "total": len(self.knowledge_bases), "page": page, "page_size": page_size, } return { "items": [ { "id": "kb-123", "name": "Shared AI KB", "status": "active", "document_count": 3, "retrieval_mode": "hybrid", "updated_at": "2026-05-04T09:00:00+00:00", } ], "total": 1, "page": page, "page_size": page_size, } def create_knowledge_base(self, current_user, name, description): item = { "id": "kb-demo-001", "name": name, "description": description, "status": "active", "document_count": 0, "retrieval_mode": "hybrid", "updated_at": "2026-05-12T09:00:00+00:00", } self.knowledge_bases.append(item) self.documents_by_kb[item["id"]] = [] return item def list_documents(self, current_user, knowledge_base_id, page, page_size): items = self.documents_by_kb.get(knowledge_base_id, []) return { "items": items, "total": len(items), "page": page, "page_size": page_size, } def upload_document(self, current_user, knowledge_base_id, file_name, mime_type, content): item = { "id": "kbdoc-demo-001", "knowledge_base_id": knowledge_base_id, "file_name": file_name, "mime_type": mime_type, "file_size": len(content), "status": "indexed", "current_stage": "indexed", "parse_status": "parsed", "index_status": "indexed", "updated_at": "2026-05-12T09:01:00+00:00", } self.documents_by_kb.setdefault(knowledge_base_id, []).append(item) for knowledge_base in self.knowledge_bases: if knowledge_base["id"] == knowledge_base_id: knowledge_base["document_count"] = len(self.documents_by_kb[knowledge_base_id]) return item class StubTemplateService: def seed_default_templates(self, current_user): return { "created": [{"id": "tpl-demo-001", "name": "Procurement Template"}], "existing": [], "created_count": 1, "existing_count": 0, "total": 3, } def list_templates(self, current_user, page, page_size, contract_type): return { "items": [ { "id": "tpl-123", "name": "Procurement Template", "contract_type": "procurement", "enabled": True, "updated_at": "2026-05-04T09:01:00+00:00", }, { "id": "tpl-456", "name": "NDA Template", "contract_type": "nda", "enabled": False, "updated_at": "2026-05-04T09:02:00+00:00", }, ], "total": 2, "page": page, "page_size": page_size, } class StubAuditLogService: def __init__(self): self.events = [] def record_event(self, **kwargs): self.events.append(kwargs) def list_logs(self, current_user, page, page_size, action): return { "items": [ { "id": "audit-123", "action": "upload_knowledge_document", "resource_type": "knowledge_base_document", "created_at": "2026-05-04T09:03:00+00:00", "operator": {"id": "user-admin", "display_name": "Admin"}, } ], "total": 7, "page": page, "page_size": page_size, } class StubGovernanceService: def get_release_summary(self, current_user): return { "summary": { "total_assets": 8, "pending_assignments": 2, "in_review_assignments": 1, "blocking_items": [{"type": "assignment", "status": "pending", "count": 2}], }, "recent_batches": [ { "id": "govrel-123", "batch_name": "formal-launch-import", "status": "blocked", } ], } def list_assignments(self, current_user, page, page_size, assignment_role, status_filter, assignee_user_id, governance_asset_id): return { "items": [ { "id": "govasg-123", "assignment_role": "legal_reviewer", "assignee_name": "Admin", "status": "pending", "priority": "high", "asset_name": "Procurement Template", } ], "total": 1, "page": page, "page_size": page_size, } class SystemApiTestCase(unittest.TestCase): def setUp(self): self.original_shared_ai_platform_base_url = settings.shared_ai_platform_base_url self.original_shared_ai_platform_api_key = settings.shared_ai_platform_api_key self.original_shared_ai_platform_api_prefix = settings.shared_ai_platform_api_prefix self.original_shared_ai_platform_app_id = settings.shared_ai_platform_app_id self.original_shared_ai_platform_timeout_seconds = settings.shared_ai_platform_timeout_seconds self.original_shared_ai_platform_legal_review_enabled = settings.shared_ai_platform_legal_review_enabled self.original_shared_ai_platform_legal_review_fallback_to_local = settings.shared_ai_platform_legal_review_fallback_to_local def tearDown(self): settings.shared_ai_platform_base_url = self.original_shared_ai_platform_base_url settings.shared_ai_platform_api_key = self.original_shared_ai_platform_api_key settings.shared_ai_platform_api_prefix = self.original_shared_ai_platform_api_prefix settings.shared_ai_platform_app_id = self.original_shared_ai_platform_app_id settings.shared_ai_platform_timeout_seconds = self.original_shared_ai_platform_timeout_seconds settings.shared_ai_platform_legal_review_enabled = self.original_shared_ai_platform_legal_review_enabled settings.shared_ai_platform_legal_review_fallback_to_local = self.original_shared_ai_platform_legal_review_fallback_to_local app.dependency_overrides.clear() def test_readiness_returns_pilot_assets(self): app.dependency_overrides[get_db_session] = lambda: StubReadyDb() client = TestClient(app) response = client.get("/api/v1/system/readiness") self.assertEqual(response.status_code, 200) payload = response.json()["data"] self.assertEqual(payload["status"], "ready") self.assertEqual(payload["pilot_assets"]["default_admin"]["username"], "admin@example.com") self.assertTrue(payload["pilot_assets"]["default_admin"]["password_rotated"]) self.assertGreaterEqual(len(payload["pilot_assets"]["demo_flow"]), 4) self.assertEqual(payload["pilot_scope"]["status"], "frozen") self.assertTrue(any(item["key"] == "review_chat_followup" for item in payload["pilot_scope"]["included"])) self.assertTrue(any(item["key"] == "complex_approval" for item in payload["pilot_scope"]["excluded"])) self.assertEqual(payload["pilot_materials"]["status"], "ready") self.assertTrue(any(item["key"] == "feedback_template" for item in payload["pilot_materials"]["materials"])) self.assertEqual(payload["pilot_error_feedback"]["status"], "ready") self.assertTrue(any(item["key"] == "chat_response_failed" for item in payload["pilot_error_feedback"]["items"])) self.assertEqual(payload["pilot_launch_pack"]["status"], "ready") self.assertTrue(any(item["key"] == "review_chat_followup" for item in payload["pilot_launch_pack"]["script_steps"])) self.assertEqual(payload["p0_mainline"]["status"], "aligned") self.assertEqual(len(payload["p0_mainline"]["capabilities"]), 5) self.assertTrue(all(item["source_project"] for item in payload["p0_mainline"]["capabilities"])) self.assertTrue(all(item["target_layer"] for item in payload["p0_mainline"]["capabilities"])) self.assertTrue(any(item["key"] == "database" and item["status"] == "ok" for item in payload["checks"])) self.assertTrue(any(item["key"] == "default_admin_password" and item["status"] == "ok" for item in payload["checks"])) def test_p0_mainline_returns_source_ownership_and_acceptance(self): client = TestClient(app) response = client.get("/api/v1/system/p0-mainline") self.assertEqual(response.status_code, 200) payload = response.json()["data"] self.assertEqual(payload["version"], "p0-mainline-alignment-2026-05-12") self.assertEqual(payload["status"], "aligned") capability_keys = {item["key"] for item in payload["capabilities"]} self.assertSetEqual( capability_keys, { "p0_a_document_parsing", "p0_b_shared_ai_orchestration", "p0_c_chat_experience", "p0_d_product_shell", "p0_e_auto_review_loop", }, ) self.assertTrue(all(item["source_project"] for item in payload["capabilities"])) self.assertTrue(all(item["target_layer"] for item in payload["capabilities"])) self.assertTrue(all(item["contract_platform_responsibility"] for item in payload["capabilities"])) self.assertTrue(all(item["acceptance"] for item in payload["capabilities"])) self.assertTrue(any("Dify" in item["label"] for item in payload["guardrails"])) self.assertIn("Slice 1-8", payload["phase_boundary"]["mainline_scope"]) def test_pilot_scope_returns_frozen_boundary(self): client = TestClient(app) response = client.get("/api/v1/system/pilot-scope") self.assertEqual(response.status_code, 200) payload = response.json()["data"] self.assertEqual(payload["version"], "pilot-scope-2026-05-12") self.assertEqual(payload["status"], "frozen") self.assertGreaterEqual(len(payload["included"]), 7) self.assertGreaterEqual(len(payload["excluded"]), 4) self.assertGreaterEqual(len(payload["defect_policy"]), 3) self.assertTrue(any(item["key"] == "knowledge_enhancement" for item in payload["included"])) self.assertTrue(any(item["level"] == "must_fix" for item in payload["defect_policy"])) def test_pilot_materials_returns_required_pack(self): client = TestClient(app) response = client.get("/api/v1/system/pilot-materials") self.assertEqual(response.status_code, 200) payload = response.json()["data"] self.assertEqual(payload["version"], "pilot-materials-2026-05-12") self.assertEqual(payload["status"], "ready") material_keys = {item["key"] for item in payload["materials"]} self.assertSetEqual( material_keys, {"pilot_brief", "user_guide", "admin_guide", "feedback_template", "faq"}, ) self.assertGreaterEqual(len(payload["feedback_template"]), 5) self.assertTrue(any(item["key"] == "result_rating" for item in payload["feedback_template"])) self.assertGreaterEqual(len(payload["acceptance"]), 3) def test_pilot_error_feedback_returns_required_guide(self): client = TestClient(app) response = client.get("/api/v1/system/pilot-error-feedback") self.assertEqual(response.status_code, 200) payload = response.json()["data"] self.assertEqual(payload["version"], "pilot-error-feedback-2026-05-12") self.assertEqual(payload["status"], "ready") feedback_keys = {item["key"] for item in payload["items"]} self.assertSetEqual( feedback_keys, { "ocr_pdf_parse_failed", "pdf_quality_insufficient", "auto_review_failed", "chat_response_failed", }, ) self.assertTrue(all(item["user_message"] for item in payload["items"])) self.assertTrue(all(item["next_actions"] for item in payload["items"])) self.assertTrue(any(item["qa_selector"] == "chat-exception-feedback" for item in payload["items"])) self.assertGreaterEqual(len(payload["acceptance"]), 4) def test_pilot_launch_pack_returns_demo_script_and_gates(self): client = TestClient(app) response = client.get("/api/v1/system/pilot-launch-pack") self.assertEqual(response.status_code, 200) payload = response.json()["data"] self.assertEqual(payload["version"], "pilot-launch-pack-2026-05-12") self.assertEqual(payload["status"], "ready") step_keys = {item["key"] for item in payload["script_steps"]} self.assertSetEqual( step_keys, { "kickoff_alignment", "business_contract_generation", "legal_auto_review", "review_chat_followup", "report_feedback_handoff", }, ) gate_keys = {item["key"] for item in payload["readiness_gates"]} self.assertTrue({"system_readiness", "pilot_scope_frozen", "pilot_materials_ready"}.issubset(gate_keys)) self.assertGreaterEqual(len(payload["roles"]), 5) self.assertTrue(any(item["key"] == "readiness_failed" for item in payload["blockers"])) self.assertGreaterEqual(len(payload["acceptance"]), 4) def test_readiness_warns_when_default_admin_password_is_active(self): app.dependency_overrides[get_db_session] = lambda: StubReadyDb(password="123456") client = TestClient(app) response = client.get("/api/v1/system/readiness") self.assertEqual(response.status_code, 200) payload = response.json()["data"] self.assertEqual(payload["status"], "ready") self.assertFalse(payload["pilot_assets"]["default_admin"]["password_rotated"]) self.assertGreaterEqual(payload["summary"]["warnings"], 1) self.assertTrue(any(item["key"] == "default_admin_password" and item["status"] == "warning" for item in payload["checks"])) def test_readiness_degrades_when_database_check_fails(self): app.dependency_overrides[get_db_session] = lambda: StubBrokenDb() client = TestClient(app) response = client.get("/api/v1/system/readiness") self.assertEqual(response.status_code, 200) payload = response.json()["data"] self.assertEqual(payload["status"], "degraded") self.assertEqual(payload["summary"]["failed_required"], 1) self.assertTrue(any(item["key"] == "database" and item["status"] == "failed" for item in payload["checks"])) def test_shared_ai_status_reports_local_only_mode(self): settings.shared_ai_platform_base_url = "" settings.shared_ai_platform_legal_review_enabled = False app.dependency_overrides[require_current_user] = lambda: { "id": "user-admin", "tenant_id": "tenant-demo-001", "role": "admin", } client = TestClient(app) response = client.get("/api/v1/system/shared-ai") self.assertEqual(response.status_code, 200) payload = response.json()["data"] self.assertEqual(payload["mode"], "configuration-required") self.assertFalse(payload["remote_configured"]) self.assertEqual(payload["transport"]["fallback"], "disabled") self.assertEqual(payload["knowledge_defaults"]["retrieval_mode"], "unknown") self.assertFalse(payload["capabilities"]["chat_stream"]) self.assertFalse(payload["capabilities"]["legal_review_domain_pack"]) self.assertFalse(payload["capabilities"]["local_legal_review_fallback"]) def test_shared_ai_status_reports_remote_preferred_mode(self): settings.shared_ai_platform_base_url = "http://127.0.0.1:8011" settings.shared_ai_platform_api_key = "test-key" settings.shared_ai_platform_api_prefix = "/api/v0.1" settings.shared_ai_platform_app_id = "ai-contract-platform" settings.shared_ai_platform_timeout_seconds = 45.0 settings.shared_ai_platform_legal_review_enabled = True settings.shared_ai_platform_legal_review_fallback_to_local = True app.dependency_overrides[require_current_user] = lambda: { "id": "user-member", "tenant_id": "tenant-demo-001", "role": "member", } client = TestClient(app) response = client.get("/api/v1/system/shared-ai") self.assertEqual(response.status_code, 200) payload = response.json()["data"] self.assertEqual(payload["mode"], "remote-preferred") self.assertTrue(payload["remote_configured"]) self.assertEqual(payload["base_url"], "http://127.0.0.1:8011") self.assertEqual(payload["transport"]["sdk"], "local-shared-ai-sdk") self.assertEqual(payload["transport"]["fallback"], "disabled") self.assertEqual(payload["knowledge_defaults"]["retrieval_mode"], "hybrid") self.assertTrue(payload["capabilities"]["legal_review_domain_pack"]) self.assertTrue(payload["capabilities"]["local_legal_review_fallback"]) def test_shared_ai_status_requires_api_key_for_remote_mode(self): settings.shared_ai_platform_base_url = "http://127.0.0.1:8011" settings.shared_ai_platform_api_key = "" settings.shared_ai_platform_legal_review_enabled = True app.dependency_overrides[require_current_user] = lambda: { "id": "user-member", "tenant_id": "tenant-demo-001", "role": "member", } client = TestClient(app) response = client.get("/api/v1/system/shared-ai") self.assertEqual(response.status_code, 200) payload = response.json()["data"] self.assertEqual(payload["mode"], "configuration-required") self.assertFalse(payload["remote_configured"]) self.assertFalse(payload["capabilities"]["chat_stream"]) self.assertFalse(payload["capabilities"]["legal_review_domain_pack"]) self.assertFalse(payload["capabilities"]["local_legal_review_fallback"]) self.assertEqual(payload["transport"]["sdk"], "disabled") self.assertEqual(payload["knowledge_defaults"]["retrieval_mode"], "unknown") self.assertEqual(payload["base_url"], "http://127.0.0.1:8011") def test_admin_snapshot_aggregates_management_views(self): app.dependency_overrides[require_current_user] = lambda: { "id": "user-admin", "tenant_id": "tenant-demo-001", "role": "admin", } app.dependency_overrides[get_shared_ai_platform_client] = lambda: StubSharedAIPlatformClient() app.dependency_overrides[get_template_service] = lambda: StubTemplateService() app.dependency_overrides[get_audit_log_service] = lambda: StubAuditLogService() app.dependency_overrides[get_governance_service] = lambda: StubGovernanceService() client = TestClient(app) response = client.get("/api/v1/system/admin-snapshot") self.assertEqual(response.status_code, 200) payload = response.json()["data"] self.assertEqual(payload["knowledge_bases"]["items"][0]["retrieval_mode"], "hybrid") self.assertEqual(payload["templates"]["enabled_count"], 1) self.assertEqual(payload["audit_logs"]["latest_action"], "upload_knowledge_document") self.assertEqual(payload["governance"]["pending_assignments"], 2) self.assertEqual(payload["governance"]["pending_assignment_items"][0]["asset_name"], "Procurement Template") def test_admin_can_seed_pilot_demo_assets(self): audit_log_service = StubAuditLogService() app.dependency_overrides[require_current_user] = lambda: { "id": "user-admin", "tenant_id": "tenant-demo-001", "role": "admin", } app.dependency_overrides[get_db_session] = lambda: StubReadyDb() app.dependency_overrides[get_shared_ai_platform_client] = lambda: StubSharedAIPlatformClient() app.dependency_overrides[get_template_service] = lambda: StubTemplateService() app.dependency_overrides[get_audit_log_service] = lambda: audit_log_service client = TestClient(app) response = client.post("/api/v1/system/pilot-demo-assets/seed") self.assertEqual(response.status_code, 200) payload = response.json()["data"] self.assertEqual(payload["templates"]["total"], 3) self.assertEqual(payload["knowledge"]["status"], "ready") self.assertEqual(payload["knowledge"]["knowledge_base"]["name"], "AI合同试点演示知识库") self.assertEqual(payload["knowledge"]["document"]["status"], "indexed") self.assertIn("legal", payload["demo_accounts"]["roles"]) self.assertEqual(audit_log_service.events[0]["action"], "seed_pilot_demo_assets") def test_non_admin_cannot_seed_pilot_demo_assets(self): app.dependency_overrides[require_current_user] = lambda: { "id": "user-legal", "tenant_id": "tenant-demo-001", "role": "legal", } app.dependency_overrides[get_db_session] = lambda: StubReadyDb() app.dependency_overrides[get_shared_ai_platform_client] = lambda: StubSharedAIPlatformClient() app.dependency_overrides[get_template_service] = lambda: StubTemplateService() app.dependency_overrides[get_audit_log_service] = lambda: StubAuditLogService() client = TestClient(app) response = client.post("/api/v1/system/pilot-demo-assets/seed") self.assertEqual(response.status_code, 403) if __name__ == "__main__": unittest.main()