"""Unit and integration tests for multi-version document management, AI diffing, and audit readiness.""" from __future__ import annotations import io import json import unittest from typing import Any from unittest.mock import MagicMock, 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 TestDocumentVersioningAndAudit(unittest.TestCase): def setUp(self) -> None: self.erp_client = ErpDatabaseClient() self.created_doc_ids: list[str] = [] def tearDown(self) -> None: with self.erp_client.get_connection() as conn: with conn.cursor() as cur: for doc_id in self.created_doc_ids: try: cur.execute( "DELETE FROM erp_documents WHERE id = %s OR parent_document_id = %s OR root_document_id = %s;", (doc_id, doc_id, doc_id), ) except Exception: pass conn.commit() def test_create_and_version_document(self) -> None: # 1. Tạo document v1.0 with self.erp_client.get_connection() as conn, conn.cursor() as cur: cur.execute("SELECT id FROM companies LIMIT 1;") row = cur.fetchone() company_id = str(row["id"]) if row else None initial_payload = { "company_id": company_id, "document_code": "01/2026/CV-TEST", "document_title": "Công văn xin gia hạn tiến độ lần 1", "document_group": "dispatch", "document_type": "CV", "signature_status": "draft", "project_stage": "stage_01_initiation", "version_number": "v1.0", "summary_content": "Bản dự thảo ban đầu chưa ký.", } doc_v1 = self.erp_client.create_document(initial_payload) self.created_doc_ids.append(doc_v1["id"]) self.assertEqual(doc_v1["version_number"], "v1.0") self.assertEqual(doc_v1["version_index"], 1) self.assertTrue(doc_v1["is_active_version"]) self.assertEqual(doc_v1["signature_status"], "draft") # 2. Tạo phiên bản sửa đổi v1.1 v2_payload = { "document_title": "Công văn xin gia hạn tiến độ lần 1 (Đã sửa lỗi ngày tháng)", "signature_status": "signed_unsigned_seal", "change_summary": "Sửa lại ngày kết thúc từ 30/08 sang 15/09", "diff_analysis": { "summary_of_changes": "Điều chỉnh mốc thời gian hoàn thành" }, "file_url": "/static/uploads/test_v1_1.pdf", } doc_v1_1 = self.erp_client.create_document_version( parent_doc_id=doc_v1["id"], payload=v2_payload ) self.created_doc_ids.append(doc_v1_1["id"]) self.assertEqual(doc_v1_1["version_number"], "v1.1") self.assertEqual(doc_v1_1["version_index"], 2) self.assertTrue(doc_v1_1["is_active_version"]) self.assertEqual(doc_v1_1["parent_document_id"], doc_v1["id"]) self.assertEqual(doc_v1_1["root_document_id"], doc_v1["id"]) self.assertEqual(doc_v1_1["signature_status"], "signed_unsigned_seal") # Kiểm tra bản cũ v1.0 đã chuyển is_active_version = False old_doc = self.erp_client.get_document(doc_v1["id"]) self.assertFalse(old_doc["is_active_version"]) # 3. Tạo phiên bản đã ký đóng dấu v3.0_Final v3_payload = { "signature_status": "fully_executed", "change_summary": "Đã có chữ ký Giám đốc và đóng dấu đỏ pháp nhân", "file_url": "/static/uploads/test_v_final_signed.pdf", } doc_v_final = self.erp_client.create_document_version( parent_doc_id=doc_v1_1["id"], payload=v3_payload ) self.created_doc_ids.append(doc_v_final["id"]) self.assertEqual(doc_v_final["version_number"], "v3.0_Final") self.assertEqual(doc_v_final["signature_status"], "fully_executed") self.assertTrue(doc_v_final["is_active_version"]) # 4. Kiểm tra danh sách phiên bản list_document_versions versions = self.erp_client.list_document_versions(doc_v_final["id"]) self.assertEqual(len(versions), 3) self.assertEqual(versions[0]["version_number"], "v1.0") self.assertEqual(versions[1]["version_number"], "v1.1") self.assertEqual(versions[2]["version_number"], "v3.0_Final") def test_audit_readiness_report(self) -> None: report = self.erp_client.get_audit_readiness_report() self.assertIn("total_documents", report) self.assertIn("fully_executed_count", report) self.assertIn("missing_seal_count", report) self.assertIn("draft_count", report) self.assertIn("audit_readiness_score", report) self.assertIn("readiness_status", report) self.assertIn("audit_warning_notes", report) @patch("app.modules.agents.application.llm_client.LLMClient.chat") def test_compare_document_versions_ai(self, mock_chat: MagicMock) -> None: import asyncio mock_diff_response = json.dumps( { "summary_of_changes": "Sửa đổi điều khoản thanh toán từ 30 ngày thành 15 ngày", "financial_impact": "Rút ngắn thời gian thu hồi công nợ", "legal_impact": "Phù hợp tiến độ thi công mới", "key_differences": [ { "section": "Điều 5", "old_value": "30 ngày", "new_value": "15 ngày", "reason": "Thỏa thuận lại", } ], } ) async def fake_chat(*args: Any, **kwargs: Any) -> str: return mock_diff_response mock_chat.side_effect = fake_chat service = DocumentProcessingService() result = asyncio.run( service.compare_document_versions( old_text="Hợp đồng thanh toán trong vòng 30 ngày kể từ ngày nghiệm thu.", new_text="Hợp đồng thanh toán trong vòng 15 ngày kể từ ngày nghiệm thu.", filename_old="HD_v1.0.docx", filename_new="HD_v1.1_Final.docx", ) ) self.assertIn("summary_of_changes", result) self.assertEqual( result["financial_impact"], "Rút ngắn thời gian thu hồi công nợ" ) self.assertEqual(len(result["key_differences"]), 1) @patch( "app.modules.core.application.document_processing_service.DocumentProcessingService.run_pipeline" ) @patch( "app.modules.core.application.document_processing_service.DocumentProcessingService.compare_document_versions" ) def test_api_document_version_and_diff_endpoints( self, mock_diff: MagicMock, mock_pipeline: MagicMock ) -> None: async def fake_diff(*args: Any, **kwargs: Any) -> dict[str, Any]: return { "summary_of_changes": "Bản đã ký nháy của Trưởng phòng Kỹ thuật", "financial_impact": "Không phát hiện thay đổi", "legal_impact": "Đã ký nháy", "key_differences": [], } mock_diff.side_effect = fake_diff mock_pipeline.return_value = {"status": "success"} app = create_app() def override_get_current_user() -> dict[str, Any]: return { "user_id": "test-superadmin-id", "email": "superadmin@dscons.vn", "role": "superadmin", "full_name": "Test Super Admin", } app.dependency_overrides[get_current_user] = override_get_current_user client = TestClient(app) # 1. Tạo document ban đầu qua API create_res = client.post( "/v1/erp/documents", json={ "document_title": "Tờ trình phê duyệt biện pháp thi công", "document_type": "TTr", "category": "outgoing", "project_stage": "stage_02_technical", "signature_status": "draft", }, ) self.assertEqual(create_res.status_code, 200) doc_data = create_res.json() doc_id = doc_data["id"] self.created_doc_ids.append(doc_id) # 2. Upload phiên bản mới cho document đó mock_file_content = b"Noi dung to trinh da ky nhay va dong dau" upload_ver_res = client.post( f"/v1/erp/documents/{doc_id}/upload-version", files={ "file": ("to_trinh_v2.txt", io.BytesIO(mock_file_content), "text/plain") }, data={ "change_summary": "Bản đã ký nháy của Trưởng phòng Kỹ thuật", "signature_status": "reviewed", }, ) self.assertEqual(upload_ver_res.status_code, 200) ver_doc = upload_ver_res.json() self.created_doc_ids.append(ver_doc["id"]) self.assertEqual(ver_doc["signature_status"], "reviewed") # 3. Lấy danh sách versions qua API versions_res = client.get(f"/v1/erp/documents/{doc_id}/versions") self.assertEqual(versions_res.status_code, 200) v_list = versions_res.json() self.assertEqual(len(v_list), 2) # 4. Cập nhật signature status qua API sig_update_res = client.post( f"/v1/erp/documents/{ver_doc['id']}/signature-status", json={"signature_status": "fully_executed"}, ) self.assertEqual(sig_update_res.status_code, 200) updated_doc = sig_update_res.json() self.assertEqual(updated_doc["signature_status"], "fully_executed") self.assertEqual(updated_doc["audit_readiness_score"], 100) # 5. Lấy Audit Readiness report qua API audit_res = client.get("/v1/erp/documents/audit-readiness") self.assertEqual(audit_res.status_code, 200) audit_data = audit_res.json() self.assertIn("total_documents", audit_data) self.assertIn("audit_readiness_score", audit_data)