"""Unit and Integration Tests for Document Deletion, Duplicate Detection, and Title Sanitization.""" from __future__ import annotations import unittest 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 TestDocumentDeleteAndDuplicateDetection(unittest.TestCase): def setUp(self) -> None: self.erp_client = ErpDatabaseClient() self.doc_service = DocumentProcessingService(erp_client=self.erp_client) self.created_doc_ids: list[str] = [] def tearDown(self) -> None: for doc_id in self.created_doc_ids: try: self.erp_client.delete_document(doc_id, delete_all_versions=True) except Exception: pass def test_sanitize_document_title(self) -> None: """Kiểm tra hàm làm sạch trích yếu văn bản, loại bỏ các từ khóa bảo mật và ký tự rác.""" # 1. Trích yếu có chứa từ 'bảo mật' ở cuối chuỗi title_1 = "Mặt bằng thi công bảo mật" clean_1 = DocumentProcessingService.sanitize_document_title(title_1) self.assertEqual(clean_1, "Mặt bằng thi công") # 2. Trích yếu có gắn tag [BẢO MẬT] hoặc (bảo mật) title_2 = "[BẢO MẬT] Biên bản bàn giao mốc cao độ trắc địa" clean_2 = DocumentProcessingService.sanitize_document_title(title_2) self.assertEqual(clean_2, "Biên bản bàn giao mốc cao độ trắc địa") title_3 = "Báo cáo tiến độ thi công tháng 03 (bảo mật zero-trust)" clean_3 = DocumentProcessingService.sanitize_document_title(title_3) self.assertEqual(clean_3, "Báo cáo tiến độ thi công tháng 03") # 3. Trích yếu dính thẻ nhãn XML / HTML giả mạo title_4 = " Hợp đồng kinh tế số 12/2026/HĐ-DSC " clean_4 = DocumentProcessingService.sanitize_document_title(title_4) self.assertEqual(clean_4, "Hợp đồng kinh tế số 12/2026/HĐ-DSC") # 4. Fallback khi chuỗi rỗng clean_5 = DocumentProcessingService.sanitize_document_title( "", "to_trinh_phe_duyet.pdf" ) self.assertEqual(clean_5, "to trinh phe duyet") def test_delete_single_document(self) -> None: """Kiểm tra chức năng xóa một văn bản đơn lẻ.""" # 1. Tạo document mẫu doc = self.erp_client.create_document( { "document_code": "TEST-DEL-01", "document_title": "Văn bản thử nghiệm xóa đơn lẻ", "document_type": "CV", "category": "incoming", } ) doc_id = doc["id"] # Kiểm tra văn bản đã tồn tại fetched = self.erp_client.get_document(doc_id) self.assertIsNotNone(fetched) # 2. Xóa văn bản res = self.erp_client.delete_document(doc_id, delete_all_versions=False) self.assertTrue(res["success"]) self.assertEqual(res["deleted_count"], 1) # 3. Kiểm tra văn bản không còn trong database deleted_check = self.erp_client.get_document(doc_id) self.assertIsNone(deleted_check) def test_delete_version_and_promote_remaining_active(self) -> None: """Kiểm tra khi xóa version mới nhất (active), version trước đó tự động được active.""" # 1. Tạo v1.0 doc_v1 = self.erp_client.create_document( { "document_code": "TEST-VER-01", "document_title": "Hồ sơ thiết kế v1", "document_type": "CV", "version_number": "v1.0", } ) v1_id = doc_v1["id"] self.created_doc_ids.append(v1_id) # 2. Tạo v1.1 doc_v2 = self.erp_client.create_document_version( parent_doc_id=v1_id, payload={ "document_title": "Hồ sơ thiết kế v2 sửa đổi", "change_summary": "Cập nhật cao trình móng", }, ) v2_id = doc_v2["id"] self.created_doc_ids.append(v2_id) # Kiểm tra v1.1 đang là active, v1.0 là inactive self.assertTrue(self.erp_client.get_document(v2_id)["is_active_version"]) self.assertFalse(self.erp_client.get_document(v1_id)["is_active_version"]) # 3. Xóa v1.1 (chỉ xóa bản ghi này) res_del = self.erp_client.delete_document(v2_id, delete_all_versions=False) self.assertTrue(res_del["success"]) # 4. Kiểm tra v1.0 tự động được kích hoạt lại làm is_active_version = True v1_updated = self.erp_client.get_document(v1_id) self.assertTrue(v1_updated["is_active_version"]) def test_delete_all_versions(self) -> None: """Kiểm tra xóa toàn bộ chùm phiên bản.""" doc_v1 = self.erp_client.create_document( { "document_code": "TEST-THREAD-01", "document_title": "Tờ trình v1", "document_type": "TTr", } ) v1_id = doc_v1["id"] doc_v2 = self.erp_client.create_document_version( parent_doc_id=v1_id, payload={"document_title": "Tờ trình v2"} ) v2_id = doc_v2["id"] # Xóa toàn bộ chùm res_del = self.erp_client.delete_document(v2_id, delete_all_versions=True) self.assertTrue(res_del["success"]) self.assertGreaterEqual(res_del["deleted_count"], 2) self.assertIsNone(self.erp_client.get_document(v1_id)) self.assertIsNone(self.erp_client.get_document(v2_id)) def test_check_duplicate_document(self) -> None: """Kiểm tra cơ chế phát hiện văn bản trùng lặp.""" # 1. Tạo một văn bản với file_hash cụ thể mock_hash = "a1b2c3d4e5f678901234567890abcdef1234567890abcdef1234567890abcdef" doc = self.erp_client.create_document( { "document_code": "DUP-CHECK-01", "document_title": "Biên bản khảo sát hiện trạng", "document_type": "BB", "file_size_bytes": 10240, "file_hash": mock_hash, } ) doc_id = doc["id"] self.created_doc_ids.append(doc_id) # 2. Kiểm tra trùng theo Hash dup_by_hash = self.erp_client.check_duplicate_document(file_hash=mock_hash) self.assertIsNotNone(dup_by_hash) self.assertEqual(dup_by_hash["id"], doc_id) self.assertIn("Mã băm", dup_by_hash.get("duplicate_reason", "")) # 3. Kiểm tra trùng theo Số hiệu văn bản dup_by_code = self.erp_client.check_duplicate_document( document_code="DUP-CHECK-01" ) self.assertIsNotNone(dup_by_code) self.assertEqual(dup_by_code["id"], doc_id) # 4. Kiểm tra trùng theo Tên file + Kích thước dup_by_name_size = self.erp_client.check_duplicate_document( filename="Biên bản khảo sát hiện trạng", file_size_bytes=10240 ) self.assertIsNotNone(dup_by_name_size) self.assertEqual(dup_by_name_size["id"], doc_id) # 5. File không trùng lặp no_dup = self.erp_client.check_duplicate_document( file_hash="ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff" ) self.assertIsNone(no_dup) def test_api_delete_document_endpoint(self) -> None: """Kiểm tra API endpoint DELETE /v1/erp/documents/{doc_id} với phân quyền RBAC.""" app = create_app() # 1. Tạo document doc = self.erp_client.create_document( { "document_code": "API-DEL-01", "document_title": "Văn bản xóa qua API", "document_type": "CV", } ) doc_id = doc["id"] self.created_doc_ids.append(doc_id) # 2. Gọi không có Token -> 401 unauth_client = TestClient(app) res_unauth = unauth_client.delete(f"/v1/erp/documents/{doc_id}") self.assertEqual(res_unauth.status_code, 401) # 3. Gọi với Token hợp lệ (Kỹ sư / PM / Superadmin) -> 200 OK app.dependency_overrides[get_current_user] = lambda: { "user_id": "usr-01", "email": "engineer.quynh@dscons.vn", "role": "pm", } auth_client = TestClient(app) res_auth = auth_client.delete( f"/v1/erp/documents/{doc_id}?delete_all_versions=true" ) self.assertEqual(res_auth.status_code, 200) self.assertTrue(res_auth.json()["success"]) # Xác minh đã xóa trong DB self.assertIsNone(self.erp_client.get_document(doc_id)) if __name__ == "__main__": unittest.main()