import unittest from fastapi import HTTPException from fastapi.testclient import TestClient from app.api.contracts import get_audit_log_service, get_contract_generation_service, require_current_user from app.main import app class StubAuditLogService: def __init__(self): self.operations = [] def record_event(self, **kwargs): self.operations.append(kwargs) return kwargs class StubContractGenerationService: def _auto_review_summary(self, review_id="review-auto-123", status="completed"): return { "review_id": review_id, "review_title": "采购合同模板", "status": status, "review_as": "party_a", "contract_type": "procurement", "risk_level": "high", "current_stage": "report", "current_stage_label": "报告已生成", "status_detail": "文本提取、规则审查与报告阶段已完成，可继续查看详情和导出报告。", "failure_reason": None, "next_action": "查看风险详情，必要时导出报告并继续进入审查助手追问。", "can_retry": False, "review_opinion": "检测到高风险条款", "recommendation": "manual_review", "source_file_id": "file-generated-123", "source_file_name": "contract.pdf", "source_text_length": 12, "processing_time_ms": 1200, "detail_url": f"/reviews/{review_id}", "report_download_url": f"/api/v1/reviews/{review_id}/report/download", "report_file_name": f"review-report-{review_id}.pdf", "chat_url": f"/chat?review={review_id}", } def list_generated_contracts(self, current_user, page, page_size, contract_type=None, keyword=None): records = [ { "id": "contract-123", "template_id": "tpl-123", "template_name": "采购合同模板", "contract_type": "procurement", "generated_content": "采购方与供应商A签订合同", "variable_values": {"supplier_name": "供应商A"}, "generated_at": "2026-04-17T00:00:00+00:00", "export_format": "pdf", "file_name": "contract.pdf", "file_size": 128, "preview_url": "/api/v1/contracts/contract-123/preview", "download_url": "/api/v1/contracts/contract-123/download", "metadata": { "generated_by": "user-admin", "generated_by_name": "管理员", "tenant_id": "tenant-demo-001", "original_template": "采购合同模板", "export_options": {"watermark": True}, "variables_used": 1, }, "tenant_id": "tenant-demo-001", }, { "id": "contract-456", "template_id": "tpl-456", "template_name": "劳动合同模板", "contract_type": "employment", "generated_content": "甲方与乙方签订劳动合同", "variable_values": {"employee_name": "张三"}, "generated_at": "2026-04-16T00:00:00+00:00", "export_format": "docx", "file_name": "employment.docx", "file_size": 256, "preview_url": "/api/v1/contracts/contract-456/preview", "download_url": "/api/v1/contracts/contract-456/download", "metadata": { "generated_by": "user-admin", "generated_by_name": "法务", "tenant_id": "tenant-demo-001", "original_template": "劳动合同模板", "export_options": {}, "variables_used": 1, }, "tenant_id": "tenant-demo-001", }, { "id": "contract-other-tenant", "template_id": "tpl-other", "template_name": "其他租户模板", "contract_type": "procurement", "generated_content": "其他租户合同", "variable_values": {"supplier_name": "外部供应商"}, "generated_at": "2026-04-15T00:00:00+00:00", "export_format": "pdf", "file_name": "other.pdf", "file_size": 512, "preview_url": "/api/v1/contracts/contract-other-tenant/preview", "download_url": "/api/v1/contracts/contract-other-tenant/download", "metadata": { "generated_by": "user-other", "generated_by_name": "外部租户", "tenant_id": "tenant-other-001", "original_template": "其他租户模板", "export_options": {"seal": True}, "variables_used": 1, }, "tenant_id": "tenant-other-001", }, ] for record in records: record["auto_review"] = self._auto_review_summary(f"review-{record['id']}") tenant_records = [record for record in records if record["tenant_id"] == current_user["tenant_id"]] if contract_type: tenant_records = [record for record in tenant_records if record["contract_type"] == contract_type] if keyword: normalized_keyword = keyword.strip().lower() tenant_records = [ record for record in tenant_records if normalized_keyword in record["template_name"].lower() or normalized_keyword in record["file_name"].lower() or normalized_keyword in record["metadata"]["generated_by_name"].lower() ] start = (page - 1) * page_size items = tenant_records[start : start + page_size] return { "items": [{key: value for key, value in item.items() if key != "tenant_id"} for item in items], "total": len(tenant_records), "page": page, "page_size": page_size, } def list_available_templates(self, current_user, contract_type, template_category=None, industry_category=None): return [ { "id": "tpl-123", "name": "采购合同模板", "contract_type": contract_type or "procurement", "description": "标准采购合同", "variables": [ { "name": "supplier_name", "label": "供应商名称", "required": True, "default_value": "", "description": "填写供应商名称", } ], "preview_excerpt": "甲方向乙方采购货物。", } ] def preview_contract(self, current_user, template_id, variable_values): if template_id == "tpl-missing": raise HTTPException(status_code=404, detail="模板不存在或无权访问") return { "preview_content": f"采购方与{variable_values.get('supplier_name', '')}签订合同", "template_name": "采购合同模板", "variable_summary": [ { "name": "supplier_name", "label": "供应商名称", "required": True, "provided": "supplier_name" in variable_values, "value_preview": variable_values.get("supplier_name", ""), "is_empty": not bool(variable_values.get("supplier_name", "").strip()), } ], } def get_generated_contract(self, current_user, contract_id): records = [ item for item in self.list_generated_contracts(current_user, page=1, page_size=50)["items"] if item["id"] == contract_id ] if not records: raise HTTPException(status_code=404, detail="合同不存在或无权访问") return records[0] def generate_contract(self, current_user, template_id, variable_values, export_format, options): if template_id == "tpl-missing": raise HTTPException(status_code=404, detail="模板不存在或无权访问") return { "id": "contract-123", "template_id": template_id, "template_name": "采购合同模板", "contract_type": "procurement", "generated_content": f"采购方与{variable_values.get('supplier_name', '')}签订合同", "variable_values": variable_values, "generated_at": "2026-04-17T00:00:00+00:00", "export_format": export_format.value, "file_name": f"contract.{export_format.value}", "file_size": 128, "file_content_base64": "ZmlsZQ==", "preview_url": "/api/v1/contracts/contract-123/preview", "download_url": "/api/v1/contracts/contract-123/download", "auto_review": self._auto_review_summary("review-auto-123"), "metadata": { "generated_by": current_user["id"], "generated_by_name": current_user.get("display_name", "管理员"), "tenant_id": current_user["tenant_id"], "original_template": "采购合同模板", "export_options": options, "variables_used": len(variable_values), }, } def get_generated_contract_preview(self, current_user, contract_id): if contract_id == "contract-missing": raise HTTPException(status_code=404, detail="合同不存在或无权访问") return { "id": contract_id, "template_name": "采购合同模板", "contract_type": "procurement", "preview_content": "采购方与供应商A签订合同", "export_format": "pdf", "file_name": "contract.pdf", "generated_at": "2026-04-17T00:00:00+00:00", "preview_url": f"/api/v1/contracts/{contract_id}/preview", "download_url": f"/api/v1/contracts/{contract_id}/download", } def get_generated_contract_download(self, current_user, contract_id): if contract_id == "contract-missing": raise HTTPException(status_code=404, detail="合同不存在或无权访问") return { "id": contract_id, "file_name": "contract.pdf", "mime_type": "application/pdf", "content": b"pdf-bytes", } class ContractsApiTestCase(unittest.TestCase): def setUp(self): self.audit_log_service = StubAuditLogService() app.dependency_overrides[get_contract_generation_service] = lambda: StubContractGenerationService() app.dependency_overrides[get_audit_log_service] = lambda: self.audit_log_service self.client = TestClient(app) def tearDown(self): app.dependency_overrides.clear() def test_user_can_list_contract_templates(self): app.dependency_overrides[require_current_user] = lambda: { "id": "user-admin", "tenant_id": "tenant-demo-001", "display_name": "管理员", "role": "admin", } response = self.client.get("/api/v1/contracts/templates", params={"contract_type": "procurement"}) self.assertEqual(response.status_code, 200) body = response.json() self.assertEqual(body["data"][0]["name"], "采购合同模板") self.assertEqual(body["data"][0]["contract_type"], "procurement") def test_user_can_list_generated_contracts(self): app.dependency_overrides[require_current_user] = lambda: { "id": "user-admin", "tenant_id": "tenant-demo-001", "display_name": "管理员", "role": "admin", } response = self.client.get("/api/v1/contracts", params={"page": 1, "page_size": 10}) self.assertEqual(response.status_code, 200) body = response.json() self.assertEqual(body["data"]["total"], 2) self.assertEqual(body["data"]["page"], 1) self.assertEqual(body["data"]["page_size"], 10) self.assertEqual(len(body["data"]["items"]), 2) self.assertEqual(body["data"]["items"][0]["id"], "contract-123") self.assertEqual(body["data"]["items"][0]["metadata"]["generated_by_name"], "管理员") self.assertIn("preview_url", body["data"]["items"][0]) self.assertIn("download_url", body["data"]["items"][0]) def test_generated_contract_list_supports_contract_type_filter(self): app.dependency_overrides[require_current_user] = lambda: { "id": "user-admin", "tenant_id": "tenant-demo-001", "display_name": "管理员", "role": "admin", } response = self.client.get("/api/v1/contracts", params={"contract_type": "employment"}) self.assertEqual(response.status_code, 200) body = response.json() self.assertEqual(body["data"]["total"], 1) self.assertEqual(len(body["data"]["items"]), 1) self.assertEqual(body["data"]["items"][0]["contract_type"], "employment") self.assertEqual(body["data"]["items"][0]["id"], "contract-456") def test_generated_contract_list_supports_keyword_filter(self): app.dependency_overrides[require_current_user] = lambda: { "id": "user-admin", "tenant_id": "tenant-demo-001", "display_name": "管理员", "role": "admin", } response = self.client.get("/api/v1/contracts", params={"keyword": "法务"}) self.assertEqual(response.status_code, 200) body = response.json() self.assertEqual(body["data"]["total"], 1) self.assertEqual(len(body["data"]["items"]), 1) self.assertEqual(body["data"]["items"][0]["id"], "contract-456") self.assertEqual(body["data"]["items"][0]["metadata"]["generated_by_name"], "法务") def test_generated_contract_list_is_tenant_isolated(self): app.dependency_overrides[require_current_user] = lambda: { "id": "user-other", "tenant_id": "tenant-other-001", "display_name": "外部租户", "role": "admin", } response = self.client.get("/api/v1/contracts") self.assertEqual(response.status_code, 200) body = response.json() self.assertEqual(body["data"]["total"], 1) self.assertEqual(len(body["data"]["items"]), 1) self.assertEqual(body["data"]["items"][0]["id"], "contract-other-tenant") self.assertEqual(body["data"]["items"][0]["file_name"], "other.pdf") def test_user_can_get_generated_contract_detail(self): app.dependency_overrides[require_current_user] = lambda: { "id": "user-admin", "tenant_id": "tenant-demo-001", "display_name": "管理员", "role": "admin", } response = self.client.get("/api/v1/contracts/contract-123") self.assertEqual(response.status_code, 200) body = response.json() self.assertEqual(body["data"]["id"], "contract-123") self.assertEqual(body["data"]["template_id"], "tpl-123") self.assertEqual(body["data"]["export_format"], "pdf") self.assertEqual(body["data"]["variable_values"]["supplier_name"], "供应商A") self.assertEqual(body["data"]["metadata"]["export_options"], {"watermark": True}) self.assertEqual(body["data"]["auto_review"]["status"], "completed") self.assertEqual(body["data"]["auto_review"]["detail_url"], "/reviews/review-contract-123") self.assertIn("generated_content", body["data"]) def test_generated_contract_detail_is_tenant_isolated(self): app.dependency_overrides[require_current_user] = lambda: { "id": "user-admin", "tenant_id": "tenant-demo-001", "display_name": "管理员", "role": "admin", } response = self.client.get("/api/v1/contracts/contract-other-tenant") self.assertEqual(response.status_code, 404) self.assertEqual(response.json()["message"], "合同不存在或无权访问") def test_user_can_generate_contract_and_write_audit_log(self): app.dependency_overrides[require_current_user] = lambda: { "id": "user-admin", "tenant_id": "tenant-demo-001", "display_name": "管理员", "role": "admin", } response = self.client.post( "/api/v1/contracts/generate", data={ "template_id": "tpl-123", "variable_values_json": '{"supplier_name":"供应商A"}', "export_format": "pdf", "options_json": '{"watermark":true}', }, ) self.assertEqual(response.status_code, 200) body = response.json() self.assertEqual(body["data"]["export_format"], "pdf") self.assertEqual(body["data"]["file_name"], "contract.pdf") self.assertEqual(body["data"]["metadata"]["export_options"], {"watermark": True}) self.assertEqual(body["data"]["auto_review"]["status"], "completed") self.assertEqual(body["data"]["auto_review"]["report_download_url"], "/api/v1/reviews/review-auto-123/report/download") self.assertEqual(self.audit_log_service.operations[0]["action"], "contract_generation") self.assertEqual(self.audit_log_service.operations[0]["resource_id"], "contract-123") def test_user_can_get_generated_contract_preview(self): app.dependency_overrides[require_current_user] = lambda: { "id": "user-admin", "tenant_id": "tenant-demo-001", "display_name": "管理员", "role": "admin", } response = self.client.get("/api/v1/contracts/contract-123/preview") self.assertEqual(response.status_code, 200) body = response.json() self.assertEqual(body["data"]["id"], "contract-123") self.assertEqual(body["data"]["file_name"], "contract.pdf") self.assertIn("供应商A", body["data"]["preview_content"]) def test_user_can_download_generated_contract_and_write_audit_log(self): app.dependency_overrides[require_current_user] = lambda: { "id": "user-admin", "tenant_id": "tenant-demo-001", "display_name": "管理员", "role": "admin", } response = self.client.get("/api/v1/contracts/contract-123/download") self.assertEqual(response.status_code, 200) self.assertEqual(response.content, b"pdf-bytes") self.assertEqual(response.headers["content-type"], "application/pdf") self.assertIn("attachment;", response.headers["content-disposition"]) self.assertIn("contract.pdf", response.headers["content-disposition"]) self.assertEqual(self.audit_log_service.operations[0]["action"], "download_contract") self.assertEqual( self.audit_log_service.operations[0]["detail_json"], { "file_name": "contract.pdf", "mime_type": "application/pdf", }, ) self.assertEqual(self.audit_log_service.operations[0]["resource_id"], "contract-123") def test_preview_rejects_invalid_json(self): app.dependency_overrides[require_current_user] = lambda: { "id": "user-admin", "tenant_id": "tenant-demo-001", "display_name": "管理员", "role": "admin", } response = self.client.post( "/api/v1/contracts/preview", data={ "template_id": "tpl-123", "variable_values_json": "{bad json}", }, ) self.assertEqual(response.status_code, 400) self.assertEqual(response.json()["message"], "JSON解析错误") def test_generate_rejects_invalid_export_format(self): app.dependency_overrides[require_current_user] = lambda: { "id": "user-admin", "tenant_id": "tenant-demo-001", "display_name": "管理员", "role": "admin", } response = self.client.post( "/api/v1/contracts/generate", data={ "template_id": "tpl-123", "variable_values_json": '{"supplier_name":"供应商A"}', "export_format": "xlsx", }, ) self.assertEqual(response.status_code, 400) self.assertIn("不支持的导出格式", response.json()["message"]) def test_generate_surfaces_service_404(self): app.dependency_overrides[require_current_user] = lambda: { "id": "user-admin", "tenant_id": "tenant-demo-001", "display_name": "管理员", "role": "admin", } response = self.client.post( "/api/v1/contracts/generate", data={ "template_id": "tpl-missing", "variable_values_json": '{"supplier_name":"供应商A"}', "export_format": "docx", }, ) self.assertEqual(response.status_code, 404) self.assertEqual(response.json()["message"], "模板不存在或无权访问") def test_generated_contract_preview_surfaces_service_404(self): app.dependency_overrides[require_current_user] = lambda: { "id": "user-admin", "tenant_id": "tenant-demo-001", "display_name": "管理员", "role": "admin", } response = self.client.get("/api/v1/contracts/contract-missing/preview") self.assertEqual(response.status_code, 404) self.assertEqual(response.json()["message"], "合同不存在或无权访问") def test_generated_contract_download_surfaces_service_404(self): app.dependency_overrides[require_current_user] = lambda: { "id": "user-admin", "tenant_id": "tenant-demo-001", "display_name": "管理员", "role": "admin", } response = self.client.get("/api/v1/contracts/contract-missing/download") self.assertEqual(response.status_code, 404) self.assertEqual(response.json()["message"], "合同不存在或无权访问") class ContractGenerationServiceVariableParsingTestCase(unittest.TestCase): def setUp(self): from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from app.db import Base from app.models.contract_template import ContractTemplate from app.services.contract_generation_service import ContractGenerationService self.engine = create_engine("sqlite:///:memory:") Base.metadata.create_all(bind=self.engine) self.SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=self.engine) self.db = self.SessionLocal() self.service = ContractGenerationService(self.db) self.service.file_service = type( "StubFileService", (), { "create_file_record": lambda _self, **kwargs: { "id": "file-generated-123", "file_name": kwargs["file_name"], "mime_type": kwargs["mime_type"], "file_size": len(kwargs["content"]), "file_type": kwargs["file_type"], } }, )() class StubReviewService: def __init__(self): self.create_calls = [] def create_review_task(self, **kwargs): self.create_calls.append(kwargs) return { "id": "review-auto-generated", "status": "completed", "review_as": kwargs["review_as"], "contract_type": kwargs["contract_type"], } def get_review_detail(self, current_user, review_id): return { "id": review_id, "title": "contract.docx", "contract_type": "procurement", "review_as": "party_a", "status": "completed", "source_type": "file", "source_text_length": 12, "source_text_preview": "采购方与供应商A签订合同", "processing_time_ms": 1200, "created_at": "2026-04-17T00:00:00+00:00", "completed_at": "2026-04-17T00:00:01+00:00", "error_message": None, "current_stage": "report", "current_stage_label": "报告已生成", "status_detail": "文本提取、规则审查与报告阶段已完成，可继续查看详情和导出报告。", "failure_reason": None, "next_action": "查看风险详情，必要时导出报告并继续进入审查助手追问。", "can_retry": False, "workflow_state": { "review_id": review_id, "risk_level": "high", "status": "completed", "current_stage": "report", }, "ai_trace": { "review_id": review_id, "status": "completed", "provider": "stub", }, "source_document": { "document_id": "file-generated-123", "file_name": "contract.docx", "source_type": "file", "parse_status": "completed", "extraction_mode": "template_render", "ocr_gate_status": "PASS", "text_length": 12, "preview": "采购方与供应商A签订合同", }, } def get_review_result(self, current_user, review_id): return { "basic_info": { "contract_type": "procurement", "party_a": "甲方", "party_b": "乙方", "amount": "待识别", "term": "待识别", }, "risk_summary": {"high": 1, "medium": 0, "low": 0, "overall": "high"}, "risk_items": [{"title": "付款条件建议明确", "level": "high", "description": "desc"}], "review_opinion": "检测到高风险条款", "recommendation": "manual_review", } def get_review_report(self, current_user, review_id): return { "file_name": f"review-report-{review_id}.pdf", "download_url": f"/api/v1/reviews/{review_id}/report/download", } self.review_service = StubReviewService() self.service.review_service = self.review_service self.ContractTemplate = ContractTemplate def tearDown(self): self.db.close() self.engine.dispose() def test_generate_contract_accepts_json_column_variable_definitions(self): template = self.ContractTemplate( id="tpl-real-json", tenant_id="tenant-demo-001", name="采购合同模板", contract_type="procurement", description="验证 JSON 列变量", content="采购方与{{supplier_name}}签订合同", variables_json=[ { "name": "supplier_name", "label": "供应商名称", "required": True, "default_value": "", "description": "填写供应商名称", } ], enabled=True, created_by="seed", updated_by="seed", ) self.db.add(template) self.db.commit() result = self.service.generate_contract( current_user={ "id": "user-admin", "tenant_id": "tenant-demo-001", "display_name": "管理员", }, template_id="tpl-real-json", variable_values={"supplier_name": "供应商A"}, ) self.assertEqual(result["template_id"], "tpl-real-json") self.assertEqual(result["variable_values"]["supplier_name"], "供应商A") self.assertEqual(self.review_service.create_calls[0]["source_file_id"], "file-generated-123") self.assertEqual(result["auto_review"]["status"], "completed") self.assertEqual(result["auto_review"]["chat_url"].startswith("/chat?review="), True) def test_get_generated_contract_download_allows_owner_access_to_file_payload(self): generated_contract = type( "GeneratedContractRecord", (), { "id": "contract-owner-001", "tenant_id": "tenant-demo-001", "file_id": "file-owner-001", "file_name": "contract.txt", "mime_type": "text/plain", "file_content_base64": "", }, )() captured = {} self.service.generated_contract_repository = type( "StubGeneratedContractRepository", (), {"get_by_id": lambda _self, contract_id, tenant_id: generated_contract}, )() self.service.file_service = type( "StubFileService", (), { "get_download_payload": lambda _self, current_user, file_id, allow_owner_access=False: captured.setdefault( "payload", { "current_user": current_user, "file_id": file_id, "allow_owner_access": allow_owner_access, "file_name": "contract.txt", "mime_type": "text/plain", "content": b"contract-bytes", }, ) }, )() result = self.service.get_generated_contract_download( current_user={"id": "user-member", "tenant_id": "tenant-demo-001", "role": "member"}, contract_id="contract-owner-001", ) self.assertEqual(result["file_name"], "contract.txt") self.assertEqual(captured["payload"]["file_id"], "file-owner-001") self.assertTrue(captured["payload"]["allow_owner_access"]) if __name__ == "__main__": unittest.main()