import unittest from types import SimpleNamespace from app.modules.operations.application.company_operational_state_service import ( CompanyOperationalStateService, ) 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): return list(self._sessions) class FailingDossierReviewPostgresClient(StubPostgresClient): def fetch_dossier_review_sessions(self): raise RuntimeError("dossier-reviews-unavailable") class StubProjectManagementService: def get_project_management_data(self): risk = SimpleNamespace( title="Thiếu hồ sơ pháp lý", severity="high", status="open", owner="legal", mitigation="Bổ sung quyết định phê duyệt", ) task = SimpleNamespace( title="Theo dõi tiến độ bổ sung", assignee="NV001", status="pending", priority="high", due_date=None, note="Cần theo dõi hàng ngày", ) project = SimpleNamespace( project_code="PRJ-001", project_name="Dinh Son", client_name="Client A", location="Kien Minh", project_manager="PM001", status="active", current_phase="implementation", overall_progress_percent=55, budget_status="on_track", risks=[risk], upcoming_tasks=[task], milestones=[], phase_progress=[], timeline=[], summary_cards=[], ) return SimpleNamespace( items=[project], source="postgres", used_fallback=False, message=None, ) class StubEmployeeLogsService: def get_employee_logs(self): item = SimpleNamespace(status="in_progress") return SimpleNamespace( items=[item], source="postgres", used_fallback=False, message=None, ) class FailingOperationsCaseManagementService: def list_cases(self): raise RuntimeError("cases-unavailable") def list_actions(self): raise RuntimeError("actions-unavailable") def list_blockers(self): raise RuntimeError("blockers-unavailable") class FailingRemediationPlanningService: def list_plans(self): raise RuntimeError("remediation-unavailable") class FailingProjectManagementService: def get_project_management_data(self): raise RuntimeError("projects-unavailable") class StubRemediationPlan: def __init__(self, *, finding_code: object = "F-PLAN-001") -> None: self.project_code = "PRJ-001" self.review_session_id = "review-001" self.finding_id = "finding-001" self.finding_code = finding_code self.document_type = "decision" self.dossier_stage = "legal" self.priority = "high" self.owner_department_code = "legal" self.owner_employee_code = "EMP-001" self.readiness_summary = SimpleNamespace( status="blocked", blocking_reasons=["missing-input"] ) self.destination_folder = "/tmp/out" self.output_filename = "decision.docx" self.source = "derived" self.used_fallback = False self.message = None class StubRemediationPlanningService: def list_plans(self): return SimpleNamespace( items=[StubRemediationPlan()], source="postgres", used_fallback=False, message=None, ) class CompanyOperationalStateServiceTests(unittest.TestCase): def test_get_company_operational_state_returns_fallbacks_when_optional_services_fail( self, ) -> None: service = CompanyOperationalStateService( postgres_client=StubPostgresClient( sessions=[ { "review_id": "review-001", "review_code": "RVW-001", "project_code": "PRJ-001", "project_name": "Dinh Son", "dossier_scope": "project_dossier", "trigger_source": "system", "status": "assigned", "findings": [ { "finding_id": "finding-001", "finding_code": "F-001", "title": "Thiếu quyết định", "finding_group": "compliance", "finding_type": "missing_document", "severity": "high", "impact_level": "high", "status": "open", "supplement_status": "pending", "detected_at": "2026-04-03T09:00:00Z", "responsible_department_code": "legal", "metadata": { "blocking_reasons": [ "Chưa có quyết định phê duyệt" ], }, } ], } ] ), project_management_service=StubProjectManagementService(), employee_logs_service=StubEmployeeLogsService(), operations_case_management_service=FailingOperationsCaseManagementService(), remediation_planning_service=FailingRemediationPlanningService(), ) response = service.get_company_operational_state() self.assertTrue(response.used_fallback) self.assertEqual(response.summary.total_projects, 1) self.assertEqual(response.summary.open_cases_count, 0) self.assertEqual(response.summary.open_actions_count, 0) self.assertEqual(len(response.projects), 1) self.assertEqual(len(response.risks), 2) self.assertEqual(len(response.blocked_items), 2) self.assertGreaterEqual(response.summary.blocked_items_count, 2) self.assertIn("cases-unavailable", response.message) self.assertIn("actions-unavailable", response.message) self.assertIn("blockers-unavailable", response.message) self.assertIn("remediation-unavailable", response.message) source_status_by_name = { item.source_name: item for item in response.source_status } self.assertEqual( source_status_by_name["operations_cases"].source, "unavailable" ) self.assertTrue(source_status_by_name["operations_cases"].used_fallback) self.assertFalse(source_status_by_name["operations_cases"].available) self.assertEqual( source_status_by_name["operations_actions"].source, "unavailable" ) self.assertTrue(source_status_by_name["operations_actions"].used_fallback) self.assertFalse(source_status_by_name["operations_actions"].available) self.assertEqual( source_status_by_name["operations_blockers"].source, "unavailable" ) self.assertTrue(source_status_by_name["operations_blockers"].used_fallback) self.assertFalse(source_status_by_name["operations_blockers"].available) self.assertEqual( source_status_by_name["remediation_plans"].source, "unavailable" ) self.assertTrue(source_status_by_name["remediation_plans"].used_fallback) self.assertFalse(source_status_by_name["remediation_plans"].available) def test_get_company_operational_state_returns_safe_fallback_when_primary_project_source_fails( self, ) -> None: service = CompanyOperationalStateService( postgres_client=StubPostgresClient([]), project_management_service=FailingProjectManagementService(), employee_logs_service=StubEmployeeLogsService(), ) response = service.get_company_operational_state() self.assertTrue(response.used_fallback) self.assertEqual(response.summary.total_projects, 0) self.assertEqual(len(response.projects), 0) self.assertEqual(len(response.risks), 0) self.assertEqual(len(response.blocked_items), 0) self.assertEqual(len(response.backlog), 0) self.assertIn("projects-unavailable", response.message) source_status_by_name = { item.source_name: item for item in response.source_status } self.assertEqual(source_status_by_name["projects"].source, "unavailable") self.assertTrue(source_status_by_name["projects"].used_fallback) self.assertFalse(source_status_by_name["projects"].available) def test_get_company_operational_state_marks_dossier_review_source_unavailable_when_fetch_fails( self, ) -> None: service = CompanyOperationalStateService( postgres_client=FailingDossierReviewPostgresClient(), project_management_service=StubProjectManagementService(), employee_logs_service=StubEmployeeLogsService(), ) response = service.get_company_operational_state() self.assertTrue(response.used_fallback) self.assertIn("dossier-reviews-unavailable", response.message) source_status_by_name = { item.source_name: item for item in response.source_status } self.assertEqual(source_status_by_name["dossier_reviews"].source, "unavailable") self.assertTrue(source_status_by_name["dossier_reviews"].used_fallback) self.assertFalse(source_status_by_name["dossier_reviews"].available) self.assertEqual(source_status_by_name["dossier_reviews"].record_count, 0) def test_get_company_operational_state_returns_fallback_optional_payload_when_summary_builders_fail( self, ) -> None: service = CompanyOperationalStateService( postgres_client=StubPostgresClient([]), project_management_service=StubProjectManagementService(), employee_logs_service=StubEmployeeLogsService(), remediation_planning_service=StubRemediationPlanningService(), ) original_build_policy_overview = service._build_policy_overview original_build_state_summaries = service._build_state_summaries original_build_remediation_plan_summary = ( service._build_remediation_plan_summary ) def failing_build_policy_overview(*, dossier_reviews): raise RuntimeError("policy-overview-unavailable") def failing_build_state_summaries(**kwargs): raise RuntimeError("state-summaries-unavailable") def failing_build_remediation_plan_summary(item): raise RuntimeError("remediation-plan-summary-unavailable") service._build_policy_overview = failing_build_policy_overview service._build_state_summaries = failing_build_state_summaries service._build_remediation_plan_summary = failing_build_remediation_plan_summary try: response = service.get_company_operational_state() finally: service._build_policy_overview = original_build_policy_overview service._build_state_summaries = original_build_state_summaries service._build_remediation_plan_summary = ( original_build_remediation_plan_summary ) self.assertTrue(response.used_fallback) self.assertEqual(response.summary.total_projects, 1) self.assertEqual(response.policy_overview, {}) self.assertEqual(response.remediation_summary, []) self.assertEqual(response.remediation_plans, []) self.assertIn("policy-overview-unavailable", response.message) self.assertIn("state-summaries-unavailable", response.message) self.assertIn("remediation-plan-summary-unavailable", response.message) if __name__ == "__main__": unittest.main()