from __future__ import annotations import unittest from unittest.mock import patch from app.models.schemas import WorkflowReviewStartRequest from app.modules.core.application.workflow_persistence_service import WorkflowPersistenceService try: from test_workflow_persistence_support import ( FailingOperationalStateService, FakePostgresClient, WorkflowPersistenceServiceTestBase, assert_review_code_pattern, ) except ImportError: from .test_workflow_persistence_support import ( FailingOperationalStateService, FakePostgresClient, WorkflowPersistenceServiceTestBase, assert_review_code_pattern, ) class WorkflowPersistenceServiceUnitTests(WorkflowPersistenceServiceTestBase): def test_build_review_code_has_expected_prefix_and_unique_timestamp_suffix( self, ) -> None: with patch( "app.modules.core.application.workflow_persistence_service.datetime" ) as mocked_datetime: mocked_datetime.now.side_effect = [ type("T", (), {"strftime": lambda self, fmt: "20260101010101"})(), type("T", (), {"strftime": lambda self, fmt: "20260101010102"})(), ] first = WorkflowPersistenceService._build_review_code( "prj-001", "legal_agent" ) second = WorkflowPersistenceService._build_review_code( "prj-001", "legal_agent" ) assert_review_code_pattern(self, first) assert_review_code_pattern(self, second) self.assertNotEqual(first, second) def test_normalize_work_log_status_maps_known_values(self) -> None: self.assertEqual( WorkflowPersistenceService._normalize_work_log_status("Hoàn thành"), "completed", ) self.assertEqual( WorkflowPersistenceService._normalize_work_log_status( "awaiting_assignment" ), "in_progress", ) self.assertEqual( WorkflowPersistenceService._normalize_work_log_status("chưa bắt đầu"), "planned", ) self.assertEqual( WorkflowPersistenceService._normalize_work_log_status("cancelled"), "blocked", ) self.assertEqual( WorkflowPersistenceService._normalize_work_log_status("something-else"), "in_progress", ) def test_normalize_work_log_action_type_maps_workflow_and_text_hints(self) -> None: self.assertEqual( WorkflowPersistenceService._normalize_work_log_action_type( "finding_created", None ), "issue_recorded", ) self.assertEqual( WorkflowPersistenceService._normalize_work_log_action_type( "session_created", None ), "system_sync", ) self.assertEqual( WorkflowPersistenceService._normalize_work_log_action_type( None, "Cập nhật tiến độ hồ sơ" ), "progress_update", ) self.assertEqual( WorkflowPersistenceService._normalize_work_log_action_type( None, "Thiếu biên bản nghiệm thu" ), "issue_recorded", ) self.assertEqual( WorkflowPersistenceService._normalize_work_log_action_type( None, "Công việc thông thường" ), "task_update", ) class WorkflowPersistenceServiceFlowTests(WorkflowPersistenceServiceTestBase): @patch("app.modules.core.application.workflow_policy_service.get_persona") @patch("app.modules.core.application.workflow_persistence_service.get_persona") def test_start_review_run_persists_review_findings_assignments_and_normalized_work_logs( self, mock_get_persona, mock_policy_get_persona, ) -> None: mock_get_persona.return_value = self.mock_persona_payload() mock_policy_get_persona.return_value = self.mock_policy_persona_payload() fake_postgres = FakePostgresClient() service = WorkflowPersistenceService(postgres_client=fake_postgres) request = WorkflowReviewStartRequest.model_validate( { "project_code": "PRJ-001", "lead_agent_code": "legal_agent", "lead_reviewer_employee_code": "EMP-LEAD", "initiated_by_employee_code": "EMP-INIT", "assigned_departments": ["legal"], "baseline_snapshot": { "snapshot_type": "initial_baseline", "captured_by_employee_code": "EMP-LEAD", "captured_by_agent_code": "legal_agent", "label": "Baseline", "summary": "Tổng quan", }, "findings": [ { "finding_code": "F-001", "finding_group": "legal", "finding_type": "missing_document", "title": "Thiếu quyết định phê duyệt", "responsible_department_code": "legal", "assignment": { "assigned_department_code": "legal", "assigned_employee_code": "EMP-LAW-01", "priority": "high", "due_date": "2026-01-10", "status": "pending", }, "employee_actions": [ { "employee_code": "EMP-LAW-01", "agent_code": "legal_agent", "department_code": "legal", "action": "Thiếu quyết định phê duyệt", "status": "Đang xử lý", "ai_result": ["Đã xác định văn bản còn thiếu"], "metadata": {"action_type": "finding_created"}, } ], } ], "participating_employees": [ { "employee_code": "EMP-OPS-01", "agent_code": "legal_agent", "department_code": "legal", "action": "Cập nhật tiến độ rà soát", "status": "awaiting_assignment", "summary": "Đang rà soát checklist", "notes": ["Đã gửi nhắc việc"], "metadata": {"action_type": "session_created"}, } ], } ) summary = service.start_review_run(request) self.assertEqual(summary.review.project_code, "PRJ-001") self.assertEqual(summary.review.status, "closed") self.assertEqual(summary.finding_summaries[0].finding_code, "F-001") self.assertEqual(summary.finding_summaries[0].assignment_status, "verified") self.assertEqual(len(summary.employee_log_summaries), 3) self.assertEqual(summary.created_snapshot_ids, ["snapshot-1"]) self.assertGreaterEqual(summary.review_action_count, 1) self.assertTrue(summary.close_executed) self.assertEqual(summary.automation_status, "completed") self.assertIsNotNone(fake_postgres.created_review_payload) assert fake_postgres.created_review_payload is not None assert_review_code_pattern( self, fake_postgres.created_review_payload["review_code"] ) self.assertEqual( fake_postgres.created_review_payload["metadata"]["workflow_type"], "proactive_dossier_review", ) session_statuses = { item["employee_code"]: item["status"] for item in fake_postgres.upserted_sessions } self.assertEqual(session_statuses["EMP-OPS-01"], "in_progress") self.assertEqual(session_statuses["EMP-LAW-01"], "in_progress") action_types = { item["action"]: item["action_type"] for item in fake_postgres.appended_actions } self.assertEqual(action_types["Cập nhật tiến độ rà soát"], "system_sync") self.assertEqual(action_types["Thiếu quyết định phê duyệt"], "issue_recorded") @patch("app.modules.core.application.workflow_policy_service.get_persona") @patch("app.modules.core.application.workflow_persistence_service.get_persona") def test_start_review_run_returns_fallback_operational_snapshot_when_enrichment_fails( self, mock_get_persona, mock_policy_get_persona, ) -> None: mock_get_persona.return_value = self.mock_persona_payload() mock_policy_get_persona.return_value = self.mock_policy_persona_payload() fake_postgres = FakePostgresClient() service = WorkflowPersistenceService(postgres_client=fake_postgres) service._company_operational_state_service = FailingOperationalStateService() request = WorkflowReviewStartRequest.model_validate( { "project_code": "PRJ-001", "lead_agent_code": "legal_agent", "lead_reviewer_employee_code": "EMP-LEAD", "initiated_by_employee_code": "EMP-INIT", "findings": [ { "finding_code": "F-001", "finding_group": "legal", "finding_type": "missing_document", "title": "Thiếu quyết định phê duyệt", "responsible_department_code": "legal", } ], } ) summary = service.start_review_run(request) self.assertEqual(summary.automation_status, "completed") self.assertEqual(summary.operational_state_snapshot.source, "unavailable") self.assertTrue(summary.operational_state_snapshot.used_fallback) self.assertEqual(summary.operational_state_snapshot.total_projects, 0) self.assertIn( "operational-state-unavailable", summary.operational_state_snapshot.metadata["error"], ) self.assertIn("fallback snapshot an toàn", summary.message) @patch("app.modules.core.application.workflow_policy_service.get_persona") @patch("app.modules.core.application.workflow_persistence_service.get_persona") def test_start_review_run_uses_transaction_wrapper_for_core_write_phase( self, mock_get_persona, mock_policy_get_persona, ) -> None: mock_get_persona.return_value = self.mock_persona_payload() mock_policy_get_persona.return_value = self.mock_policy_persona_payload() fake_postgres = FakePostgresClient() service = WorkflowPersistenceService(postgres_client=fake_postgres) request = WorkflowReviewStartRequest.model_validate( { "project_code": "PRJ-001", "lead_agent_code": "legal_agent", "lead_reviewer_employee_code": "EMP-LEAD", "initiated_by_employee_code": "EMP-INIT", "findings": [ { "finding_code": "F-001", "finding_group": "legal", "finding_type": "missing_document", "title": "Thiếu quyết định phê duyệt", "responsible_department_code": "legal", } ], } ) service.start_review_run(request) self.assertEqual(fake_postgres.transaction_calls, 1) self.assertEqual(fake_postgres.transaction_connection_tokens, ["tx-conn-1"]) self.assertEqual( fake_postgres.received_connections["create_dossier_review_session"], ["tx-conn-1"], ) self.assertEqual( fake_postgres.received_connections["create_dossier_review_finding"], ["tx-conn-1"], ) self.assertEqual( fake_postgres.received_connections["assign_dossier_review_finding"], ["tx-conn-1"], ) self.assertEqual( fake_postgres.received_connections["submit_dossier_review_supplement"], ["tx-conn-1"], ) self.assertEqual( fake_postgres.received_connections["verify_dossier_review_finding"], ["tx-conn-1"], ) self.assertEqual( fake_postgres.received_connections["close_dossier_review_session"], ["tx-conn-1"], ) def test_start_review_run_rolls_back_when_assign_fails(self) -> None: self.assert_failed_transaction_leaves_no_post_commit_side_effects( failure_step="assign_finding:finding-1" ) def test_start_review_run_rolls_back_when_create_session_fails(self) -> None: self.assert_failed_transaction_leaves_no_post_commit_side_effects( failure_step="create_session" ) def test_start_review_run_rolls_back_when_submit_fails(self) -> None: self.assert_failed_transaction_leaves_no_post_commit_side_effects( failure_step="submit_finding:finding-1" ) def test_start_review_run_rolls_back_when_close_fails(self) -> None: self.assert_failed_transaction_leaves_no_post_commit_side_effects( failure_step="close_session" ) if __name__ == "__main__": unittest.main()