import asyncio import unittest import uuid from unittest.mock import AsyncMock, patch from fastapi.testclient import TestClient from app.core.postgres.erp_client import ErpDatabaseClient from app.main import app from app.modules.auth.application.auth_service import AuthService from app.modules.core.application.document_processing_service import DocumentProcessingService class TestProjectAutoCreationFromDocument(unittest.TestCase): def setUp(self): self.client = TestClient(app) self.erp = ErpDatabaseClient() self.created_doc_ids = [] self.created_project_codes = [] # Superadmin token for authorized endpoints auth = AuthService() self.admin_token = auth.create_jwt_token( { "id": "admin-id-test", "email": "admin@dscons.vn", "role": "superadmin", "full_name": "Tổng Giám Đốc", "company_id": "0753d827-90e1-4894-9c16-266ca572c5eb", } ) self.headers = {"Authorization": f"Bearer {self.admin_token}"} def tearDown(self): # Cleanup created test documents & projects with self.erp.get_connection() as conn: with conn.cursor() as cur: if self.created_doc_ids: cur.execute( "DELETE FROM erp_documents WHERE id = ANY(%s);", (self.created_doc_ids,), ) if self.created_project_codes: cur.execute( "DELETE FROM projects WHERE project_code = ANY(%s);", (self.created_project_codes,), ) conn.commit() def test_create_and_get_project_db_methods(self): """Kiểm tra tạo mới và truy vấn dự án bằng erp_client.""" unique_code = f"DA-TEST-{uuid.uuid4().hex[:6].upper()}" self.created_project_codes.append(unique_code) payload = { "project_code": unique_code, "project_name": "Dự án Nạo vét sông và xây kè test", "client_name": "Ban Quản Lý Dự Án Test", "location": "Kiến Thụy, Hải Phòng", "project_type": "Thủy lợi", "status": "bidding", "priority": "high", "budget_amount": 150000000, "contract_duration_days": 180, "contract_type": "Trọn gói", "notes": "Kiểm thử tự động", } proj = self.erp.create_project(payload) self.assertIsNotNone(proj) self.assertEqual(proj["project_code"], unique_code) self.assertEqual(proj["status"], "bidding") # Query back fetched = self.erp.get_project(unique_code) self.assertIsNotNone(fetched) self.assertEqual(fetched["project_name"], "Dự án Nạo vét sông và xây kè test") def test_create_project_from_document_direct(self): """Kiểm tra khởi tạo dự án tự động từ văn bản TBMT và liên kết project_id.""" doc_code = f"IB{uuid.uuid4().hex[:8].upper()}" doc_payload = { "document_code": doc_code, "document_title": "Thông báo mời thầu - Gói thầu số 05: Xây dựng cống hộp kênh test", "document_group": "legal", "document_type": "TB", "issuer_name": "UBND Xã Test", "partner_name": "UBND Xã Test", "summary_content": "Thông báo mời thầu thi công xây lắp trọn gói 270 ngày.", "status": "pending_review", } doc = self.erp.create_document(doc_payload) doc_id = str(doc["id"]) self.created_doc_ids.append(doc_id) # Execute project creation from document res = self.erp.create_project_from_document(doc_id) self.assertTrue(res["created_new"]) self.assertIsNotNone(res["project_id"]) created_proj = res["project"] self.created_project_codes.append(created_proj["project_code"]) self.assertIn("cống hộp", created_proj["project_name"].lower()) self.assertEqual(created_proj["status"], "bidding") # Verify document was updated with project_id updated_doc = self.erp.get_document(doc_id) self.assertEqual(str(updated_doc["project_id"]), str(created_proj["id"])) self.assertEqual(updated_doc["project_stage"], "stage_01_initiation") def test_api_initiate_project_from_document(self): """Kiểm thử API POST /v1/erp/documents/{doc_id}/initiate-project.""" doc_code = f"TB-{uuid.uuid4().hex[:6].upper()}" doc = self.erp.create_document( { "document_code": doc_code, "document_title": "Thông báo mời thầu Dự án Đường giao thông nông thôn xã Test", "document_group": "legal", "document_type": "TB", "status": "pending_review", } ) doc_id = str(doc["id"]) self.created_doc_ids.append(doc_id) custom_payload = { "project_name": "Dự án Đường giao thông nông thôn xã Test", "project_code": f"DA-TEST-ROAD-{uuid.uuid4().hex[:4].upper()}", "project_type": "Giao thông", "client_name": "UBND Huyện Test", "location": "Hải Phòng", "contract_duration_days": 120, "status": "bidding", "notes": "Hạn nộp hồ sơ e-HSDT 15/09/2026", } self.created_project_codes.append(custom_payload["project_code"]) response = self.client.post( f"/v1/erp/documents/{doc_id}/initiate-project", json=custom_payload, headers=self.headers, ) self.assertEqual(response.status_code, 200) data = response.json() self.assertTrue(data["created_new"]) self.assertEqual( data["project"]["project_code"], custom_payload["project_code"] ) # Check document is linked doc_check = self.erp.get_document(doc_id) self.assertEqual(str(doc_check["project_id"]), str(data["project"]["id"])) def test_api_list_projects_with_status_filter(self): """Kiểm thử API GET /v1/erp/projects kèm bộ lọc trạng thái.""" unique_code = f"DA-BID-{uuid.uuid4().hex[:6].upper()}" self.created_project_codes.append(unique_code) self.erp.create_project( { "project_code": unique_code, "project_name": "Dự án Đang Bidding Test", "status": "bidding", "project_type": "Thủy lợi", } ) res_bidding = self.client.get( "/v1/erp/projects?status=bidding", headers=self.headers ) self.assertEqual(res_bidding.status_code, 200) items = res_bidding.json() self.assertTrue(any(p["project_code"] == unique_code for p in items)) def test_run_pipeline_auto_initiates_project_for_tender_doc(self): """Kiểm tra Pipeline AI tự động khởi tạo dự án khi gặp Thông Báo Mời Thầu.""" doc = self.erp.create_document( { "document_code": f"IB{uuid.uuid4().hex[:6].upper()}", "document_title": "Văn bản chờ bóc tách", "status": "processing", } ) doc_id = str(doc["id"]) self.created_doc_ids.append(doc_id) svc = DocumentProcessingService() # Mock analyze_with_ai to return TBMT result mock_ai_result = { "document_code": f"IB{uuid.uuid4().hex[:6].upper()}", "document_title": "Thông báo mời thầu - Gói thầu số 01: Thi công cống ngăn triều", "document_group": "legal", "document_type": "TB", "signature_status": "fully_executed", "document_direction": "incoming", "issuer_name": "Công ty Thủy Lợi Test", "summary_content": "Thông báo mời thầu qua mạng gói thầu xây lắp cống ngăn triều.", "issue_date": "2026-08-24", } with ( patch.object( svc, "extract_text", return_value="Nội dung thông báo mời thầu..." ), patch.object( svc, "analyze_with_ai", new=AsyncMock(return_value=mock_ai_result) ), patch.object( svc, "run_cross_agent_audit", new=AsyncMock( return_value={ "confidence_score": 90, "flags": [], "cross_audit_summary": "Khớp hoàn toàn", } ), ), ): asyncio.run( svc.run_pipeline( doc_id=doc_id, file_path_str="dummy.pdf", filename="TBMT_Cong_Ngan_Trieu.pdf", file_format="pdf", ) ) updated_doc = self.erp.get_document(doc_id) self.assertIsNotNone(updated_doc["project_id"]) # Track created project code for cleanup linked_proj = self.erp.get_project(str(updated_doc["project_id"])) if linked_proj: self.created_project_codes.append(linked_proj["project_code"]) self.assertEqual(linked_proj["status"], "bidding") if __name__ == "__main__": unittest.main()