"""Unit and integration tests for document upload and AI classification pipeline.""" from __future__ import annotations import io import unittest import zipfile from pathlib import Path from unittest.mock import MagicMock, patch from fastapi.testclient import TestClient from app.modules.auth.presentation.auth import get_current_user from app.main import create_app from app.modules.core.application.document_processing_service import DocumentProcessingService class TestDocumentUploadAi(unittest.TestCase): def setUp(self) -> None: self.temp_files: list[Path] = [] def tearDown(self) -> None: for p in self.temp_files: try: if p.exists(): p.unlink() except Exception: pass def test_save_file(self) -> None: service = DocumentProcessingService() file_content = b"Mock document content" filename = "test_doc.txt" file_url, file_path_str, file_size, file_format, file_hash = service.save_file( file_content, filename ) file_path = Path(file_path_str) self.temp_files.append(file_path) self.assertTrue(file_path.exists()) self.assertEqual(file_size, len(file_content)) self.assertEqual(file_format, "txt") self.assertTrue(file_url.startswith("/v1/erp/documents/vault/")) self.assertEqual(file_path.read_bytes(), file_content) self.assertIsNotNone(file_hash) def test_extract_text_docx(self) -> None: service = DocumentProcessingService() docx_buffer = io.BytesIO() # Build a valid minified docx zip package structure with zipfile.ZipFile(docx_buffer, "w") as z: xml_content = ( '\n' '\n' " \n" " \n" " \n" " Biên bản nghiệm thu công trình số 456\n" " \n" " \n" " \n" "" ) z.writestr("word/document.xml", xml_content) docx_data = docx_buffer.getvalue() file_url, file_path_str, file_size, file_format, file_hash = service.save_file( docx_data, "test_document.docx" ) self.temp_files.append(Path(file_path_str)) extracted_text = service.extract_text(file_path_str, file_format) self.assertEqual(extracted_text, "Biên bản nghiệm thu công trình số 456") @patch("app.modules.agents.application.llm_client.LLMClient.chat") def test_analyze_with_ai(self, mock_chat: MagicMock) -> None: import asyncio mock_response = ( "```json\n" "{\n" ' "document_group": "acceptance",\n' ' "document_type": "BB",\n' ' "document_code": "BB-NT-456",\n' ' "document_title": "Biên bản nghiệm thu hoàn thành",\n' ' "issue_date": "2026-08-19",\n' ' "issuer_name": "DSCons Corp",\n' ' "signer_name": "Nguyễn Văn A",\n' ' "summary_content": "Nghiệm thu hoàn thành hạng mục kết cấu tầng hầm.",\n' ' "ai_risk_assessment": "Không phát hiện rủi ro.",\n' ' "ai_legal_references": ["Luật Xây dựng 2025 (Điều 135)"]\n' "}\n" "```" ) async def fake_chat(*args: Any, **kwargs: Any) -> str: return mock_response mock_chat.side_effect = fake_chat service = DocumentProcessingService() async def run_test(): result = await service.analyze_with_ai( "Mock document text content", "test.pdf" ) self.assertEqual(result["document_group"], "acceptance") self.assertEqual(result["document_type"], "BB") self.assertEqual(result["document_code"], "BB-NT-456") self.assertEqual(result["document_title"], "Biên bản nghiệm thu hoàn thành") self.assertEqual(result["signer_name"], "Nguyễn Văn A") asyncio.run(run_test()) @patch( "app.core.postgres.erp_client.ErpDatabaseClient.create_project_from_document" ) @patch("app.core.postgres.erp_client.ErpDatabaseClient.get_document") @patch( "app.modules.core.application.document_processing_service.DocumentProcessingService.run_cross_agent_audit" ) @patch( "app.modules.core.application.document_processing_service.DocumentProcessingService.analyze_with_ai" ) @patch( "app.modules.core.application.document_processing_service.DocumentProcessingService.extract_text" ) @patch("app.core.postgres.erp_client.ErpDatabaseClient.update_document") def test_run_pipeline( self, mock_update: MagicMock, mock_extract: MagicMock, mock_analyze: MagicMock, mock_cross_audit: MagicMock, mock_get_doc: MagicMock, mock_create_proj: MagicMock, ) -> None: import asyncio mock_get_doc.return_value = { "id": "doc-uuid-1", "document_code": "TEMP-123", "project_id": None, } mock_create_proj.return_value = { "created_new": True, "project": {"id": "p-1", "project_name": "DA"}, } mock_extract.return_value = "Cộng hòa Xã hội Chủ nghĩa Việt Nam. extracted sample text hợp đồng kinh tế đầy đủ." mock_cross_audit.return_value = { "confidence_score": 95, "cross_audit_summary": "Hồ sơ trích xuất hoàn chỉnh, chuẩn xác.", "flags": [], } mock_analyze.return_value = { "document_group": "contract", "document_type": "HD", "document_code": "HD-123", "document_title": "Hợp đồng thiết kế", "issue_date": "2026-08-19", "issuer_name": "DSCons Ltd", "signer_name": "Trần B", "summary_content": "Hợp đồng thiết kế kiến trúc cảnh quan.", "ai_risk_assessment": "Có rủi ro chậm thanh toán.", "ai_legal_references": ["Nghị định 206/2026/NĐ-CP"], } service = DocumentProcessingService() async def run_test(): await service.run_pipeline( "doc-uuid-1", "/tmp/fake_path.pdf", "fake_path.pdf", "pdf" ) mock_extract.assert_called_once_with("/tmp/fake_path.pdf", "pdf") mock_analyze.assert_called_once_with( mock_extract.return_value, "fake_path.pdf" ) mock_update.assert_called_once() # Verify target fields call_args = mock_update.call_args[0] self.assertEqual(call_args[0], "doc-uuid-1") payload = call_args[1] self.assertEqual(payload["document_code"], "HD-123") self.assertEqual(payload["document_title"], "Hợp đồng thiết kế") self.assertEqual(payload["document_group"], "contract") self.assertEqual(payload["document_type"], "HD") self.assertEqual(payload["status"], "pending_review") self.assertEqual(payload["verification_status"], "pending_review") asyncio.run(run_test()) @patch("app.core.postgres.erp_client.ErpDatabaseClient.check_duplicate_document") @patch("app.core.postgres.erp_client.ErpDatabaseClient.create_document") @patch("app.core.postgres.erp_client.ErpDatabaseClient.update_document") @patch("app.core.postgres.erp_client.ErpDatabaseClient.get_connection") @patch( "app.modules.core.application.document_processing_service.DocumentProcessingService.save_file" ) def test_api_upload_document( self, mock_save: MagicMock, mock_get_conn: MagicMock, mock_update_doc: MagicMock, mock_create_doc: MagicMock, mock_check_dup: MagicMock, ) -> None: # Mock database queries for company and project resolution mock_conn = MagicMock() mock_cur = MagicMock() mock_cur.fetchone.side_effect = [ {"id": "company-uuid-1"}, # for select company {"id": "project-uuid-1"}, # for select project ] mock_conn.cursor.return_value.__enter__.return_value = mock_cur mock_get_conn.return_value.__enter__.return_value = mock_conn mock_save.return_value = ( "/static/uploads/file.pdf", "/disk/file.pdf", 100, "pdf", "mock_hash_123", ) mock_check_dup.return_value = None mock_create_doc.return_value = {"id": "doc-uuid-101"} mock_update_doc.return_value = {"id": "doc-uuid-101", "status": "processing"} app = create_app() app.dependency_overrides[get_current_user] = lambda: { "id": "user-1", "email": "test@dscons.vn", "role": "superadmin", } client = TestClient(app) with patch("fastapi.BackgroundTasks.add_task") as mock_add_task: response = client.post( "/v1/erp/documents/upload", files={"file": ("report.pdf", b"pdf content bytes", "application/pdf")}, data={"project_code": "PRJ-999"}, ) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["id"], "doc-uuid-101") mock_create_doc.assert_called_once() mock_update_doc.assert_called_once() mock_add_task.assert_called_once() @patch("app.core.postgres.erp_client.ErpDatabaseClient.update_document") def test_api_approve_document(self, mock_update: MagicMock) -> None: mock_update.return_value = { "id": "doc-uuid-101", "status": "valid", "verification_status": "verified", } app = create_app() app.dependency_overrides[get_current_user] = lambda: { "id": "user-1", "email": "test@dscons.vn", "role": "superadmin", } client = TestClient(app) response = client.post( "/v1/erp/documents/doc-uuid-101/approve", json={ "document_title": "Đã phê duyệt trích xuất", "document_code": "CV-OK-101", }, ) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["status"], "valid") self.assertEqual(response.json()["verification_status"], "verified") mock_update.assert_called_once() payload = mock_update.call_args[0][1] self.assertEqual(payload["status"], "valid") self.assertEqual(payload["verification_status"], "verified") self.assertEqual(payload["document_title"], "Đã phê duyệt trích xuất") self.assertEqual(payload["document_code"], "CV-OK-101")