"""Unit and Integration Tests for Cross-Agent OCR Validation and Human-in-the-Loop Quarantine.""" import unittest from unittest.mock import AsyncMock, patch from fastapi.testclient import TestClient from app.modules.auth.presentation.auth import get_current_user from app.core.postgres.erp_client import ErpDatabaseClient from app.main import create_app from app.modules.core.application.document_processing_service import DocumentProcessingService class TestDocumentCrossAgentAudit(unittest.TestCase): def _cleanup_test_documents(self): for doc_id in list(self.created_doc_ids): try: self.erp_client.delete_document(doc_id, delete_all_versions=True) except Exception: pass self.created_doc_ids.clear() # Hard fallback sweep: purge any test documents created by this test suite try: with self.erp_client.get_connection() as conn, conn.cursor() as cur: cur.execute(""" DELETE FROM erp_documents WHERE document_code IN ('TEST-BB-QUARANTINE-001', 'TEST-CROSS-AUDIT-001', '99/2026/CV-TEST') OR document_title LIKE '%kỹ sư Quỳnh%' OR document_title = 'Biên bản nghi ngờ sai số' OR document_title = 'File scan CV 99'; """) conn.commit() except Exception: pass def setUp(self): self.erp_client = ErpDatabaseClient() self.doc_service = DocumentProcessingService(erp_client=self.erp_client) self.created_doc_ids: list[str] = [] self._cleanup_test_documents() def tearDown(self): self._cleanup_test_documents() def test_validate_deterministic_rules(self): """Kiểm tra các quy tắc logic cứng (MST, ngày tháng, độ dài văn bản).""" # Test case 1: Dữ liệu chuẩn valid_extracted = { "partner_tax_code": "0202111150", "issue_date": "2026-03-15", "document_title": "Biên bản nghiệm thu công việc xây dựng cọc khoan nhồi", } flags = self.doc_service.validate_deterministic_rules( valid_extracted, "CỘNG HÒA XÃ HỘI CHỦ NGHĨA VIỆT NAM. Biên bản nghiệm thu công việc xây dựng công trình.", ) # Không có lỗi nghiêm trọng error_flags = [f for f in flags if f.get("severity") == "error"] self.assertEqual(len(error_flags), 0) # Test case 2: MST sai quy cách & năm bất thường invalid_extracted = { "partner_tax_code": "02021ABC99", # Sai MST (chứa chữ cái) "issue_date": "2099-01-01", # Năm 2099 bất thường "document_title": "CV", # Trích yếu quá ngắn } flags_invalid = self.doc_service.validate_deterministic_rules( invalid_extracted, "Scan mờ", # < 25 ký tự ) fields_flagged = [f["field"] for f in flags_invalid] self.assertIn("partner_tax_code", fields_flagged) self.assertIn("issue_date", fields_flagged) self.assertIn("document_title", fields_flagged) self.assertIn("raw_text", fields_flagged) @patch("app.modules.agents.application.llm_client.LLMClient.chat", new_callable=AsyncMock) def test_cross_agent_audit_and_quarantine_pipeline(self, mock_llm_chat): """Kiểm tra quy trình chạy pipeline với Agent Thảo + Thủy và phân luồng cách ly.""" import asyncio # Giả lập phản hồi của Agent Thảo thao_response = """{ "document_code": "99/2026/CV-TEST", "document_title": "Công văn cảnh báo tiến độ và khối lượng cống Kiến Minh", "document_type": "CV", "signature_status": "fully_executed", "partner_tax_code": "0202111150", "issue_date": "2026-04-10", "issuer_name": "Công ty TNHH Xây Dựng Định Sơn", "signer_name": "Đinh Văn Sơn", "summary_content": "Thông báo tiến độ thi công cống Bến Kem và trạm bơm Kiến Minh." }""" # Giả lập phản hồi của Agent Thủy (Kiểm toán) thuy_response = """{ "confidence_score": 92, "cross_audit_summary": "Dữ liệu bóc tách trùng khớp 100% với nội dung văn bản gốc, chữ ký giám đốc rõ ràng.", "flags": [] }""" mock_llm_chat.side_effect = [thao_response, thuy_response] # 1. Tạo văn bản ban đầu initial_doc = self.erp_client.create_document( { "document_code": "TEST-CROSS-AUDIT-001", "document_title": "File scan CV 99", "document_type": "CV", "category": "incoming", } ) doc_id = initial_doc["id"] self.created_doc_ids.append(doc_id) # 2. Chạy pipeline asyncio.run( self.doc_service.run_pipeline( doc_id=doc_id, file_path_str="", filename="cong_van_99.pdf", file_format="pdf", ) ) # 3. Lấy lại văn bản sau khi qua 2 Agent updated_doc = self.erp_client.get_document(doc_id) self.assertIsNotNone(updated_doc) self.assertEqual(updated_doc["document_code"], "99/2026/CV-TEST") self.assertGreaterEqual(updated_doc["ocr_confidence_score"], 90) self.assertEqual(updated_doc["quarantine_status"], "clear") self.assertIn("Dữ liệu bóc tách trùng khớp", updated_doc["cross_audit_summary"]) def test_human_approval_quarantine_release(self): """Kiểm tra luồng con người thẩm duyệt giải tỏa cách ly và ghi nhận audit trail.""" app = create_app() app.dependency_overrides[get_current_user] = lambda: { "user_id": "usr-01", "email": "engineer.quynh@dscons.vn", "role": "pm", } client = TestClient(app) # 1. Tạo văn bản bị cách ly doc = self.erp_client.create_document( { "document_code": "TEST-BB-QUARANTINE-001", "document_title": "Biên bản nghi ngờ sai số", "document_type": "BB", "quarantine_status": "quarantined", "status": "quarantined", } ) doc_id = doc["id"] self.created_doc_ids.append(doc_id) # 2. Kiểm tra endpoint audit-details audit_res = client.get(f"/v1/erp/documents/{doc_id}/audit-details") self.assertEqual(audit_res.status_code, 200) audit_data = audit_res.json() self.assertEqual(audit_data["quarantine_status"], "quarantined") # 3. Kỹ sư Quỳnh bấm Duyệt (Approve) approve_res = client.post( f"/v1/erp/documents/{doc_id}/approve", json={ "document_title": "Biên bản đã được kỹ sư Quỳnh chỉnh sửa và chuẩn hóa" }, ) self.assertEqual(approve_res.status_code, 200) approved_doc = approve_res.json() self.assertEqual(approved_doc["status"], "valid") self.assertEqual(approved_doc["quarantine_status"], "approved") self.assertEqual( approved_doc["verified_by_user_id"], "engineer.quynh@dscons.vn" ) self.assertIsNotNone(approved_doc["verified_at"]) if __name__ == "__main__": unittest.main()