from __future__ import annotations import re import unittest from copy import deepcopy from typing import Any from unittest.mock import patch from app.models.schemas import WorkflowReviewStartRequest from app.modules.core.application.workflow_persistence_service import WorkflowPersistenceService class FailingOperationalStateService: def get_company_operational_state(self) -> dict[str, Any]: raise RuntimeError("operational-state-unavailable") def make_review_session( *, review_id: str = "review-1", review_code: str = "RVW-PRJ-AGENT-20260101000000" ) -> dict[str, Any]: return { "review_id": review_id, "review_code": review_code, "project_code": "PRJ-001", "project_name": "Dự án thử nghiệm", "dossier_scope": "project_dossier", "trigger_source": "system_proactive_review", "initiated_by_employee_code": "EMP-INIT", "initiated_by_employee_name": "Người khởi tạo", "lead_reviewer_employee_code": "EMP-LEAD", "lead_reviewer_employee_name": "Trưởng nhóm", "lead_agent_code": "legal_agent", "status": "in_review", "review_reason": "Rà soát chủ động", "review_summary": None, "assigned_departments": [], "open_findings_count": 0, "resolved_findings_count": 0, "overdue_assignments_count": 0, "last_reviewed_at": "2026-01-01T00:00:00Z", "closed_at": None, "findings": [], "snapshots": [ { "snapshot_id": "snapshot-1", "snapshot_type": "initial_baseline", "label": "Baseline", "summary": "Tổng quan ban đầu", "captured_at": "2026-01-01T00:00:00Z", "captured_by_agent_code": "legal_agent", "captured_by_employee_code": "EMP-LEAD", "captured_by_employee_name": "Trưởng nhóm", "dossier_state": {}, "missing_matrix": {}, "readiness_summary": {}, } ], "actions": [ { "action_id": "action-session-created", "action_type": "session_created", "actor_type": "agent", "actor_employee_code": "EMP-LEAD", "actor_employee_name": "Trưởng nhóm", "actor_agent_code": "legal_agent", "action_summary": "Tạo phiên rà soát", "before_state": {}, "after_state": {"status": "in_review"}, "action_payload": {}, "created_at": "2026-01-01T00:00:00Z", } ], "metadata": {"workflow_type": "proactive_dossier_review"}, } class FakePostgresClient: def __init__(self) -> None: self.session = make_review_session() self.created_review_payload: dict[str, Any] | None = None self.created_findings_payloads: list[dict[str, Any]] = [] self.created_assignments_payloads: list[dict[str, Any]] = [] self.submitted_supplements_payloads: list[dict[str, Any]] = [] self.verified_findings_payloads: list[dict[str, Any]] = [] self.closed_review_payloads: list[dict[str, Any]] = [] self.upserted_sessions: list[dict[str, Any]] = [] self.appended_actions: list[dict[str, Any]] = [] self.transaction_calls = 0 self.transaction_connection_tokens: list[str] = [] self.transaction_snapshots: list[dict[str, Any]] = [] self.received_connections: dict[str, list[Any]] = { "create_dossier_review_session": [], "create_dossier_review_finding": [], "assign_dossier_review_finding": [], "submit_dossier_review_supplement": [], "verify_dossier_review_finding": [], "close_dossier_review_session": [], } def is_enabled(self) -> bool: return False # Giả lập rollback để test giao dịch. def run_in_transaction(self, operation: Any) -> Any: self.transaction_calls += 1 connection_token = f"tx-conn-{self.transaction_calls}" self.transaction_connection_tokens.append(connection_token) snapshot = { "session": deepcopy(self.session), "created_review_payload": deepcopy(self.created_review_payload), "created_findings_payloads": deepcopy(self.created_findings_payloads), "created_assignments_payloads": deepcopy(self.created_assignments_payloads), "submitted_supplements_payloads": deepcopy( self.submitted_supplements_payloads ), "verified_findings_payloads": deepcopy(self.verified_findings_payloads), "closed_review_payloads": deepcopy(self.closed_review_payloads), } self.transaction_snapshots.append(snapshot) try: return operation(connection_token) except Exception: self.session = snapshot["session"] self.created_review_payload = snapshot["created_review_payload"] self.created_findings_payloads = snapshot["created_findings_payloads"] self.created_assignments_payloads = snapshot["created_assignments_payloads"] self.submitted_supplements_payloads = snapshot[ "submitted_supplements_payloads" ] self.verified_findings_payloads = snapshot["verified_findings_payloads"] self.closed_review_payloads = snapshot["closed_review_payloads"] raise def create_dossier_review_session( self, payload: dict[str, Any], connection: Any | None = None ) -> dict[str, Any]: self.received_connections["create_dossier_review_session"].append(connection) self.created_review_payload = deepcopy(payload) self.session["review_code"] = payload["review_code"] self.session["status"] = payload["status"] self.session["assigned_departments"] = list( payload.get("assigned_departments") or [] ) self.session["metadata"] = deepcopy(payload.get("metadata") or {}) return deepcopy(self.session) def create_dossier_review_finding( self, review_id: str, payload: dict[str, Any], connection: Any | None = None ) -> dict[str, Any]: self.received_connections["create_dossier_review_finding"].append(connection) finding_id = f"finding-{len(self.session['findings']) + 1}" finding = { "finding_id": finding_id, "finding_code": payload["finding_code"], "finding_group": payload["finding_group"], "finding_type": payload["finding_type"], "title": payload["title"], "description": payload.get("description"), "document_type": payload.get("document_type"), "dossier_stage": payload.get("dossier_stage"), "severity": payload.get("severity", "medium"), "impact_level": payload.get("impact_level", "medium"), "responsible_department_code": payload.get( "responsible_department_code", "synthesis" ), "supplement_status": payload.get("supplement_status", "requested"), "status": payload.get("status", "open"), "is_blocking": payload.get("is_blocking", False), "due_date": payload.get("due_date"), "detected_at": "2026-01-01T00:00:01Z", "resolved_at": None, "resolution_note": None, "evidences": deepcopy(payload.get("evidences") or []), "assignments": [], "metadata": deepcopy(payload.get("metadata") or {}), } self.created_findings_payloads.append(deepcopy(payload)) self.session["findings"].append(finding) self.session["open_findings_count"] = sum( 1 for item in self.session["findings"] if item["status"] != "resolved" ) return deepcopy(finding) def assign_dossier_review_finding( self, review_id: str, finding_id: str, payload: dict[str, Any], connection: Any | None = None, ) -> dict[str, Any]: self.received_connections["assign_dossier_review_finding"].append(connection) assignment_id = f"assignment-{len(self.created_assignments_payloads) + 1}" assignment = { "assignment_id": assignment_id, "assigned_department_code": payload["assigned_department_code"], "assigned_employee_code": payload.get("assigned_employee_code"), "assigned_employee_name": payload.get("assigned_employee_name"), "assigned_by_employee_code": payload.get("assigned_by_employee_code"), "assigned_by_employee_name": payload.get("assigned_by_employee_name"), "status": payload.get("status", "pending"), "priority": payload.get("priority", "medium"), "due_date": payload.get("due_date"), "assignment_note": payload.get("assignment_note"), "latest_submission_note": None, "accepted_at": None, "submitted_at": None, "verified_at": None, "closed_at": None, "submissions": [], "metadata": deepcopy(payload.get("metadata") or {}), } self.created_assignments_payloads.append(deepcopy(payload)) for finding in self.session["findings"]: if finding["finding_id"] == finding_id: finding["assignments"].append(assignment) finding["status"] = payload.get("finding_status", "assigned") break self.session["assigned_departments"] = sorted( { finding["responsible_department_code"] for finding in self.session["findings"] } ) return deepcopy(assignment) def submit_dossier_review_supplement( self, review_id: str, finding_id: str, payload: dict[str, Any], connection: Any | None = None, ) -> dict[str, Any]: self.received_connections["submit_dossier_review_supplement"].append(connection) self.submitted_supplements_payloads.append( { "review_id": review_id, "finding_id": finding_id, "payload": deepcopy(payload), } ) for finding in self.session["findings"]: if finding["finding_id"] == finding_id: finding["supplement_status"] = "submitted" break return { "submission_id": f"submission-{len(self.submitted_supplements_payloads)}", "finding_id": finding_id, "status": "submitted", } def verify_dossier_review_finding( self, review_id: str, finding_id: str, payload: dict[str, Any], connection: Any | None = None, ) -> dict[str, Any]: self.received_connections["verify_dossier_review_finding"].append(connection) self.verified_findings_payloads.append( { "review_id": review_id, "finding_id": finding_id, "payload": deepcopy(payload), } ) for finding in self.session["findings"]: if finding["finding_id"] == finding_id: finding["status"] = "resolved" finding["supplement_status"] = "accepted" finding["resolved_at"] = "2026-01-01T00:00:03Z" finding["resolution_note"] = payload.get("resolution_note") break self.session["resolved_findings_count"] = sum( 1 for item in self.session["findings"] if item["status"] == "resolved" ) self.session["open_findings_count"] = sum( 1 for item in self.session["findings"] if item["status"] != "resolved" ) return {"finding_id": finding_id, "status": "resolved"} def close_dossier_review_session( self, review_id: str, payload: dict[str, Any], connection: Any | None = None ) -> dict[str, Any]: self.received_connections["close_dossier_review_session"].append(connection) self.closed_review_payloads.append( {"review_id": review_id, "payload": deepcopy(payload)} ) self.session["status"] = "closed" self.session["closed_at"] = "2026-01-01T00:00:04Z" return deepcopy(self.session) def upsert_employee_work_log_session( self, payload: dict[str, Any] ) -> dict[str, Any]: self.upserted_sessions.append(deepcopy(payload)) return { "work_log_session_id": f"work-log-{len(self.upserted_sessions)}", "status": payload["status"], "project_code": payload["project_code"], "shift_date": payload["shift_date"], } def append_employee_work_log_action( self, payload: dict[str, Any] ) -> dict[str, Any]: self.appended_actions.append(deepcopy(payload)) return { "action_id": f"work-log-action-{len(self.appended_actions)}", "work_log_session_id": payload["work_log_session_id"], } def fetch_dossier_review_sessions( self, review_id: str | None = None, project_code: str | None = None ) -> list[dict[str, Any]]: session = deepcopy(self.session) if session["findings"] and session["status"] != "closed": session["status"] = "assigned" session["open_findings_count"] = sum( 1 for item in session["findings"] if item["status"] != "resolved" ) session["actions"].append( { "action_id": "action-finding-created", "action_type": "finding_created", "actor_type": "agent", "actor_employee_code": "EMP-LEAD", "actor_employee_name": "Trưởng nhóm", "actor_agent_code": "legal_agent", "action_summary": "Đã ghi nhận finding", "before_state": {}, "after_state": {"status": session["status"]}, "action_payload": {}, "created_at": "2026-01-01T00:00:02Z", } ) return [session] class WorkflowPersistenceServiceTestBase(unittest.TestCase): @staticmethod def mock_persona_payload() -> dict[str, Any]: return { "display_name": "AI Pháp chế", "role_title": "Chuyên viên pháp chế AI", "agent_code": "legal_agent", } @staticmethod def mock_policy_persona_payload() -> dict[str, Any]: return { "display_name": "AI Pháp chế", "role_title": "Chuyên viên pháp chế AI", "agent_code": "legal_agent", "escalation_targets": [], } @staticmethod def build_minimal_review_start_request() -> WorkflowReviewStartRequest: return 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", "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ý", "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", "metadata": {"action_type": "session_created"}, } ], } ) def assert_failed_transaction_leaves_no_post_commit_side_effects( self, *, failure_step: str ) -> None: class RollbackAtStepFakePostgresClient(FakePostgresClient): def __init__(self, target_step: str) -> None: super().__init__() self.target_step = target_step self.call_sequence: list[str] = [] def _maybe_fail(self, step: str, message: str) -> None: self.call_sequence.append(step) if self.target_step == step: raise RuntimeError(message) def create_dossier_review_session( self, payload: dict[str, Any], connection: Any | None = None ) -> dict[str, Any]: self._maybe_fail("create_session", "create-session-failed") return super().create_dossier_review_session( payload, connection=connection ) def create_dossier_review_finding( self, review_id: str, payload: dict[str, Any], connection: Any | None = None, ) -> dict[str, Any]: self._maybe_fail( f"create_finding:{payload['finding_code']}", "create-finding-failed" ) return super().create_dossier_review_finding( review_id, payload, connection=connection ) def assign_dossier_review_finding( self, review_id: str, finding_id: str, payload: dict[str, Any], connection: Any | None = None, ) -> dict[str, Any]: self._maybe_fail( f"assign_finding:{finding_id}", "assign-finding-failed" ) return super().assign_dossier_review_finding( review_id, finding_id, payload, connection=connection ) def submit_dossier_review_supplement( self, review_id: str, finding_id: str, payload: dict[str, Any], connection: Any | None = None, ) -> dict[str, Any]: self._maybe_fail( f"submit_finding:{finding_id}", "submit-finding-failed" ) return super().submit_dossier_review_supplement( review_id, finding_id, payload, connection=connection ) def verify_dossier_review_finding( self, review_id: str, finding_id: str, payload: dict[str, Any], connection: Any | None = None, ) -> dict[str, Any]: self._maybe_fail( f"verify_finding:{finding_id}", "verify-finding-failed" ) return super().verify_dossier_review_finding( review_id, finding_id, payload, connection=connection ) def close_dossier_review_session( self, review_id: str, payload: dict[str, Any], connection: Any | None = None, ) -> dict[str, Any]: self._maybe_fail("close_session", "close-session-failed") return super().close_dossier_review_session( review_id, payload, connection=connection ) def upsert_employee_work_log_session( self, payload: dict[str, Any] ) -> dict[str, Any]: self.call_sequence.append( f"upsert_work_log_session:{payload['employee_code']}" ) return super().upsert_employee_work_log_session(payload) def append_employee_work_log_action( self, payload: dict[str, Any] ) -> dict[str, Any]: self.call_sequence.append( f"append_work_log_action:{payload['work_log_session_id']}" ) return super().append_employee_work_log_action(payload) def fetch_dossier_review_sessions( self, review_id: str | None = None, project_code: str | None = None ) -> list[dict[str, Any]]: self.call_sequence.append("fetch_review_session") return super().fetch_dossier_review_sessions(review_id, project_code) fake_postgres = RollbackAtStepFakePostgresClient(failure_step) service = WorkflowPersistenceService(postgres_client=fake_postgres) with ( patch( "app.modules.core.application.workflow_persistence_service.get_persona" ) as mock_get_persona, patch( "app.modules.core.application.workflow_policy_service.get_persona" ) as mock_policy_get_persona, ): mock_get_persona.return_value = self.mock_persona_payload() mock_policy_get_persona.return_value = self.mock_policy_persona_payload() with self.assertRaisesRegex(RuntimeError, "-failed"): service.start_review_run(self.build_minimal_review_start_request()) self.assertEqual(fake_postgres.transaction_calls, 1) self.assertEqual(fake_postgres.session, make_review_session()) self.assertEqual(fake_postgres.created_findings_payloads, []) self.assertEqual(fake_postgres.created_assignments_payloads, []) self.assertEqual(fake_postgres.submitted_supplements_payloads, []) self.assertEqual(fake_postgres.verified_findings_payloads, []) self.assertEqual(fake_postgres.closed_review_payloads, []) self.assertEqual(fake_postgres.upserted_sessions, []) self.assertEqual(fake_postgres.appended_actions, []) self.assertNotIn("fetch_review_session", fake_postgres.call_sequence) self.assertFalse( any( item.startswith("upsert_work_log_session:") for item in fake_postgres.call_sequence ) ) self.assertFalse( any( item.startswith("append_work_log_action:") for item in fake_postgres.call_sequence ) ) def assert_review_code_pattern(test_case: unittest.TestCase, value: str) -> None: test_case.assertTrue(re.match(r"^RVW-PRJ-001-LEGAL_AGENT-\d{14}$", value))