import json import tempfile import unittest from pathlib import Path from app.models.schemas import ( CompanyOperationalStateSummaryItem, DossierCoverageResponse, RemediationArtifactRequirementItem, RemediationBuildRequest, RemediationGapItem, RemediationPlanItem, RemediationRequiredInputItem, RemediationTaskItem, ) from app.modules.core.application.remediation_planning_service import RemediationPlanningService class StubPostgresClient: def __init__(self, sessions=None, enabled=True): self._sessions = sessions if sessions is not None else [] self._enabled = enabled def is_enabled(self): return self._enabled def fetch_dossier_review_sessions(self, project_code=None, review_id=None): rows = list(self._sessions) if project_code: rows = [ session for session in rows if session.get("project_code") == project_code ] if review_id: rows = [ session for session in rows if session.get("review_id") == review_id ] return rows class RemediationPlanningSchemasTests(unittest.TestCase): def test_nested_schema_models_validate_payload(self) -> None: plan = RemediationPlanItem.model_validate( { "plan_id": "remediation-plan-1", "project_code": "PRJ-001", "review_session_id": "review-001", "finding_id": "finding-001", "finding_code": "F-001", "document_type": "bien_ban_nghiem_thu_noi_bo", "dossier_stage": "hoan_thien_ho_so", "priority": "high", "owner_department_code": "QA", "owner_employee_code": "NV001", "destination_folder": "/du-an/PRJ-001/nghiem-thu-noi-bo", "output_filename": "PRJ-001_bien-ban-nghiem-thu-noi-bo_F-001.docx", "naming_rule": "{project_code}_{document_type}_{finding_code}.docx", "required_inputs": [ { "input_code": "ma_du_an", "label": "Mã dự án", "required": True, "source_hint": "session.project_code", } ], "artifact_requirements": [ { "artifact_code": "mau_bien_ban", "label": "Mẫu biên bản nghiệm thu nội bộ", "artifact_type": "template", "required": True, "description": "Biểu mẫu đã được phê duyệt", } ], "remediation_tasks": [ { "task_id": "remediation-plan-1:task:1", "task_code": "thu_thap_minh_chung", "title": "Thu thập minh chứng", "task_type": "data_collection", "priority": "high", "sequence_no": 1, "status": "pending", "blocking": True, "expected_output_codes": ["mau_bien_ban"], } ], "readiness_summary": { "status": "needs_review", "ready_input_count": 1, "missing_input_count": 0, "blocking_input_count": 0, "artifact_count": 1, "task_count": 1, "blocking_reasons": [], "coverage_gap_count": 0, "message": "Đủ dữ liệu nền để lập kế hoạch", }, "message": "Cần rà soát mẫu và minh chứng trước khi soạn hồ sơ", "source": "derived_from_sessions", "used_fallback": False, } ) request = RemediationBuildRequest.model_validate( { "project_code": "PRJ-001", "review_session_id": "review-001", "finding_id": "finding-001", "finding_code": "F-001", "document_type": "bien_ban_nghiem_thu_noi_bo", } ) summary = CompanyOperationalStateSummaryItem.model_validate( { "code": "remediation_ready", "label": "Sẵn sàng remediation", "value": "1", "severity": "info", "metadata": { "plans": [ { "finding_code": "F-001", "document_type": "bien_ban_nghiem_thu_noi_bo", "priority": "high", "readiness_summary": "Đủ dữ liệu nền để lập kế hoạch", } ] }, } ) gap = RemediationGapItem.model_validate( { "project_code": "PRJ-001", "review_session_id": "review-001", "finding_id": "finding-001", "finding_code": "F-001", "document_type": "bien_ban_nghiem_thu_noi_bo", "gap_status": "blocked", "blocking_reasons": ["Thiếu input: Căn cứ hồ sơ chất lượng"], "missing_input_codes": ["can_cu_ho_so_chat_luong"], "missing_input_labels": ["Căn cứ hồ sơ chất lượng"], "requires_document_type_mapping": False, "can_build_plan": False, "plan_id": "remediation-plan-1", "source": "postgres", "used_fallback": False, } ) coverage = DossierCoverageResponse.model_validate( { "project_code": "PRJ-001", "manifest_path": "tmp/coverage.json", "project": { "project_code": "PRJ-001", "project_name": "Dự án 01", "total_files": 2, "extractable_files": 2, "classified_files": 1, "unknown_files": 1, "knowledge_files_ingested": 1, "knowledge_chunks_ingested": 4, "document_type_counts": {"hop_dong": 1}, "missing_document_types": ["bien_ban_nghiem_thu_noi_bo"], "documents": [ { "relative_path": "A/file.docx", "source_filename": "file.docx", "document_type": "hop_dong", "doc_stage": "hop_dong", "classification_source": "rule", "classification_confidence": 0.9, } ], "readiness_gaps": ["Thiếu document types trọng yếu"], }, "source": "manifest", "used_fallback": False, } ) self.assertIsInstance(plan.required_inputs[0], RemediationRequiredInputItem) self.assertIsInstance( plan.artifact_requirements[0], RemediationArtifactRequirementItem ) self.assertIsInstance(plan.remediation_tasks[0], RemediationTaskItem) self.assertEqual(request.document_type, "bien_ban_nghiem_thu_noi_bo") self.assertEqual(summary.metadata["plans"][0]["finding_code"], "F-001") self.assertEqual(gap.gap_status, "blocked") self.assertEqual(coverage.project.documents[0].document_type, "hop_dong") class RemediationPlanningServiceTests(unittest.TestCase): def _write_manifest(self, payload: dict) -> str: temp_dir = tempfile.TemporaryDirectory() self.addCleanup(temp_dir.cleanup) path = Path(temp_dir.name) / "coverage.json" path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8") return str(path) def test_specialized_plan_for_bien_ban_nghiem_thu_noi_bo(self) -> None: sessions = [ { "review_id": "review-001", "project_code": "PRJ-001", "findings": [ { "finding_id": "finding-001", "finding_code": "F-001", "document_type": "bien_ban_nghiem_thu_noi_bo", "dossier_stage": "hoan_thien_ho_so", "status": "open", "severity": "high", "impact_level": "high", "responsible_department_code": "quality_assurance", "assignments": [ { "assigned_department_code": "quality_assurance", "assigned_employee_code": "NV001", "status": "in_progress", "priority": "high", } ], "metadata": { "work_item_name": "Đổ bê tông cống hộp", "quality_reference": "BV-01, TCVN-02", "attendees": ["QA", "Chỉ huy trưởng"], "inspection_result": "Đạt yêu cầu", "inspection_date": "2026-03-28", }, } ], } ] manifest_path = self._write_manifest( { "projects": [ { "project_code": "PRJ-001", "project_name": "Dự án 01", "total_files": 3, "extractable_files": 3, "classified_files": 3, "unknown_files": 0, "knowledge_files_ingested": 3, "knowledge_chunks_ingested": 9, "document_type_counts": {"bien_ban_nghiem_thu_noi_bo": 1}, "missing_document_types": [], "documents": [], } ] } ) service = RemediationPlanningService( postgres_client=StubPostgresClient(sessions), manifest_path=manifest_path, ) response = service.list_plans(project_code="PRJ-001") self.assertEqual(response.source, "postgres") self.assertFalse(response.used_fallback) self.assertEqual(len(response.items), 1) plan = response.items[0] self.assertEqual(plan.document_type, "bien_ban_nghiem_thu_noi_bo") self.assertFalse(plan.used_fallback) self.assertGreaterEqual(len(plan.required_inputs), 5) self.assertGreaterEqual(len(plan.artifact_requirements), 3) self.assertGreaterEqual(len(plan.remediation_tasks), 4) self.assertEqual(plan.readiness_summary.status, "ready") artifact_codes = {item.artifact_code for item in plan.artifact_requirements} input_codes = {item.input_code for item in plan.required_inputs} task_codes = {item.task_code for item in plan.remediation_tasks} self.assertIn("ban_du_thao_bien_ban", artifact_codes) self.assertIn("thanh_phan_tham_du", input_codes) self.assertIn("lap_du_thao_bien_ban", task_codes) self.assertIn("bien_ban_nghiem_thu_noi_bo", plan.output_filename) def test_generic_document_type_uses_safe_fallback_plan(self) -> None: sessions = [ { "review_id": "review-002", "project_code": "PRJ-002", "findings": [ { "finding_id": "finding-002", "finding_code": "F-002", "document_type": "tai_lieu_khac", "dossier_stage": "bo_sung", "status": "open", } ], } ] service = RemediationPlanningService( postgres_client=StubPostgresClient(sessions) ) response = service.list_plans(project_code="PRJ-002") self.assertEqual(len(response.items), 1) plan = response.items[0] self.assertEqual(plan.document_type, "tai_lieu_khac") self.assertTrue(plan.used_fallback) self.assertGreaterEqual(len(plan.required_inputs), 1) self.assertGreaterEqual(len(plan.artifact_requirements), 1) self.assertGreaterEqual(len(plan.remediation_tasks), 1) self.assertIn("tai_lieu_khac", plan.output_filename) self.assertEqual(plan.readiness_summary.status, "blocked") def test_empty_data_returns_safe_empty_response(self) -> None: service = RemediationPlanningService(postgres_client=StubPostgresClient([])) response = service.list_plans(project_code="PRJ-EMPTY") self.assertEqual(response.items, []) self.assertFalse(response.used_fallback) self.assertEqual(response.source, "postgres") self.assertTrue(response.message) def test_disabled_postgres_returns_safe_empty_response(self) -> None: service = RemediationPlanningService( postgres_client=StubPostgresClient([], enabled=False) ) response = service.list_plans(project_code="PRJ-DISABLED") self.assertEqual(response.items, []) self.assertTrue(response.used_fallback) self.assertTrue(response.message) def test_build_plan_can_target_specific_finding(self) -> None: sessions = [ { "review_id": "review-003", "project_code": "PRJ-003", "findings": [ { "finding_id": "finding-003", "finding_code": "F-003", "document_type": "bien_ban_nghiem_thu_noi_bo", "status": "open", }, { "finding_id": "finding-004", "finding_code": "F-004", "document_type": "tai_lieu_khac", "status": "open", }, ], } ] service = RemediationPlanningService( postgres_client=StubPostgresClient(sessions) ) response = service.build_plan( RemediationBuildRequest( project_code="PRJ-003", review_session_id="review-003", finding_code="F-004", document_type="tai_lieu_khac", ) ) self.assertIsNotNone(response.item) self.assertEqual(response.item.finding_code, "F-004") self.assertEqual(response.item.document_type, "tai_lieu_khac") self.assertTrue(response.item.used_fallback) def test_list_plans_returns_fallback_when_coverage_manifest_is_missing( self, ) -> None: sessions = [ { "review_id": "review-020", "project_code": "PRJ-020", "findings": [ { "finding_id": "finding-020", "finding_code": "F-020", "document_type": "bien_ban_nghiem_thu_noi_bo", "status": "open", "metadata": { "work_item_name": "Hạng mục A", "quality_reference": "BV-20", "attendees": ["QA"], "inspection_result": "Đạt", "inspection_date": "2026-04-03", }, } ], } ] service = RemediationPlanningService( postgres_client=StubPostgresClient(sessions), manifest_path="tmp/manifest-does-not-exist.json", ) response = service.list_plans(project_code="PRJ-020") self.assertEqual(len(response.items), 1) self.assertTrue(response.used_fallback) self.assertIn("coverage manifest", response.message.lower()) self.assertTrue(response.items[0].used_fallback) self.assertEqual(response.items[0].readiness_summary.status, "needs_review") self.assertEqual(response.items[0].readiness_summary.blocking_reasons, []) self.assertIn("fallback an toàn", response.items[0].message) def test_build_plan_returns_fallback_when_coverage_manifest_is_invalid( self, ) -> None: sessions = [ { "review_id": "review-021", "project_code": "PRJ-021", "findings": [ { "finding_id": "finding-021", "finding_code": "F-021", "document_type": "bien_ban_nghiem_thu_noi_bo", "status": "open", } ], } ] temp_dir = tempfile.TemporaryDirectory() self.addCleanup(temp_dir.cleanup) manifest_path = Path(temp_dir.name) / "coverage.json" manifest_path.write_text("{invalid json", encoding="utf-8") service = RemediationPlanningService( postgres_client=StubPostgresClient(sessions), manifest_path=str(manifest_path), ) response = service.build_plan( RemediationBuildRequest( project_code="PRJ-021", review_session_id="review-021", finding_code="F-021", document_type="bien_ban_nghiem_thu_noi_bo", ) ) self.assertIsNotNone(response.item) self.assertTrue(response.used_fallback) self.assertTrue(response.item.used_fallback) self.assertIn("coverage manifest", response.message.lower()) def test_list_gaps_surfaces_missing_inputs_and_coverage_gaps(self) -> None: sessions = [ { "review_id": "review-010", "project_code": "PRJ-010", "findings": [ { "finding_id": "finding-010", "finding_code": "F-010", "document_type": "bien_ban_nghiem_thu_noi_bo", "dossier_stage": "bo_sung", "status": "open", "metadata": {"work_item_name": "Móng đá 1x2"}, } ], } ] manifest_path = self._write_manifest( { "projects": [ { "project_code": "PRJ-010", "project_name": "Dự án 10", "total_files": 4, "extractable_files": 4, "classified_files": 3, "unknown_files": 1, "knowledge_files_ingested": 3, "knowledge_chunks_ingested": 8, "document_type_counts": {"hop_dong": 1}, "missing_document_types": ["bien_ban_nghiem_thu_noi_bo"], "documents": [], } ] } ) service = RemediationPlanningService( postgres_client=StubPostgresClient(sessions), manifest_path=manifest_path, ) response = service.list_gaps(project_code="PRJ-010") self.assertEqual(len(response.items), 1) gap = response.items[0] self.assertEqual(gap.gap_status, "blocked") self.assertIn("can_cu_ho_so_chat_luong", gap.missing_input_codes) self.assertTrue( any("chưa phân loại" in reason.lower() for reason in gap.blocking_reasons) ) self.assertFalse(gap.can_build_plan) def test_list_gaps_marks_manifest_only_gap_as_needs_review_not_blocked( self, ) -> None: sessions = [ { "review_id": "review-011", "project_code": "PRJ-011", "findings": [ { "finding_id": "finding-011", "finding_code": "F-011", "document_type": "bien_ban_nghiem_thu_noi_bo", "dossier_stage": "bo_sung", "status": "open", "metadata": { "work_item_name": "Móng đá 1x2", "quality_reference": "BV-11", "attendees": ["QA"], "inspection_result": "Đạt", "inspection_date": "2026-04-03", }, } ], } ] manifest_path = self._write_manifest( { "projects": [ { "project_code": "PRJ-011", "project_name": "Dự án 11", "total_files": 2, "extractable_files": 2, "classified_files": 2, "unknown_files": 0, "knowledge_files_ingested": 2, "knowledge_chunks_ingested": 5, "document_type_counts": {"hop_dong": 1}, "missing_document_types": ["bien_ban_nghiem_thu_noi_bo"], "documents": [], } ] } ) service = RemediationPlanningService( postgres_client=StubPostgresClient(sessions), manifest_path=manifest_path, ) response = service.list_gaps(project_code="PRJ-011") self.assertEqual(len(response.items), 1) gap = response.items[0] self.assertEqual(gap.gap_status, "needs_review") self.assertEqual(gap.blocking_reasons, []) self.assertTrue(gap.can_build_plan) def test_get_project_coverage_reads_manifest_payload(self) -> None: manifest_path = self._write_manifest( { "ingest_batch": "hd2026-batch-001", "generated_at": "2026-04-03T02:00:00Z", "projects": [ { "project_code": "PRJ-COVER", "project_name": "Dự án Coverage", "project_root": "HĐ-2026/PRJ-COVER", "project_root_exists": True, "total_files": 5, "extractable_files": 4, "classified_files": 3, "unknown_files": 1, "knowledge_files_ingested": 3, "knowledge_chunks_ingested": 10, "document_type_counts": {"hop_dong": 2}, "missing_document_types": ["bien_ban_nghiem_thu_noi_bo"], "sample_unknown_files": ["unknown/a.txt"], "sample_skipped_files": ["skip/b.dwl"], "missing_files": ["gone/c.docx"], "ingested_document_types": ["hop_dong"], "stages": ["hop_dong"], "documents": [ { "relative_path": "PRJ-COVER/contract.docx", "source_filename": "contract.docx", "document_type": "hop_dong", "doc_stage": "hop_dong", "classification_source": "rule", "classification_confidence": 0.95, } ], } ], } ) service = RemediationPlanningService( postgres_client=StubPostgresClient([]), manifest_path=manifest_path, ) response = service.get_project_coverage("PRJ-COVER") self.assertEqual(response.project_code, "PRJ-COVER") self.assertEqual(response.source, "manifest") self.assertEqual(response.project.unknown_files, 1) self.assertEqual(response.project.documents[0].document_type, "hop_dong") self.assertTrue(response.project.readiness_gaps) def test_get_project_coverage_returns_fallback_when_manifest_is_missing( self, ) -> None: service = RemediationPlanningService( postgres_client=StubPostgresClient([]), manifest_path="tmp/coverage-manifest-missing.json", ) response = service.get_project_coverage("PRJ-FALLBACK") self.assertEqual(response.project_code, "PRJ-FALLBACK") self.assertEqual(response.source, "fallback") self.assertTrue(response.used_fallback) self.assertEqual(response.project.total_files, 0) self.assertFalse(response.project.documents) self.assertTrue(response.project.readiness_gaps) self.assertIn("fallback coverage rỗng an toàn", response.message) def test_get_project_coverage_returns_fallback_when_manifest_is_invalid( self, ) -> None: temp_dir = tempfile.TemporaryDirectory() self.addCleanup(temp_dir.cleanup) manifest_path = Path(temp_dir.name) / "coverage.json" manifest_path.write_text("{invalid json", encoding="utf-8") service = RemediationPlanningService( postgres_client=StubPostgresClient([]), manifest_path=str(manifest_path), ) response = service.get_project_coverage("PRJ-INVALID") self.assertEqual(response.project_code, "PRJ-INVALID") self.assertEqual(response.source, "fallback") self.assertTrue(response.used_fallback) self.assertEqual(response.project.unknown_files, 0) self.assertIn("coverage manifest", response.message.lower()) def test_get_project_coverage_returns_fallback_for_missing_project(self) -> None: manifest_path = self._write_manifest({"projects": []}) service = RemediationPlanningService( postgres_client=StubPostgresClient([]), manifest_path=manifest_path, ) response = service.get_project_coverage("PRJ-404") self.assertEqual(response.project_code, "PRJ-404") self.assertEqual(response.source, "fallback") self.assertTrue(response.used_fallback) self.assertEqual(response.project.total_files, 0) self.assertTrue(response.project.readiness_gaps) self.assertIn("fallback coverage rỗng an toàn", response.message) if __name__ == "__main__": unittest.main()