from __future__ import annotations import unittest from copy import deepcopy from typing import Any from unittest.mock import patch from fastapi.testclient import TestClient from app.modules.auth.presentation.auth import get_current_user from app.main import create_app from app.models.schemas import WorkflowReviewRunSummary def make_session() -> dict[str, Any]: return { "review_id": "review-1", "review_code": "RVW-PRJ-001-LEGAL_AGENT-20260101000000", "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 hồ sơ", "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": [], "actions": [ { "action_id": "action-1", "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": "Khởi tạo phiên", "before_state": {}, "after_state": {"status": "in_review"}, "action_payload": {}, "created_at": "2026-01-01T00:00:00Z", } ], "metadata": {"workflow_type": "proactive_dossier_review"}, } class FakePostgresRoutesClient: def __init__(self) -> None: self.session = make_session() self.assignment_counter = 0 self.submission_counter = 0 def create_dossier_review_session(self, payload: dict[str, Any]) -> dict[str, Any]: self.session["review_code"] = payload.get( "review_code", self.session["review_code"] ) self.session["status"] = payload.get("status", "draft") return deepcopy(self.session) def fetch_dossier_review_sessions( self, project_code: str | None = None, review_id: str | None = None ) -> list[dict[str, Any]]: return [deepcopy(self.session)] def create_dossier_review_finding( self, review_id: str, payload: dict[str, Any] ) -> dict[str, Any]: finding = { "finding_id": "finding-1", "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", "legal" ), "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": [], "assignments": [], "metadata": payload.get("metadata") or {}, } self.session["findings"] = [finding] self.session["open_findings_count"] = 1 self.session["status"] = "awaiting_assignment" self.session["actions"].append( { "action_id": "action-2", "action_type": "finding_created", "actor_type": "employee", "actor_employee_code": "EMP-LEAD", "actor_employee_name": "Trưởng nhóm", "actor_agent_code": None, "action_summary": "Thêm finding", "before_state": {}, "after_state": {"status": "awaiting_assignment"}, "action_payload": {}, "created_at": "2026-01-01T00:00:01Z", } ) return deepcopy(finding) def assign_dossier_review_finding( self, review_id: str, finding_id: str, payload: dict[str, Any] ) -> dict[str, Any]: self.assignment_counter += 1 assignment = { "assignment_id": f"assignment-{self.assignment_counter}", "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": payload.get("metadata") or {}, } finding = self.session["findings"][0] finding["assignments"].append(assignment) finding["status"] = "assigned" finding["supplement_status"] = "requested" self.session["status"] = "assigned" self.session["assigned_departments"] = [payload["assigned_department_code"]] return deepcopy(assignment) def submit_dossier_review_supplement( self, review_id: str, finding_id: str, payload: dict[str, Any] ) -> dict[str, Any]: self.submission_counter += 1 submission = { "submission_id": f"submission-{self.submission_counter}", "submission_type": payload.get("submission_type", "document_bundle"), "submitted_by_employee_code": payload.get("submitted_by_employee_code"), "submitted_by_employee_name": "Người nộp", "submission_note": payload.get("submission_note"), "attachment_count": payload.get("attachment_count", 0), "submitted_payload": payload.get("submitted_payload") or {}, "verification_result": None, "verification_note": None, "verified_by_employee_code": None, "verified_by_employee_name": None, "verified_at": None, "created_at": "2026-01-01T00:00:02Z", } assignment = self.session["findings"][0]["assignments"][0] assignment["submissions"].append(submission) assignment["status"] = "submitted" assignment["submitted_at"] = "2026-01-01T00:00:02Z" assignment["latest_submission_note"] = payload.get("submission_note") finding = self.session["findings"][0] finding["status"] = "pending_verification" finding["supplement_status"] = "submitted" return deepcopy(submission) def verify_dossier_review_finding( self, review_id: str, finding_id: str, payload: dict[str, Any] ) -> dict[str, Any]: assignment = self.session["findings"][0]["assignments"][0] submission = assignment["submissions"][0] submission["verification_result"] = payload.get( "verification_result", "accepted" ) submission["verification_note"] = payload.get("verification_note") submission["verified_by_employee_code"] = payload.get( "verified_by_employee_code" ) submission["verified_by_employee_name"] = "Người duyệt" submission["verified_at"] = "2026-01-01T00:00:03Z" assignment["status"] = "verified" assignment["verified_at"] = "2026-01-01T00:00:03Z" assignment["closed_at"] = "2026-01-01T00:00:03Z" finding = self.session["findings"][0] finding["status"] = "resolved" finding["supplement_status"] = "accepted" finding["resolved_at"] = "2026-01-01T00:00:03Z" finding["resolution_note"] = payload.get("verification_note") self.session["open_findings_count"] = 0 self.session["resolved_findings_count"] = 1 self.session["status"] = "resolved" self.session["actions"].append( { "action_id": "action-verify", "action_type": "verification_completed", "actor_type": "employee", "actor_employee_code": payload.get("verified_by_employee_code"), "actor_employee_name": "Người duyệt", "actor_agent_code": None, "action_summary": "Xác minh finding", "before_state": {}, "after_state": {"status": "resolved"}, "action_payload": {}, "created_at": "2026-01-01T00:00:03Z", } ) return deepcopy(finding) def close_dossier_review_session( self, review_id: str, payload: dict[str, Any] ) -> dict[str, Any]: self.session["status"] = payload.get("status", "closed") self.session["review_summary"] = payload.get("review_summary") self.session["closed_at"] = "2026-01-01T00:00:04Z" self.session["actions"].append( { "action_id": "action-close", "action_type": "session_closed", "actor_type": "employee", "actor_employee_code": payload.get("actor_employee_code"), "actor_employee_name": "Người đóng", "actor_agent_code": None, "action_summary": payload.get("action_summary", "Đóng phiên"), "before_state": {}, "after_state": {"status": self.session["status"]}, "action_payload": {}, "created_at": "2026-01-01T00:00:04Z", } ) return deepcopy(self.session) class FakeWorkflowPersistenceService: def start_review_run(self, payload) -> WorkflowReviewRunSummary: return WorkflowReviewRunSummary.model_validate( { "review": make_session(), "finding_summaries": [], "employee_log_summaries": [], "created_snapshot_ids": ["snapshot-1"], "review_action_count": 1, "message": "Đã lưu workflow rà soát hồ sơ chủ động và nhật ký tham gia của nhân sự AI.", } ) class DossierReviewRoutesTests(unittest.TestCase): def setUp(self) -> None: self.app = create_app() self.app.dependency_overrides[get_current_user] = lambda: { "id": "user-1", "email": "test@dscons.vn", "role": "superadmin", } self.client = TestClient(self.app) @patch( "app.api.routes.WorkflowPersistenceService", return_value=FakeWorkflowPersistenceService(), ) def test_start_workflow_route_returns_structured_summary( self, _mock_service ) -> None: response = self.client.post( "/v1/workflows/dossier-review/start", json={ "project_code": "PRJ-001", "lead_agent_code": "legal_agent", "lead_reviewer_employee_code": "EMP-LEAD", }, ) self.assertEqual(response.status_code, 200) payload = response.json() self.assertEqual(payload["review"]["review_id"], "review-1") self.assertEqual(payload["created_snapshot_ids"], ["snapshot-1"]) self.assertEqual( payload["message"], "Đã lưu workflow rà soát hồ sơ chủ động và nhật ký tham gia của nhân sự AI.", ) @patch("app.api.routes.PostgresClient", return_value=FakePostgresRoutesClient()) def test_route_level_review_lifecycle(self, _mock_postgres) -> None: create_response = self.client.post( "/v1/dossiers/reviews", json={ "project_code": "PRJ-001", "project_name": "Dự án thử nghiệm", "dossier_scope": "project_dossier", "trigger_source": "manual", "lead_agent_code": "legal_agent", "status": "draft", }, ) self.assertEqual(create_response.status_code, 200) self.assertEqual(create_response.json()["status"], "draft") finding_response = self.client.post( "/v1/dossiers/reviews/review-1/findings", json={ "finding_code": "F-001", "finding_group": "legal", "finding_type": "missing_document", "title": "Thiếu quyết định phê duyệt", "responsible_department_code": "legal", "supplement_status": "requested", "status": "open", "severity": "high", "impact_level": "high", }, ) self.assertEqual(finding_response.status_code, 200) finding_payload = finding_response.json() self.assertEqual(finding_payload["status"], "awaiting_assignment") self.assertEqual(finding_payload["open_findings_count"], 1) self.assertEqual(finding_payload["findings"][0]["status"], "open") assignment_response = self.client.post( "/v1/dossiers/reviews/review-1/findings/finding-1/assignments", json={ "assigned_department_code": "legal", "assigned_employee_code": "EMP-LAW-01", "assigned_by_employee_code": "EMP-LEAD", "priority": "high", "status": "pending", "due_date": "2026-01-10", }, ) self.assertEqual(assignment_response.status_code, 200) assignment_payload = assignment_response.json() self.assertEqual(assignment_payload["status"], "assigned") self.assertEqual(assignment_payload["assigned_departments"], ["legal"]) self.assertEqual( assignment_payload["findings"][0]["assignments"][0]["status"], "pending" ) submission_response = self.client.post( "/v1/dossiers/reviews/review-1/findings/finding-1/assignments/assignment-1/submissions", json={ "submitted_by_employee_code": "EMP-LAW-01", "submission_type": "document_bundle", "submission_note": "Đã bổ sung hồ sơ", "attachment_count": 2, "submitted_payload": {"files": ["qd.pdf"]}, }, ) self.assertEqual(submission_response.status_code, 200) submission_payload = submission_response.json() self.assertEqual( submission_payload["findings"][0]["status"], "pending_verification" ) self.assertEqual( submission_payload["findings"][0]["supplement_status"], "submitted" ) self.assertEqual( submission_payload["findings"][0]["assignments"][0]["status"], "submitted" ) verify_response = self.client.post( "/v1/dossiers/reviews/review-1/findings/finding-1/assignments/assignment-1/verify", json={ "assignment_id": "assignment-1", "submission_id": "submission-1", "verified_by_employee_code": "EMP-QA-01", "verification_result": "accepted", "verification_note": "Hồ sơ đạt yêu cầu", }, ) self.assertEqual(verify_response.status_code, 200) verify_payload = verify_response.json() self.assertEqual(verify_payload["status"], "resolved") self.assertEqual(verify_payload["open_findings_count"], 0) self.assertEqual(verify_payload["resolved_findings_count"], 1) self.assertEqual(verify_payload["findings"][0]["status"], "resolved") self.assertEqual(verify_payload["findings"][0]["supplement_status"], "accepted") self.assertEqual( verify_payload["findings"][0]["assignments"][0]["status"], "verified" ) self.assertEqual( verify_payload["actions"][-1]["action_type"], "verification_completed" ) close_response = self.client.post( "/v1/dossiers/reviews/review-1/close", json={ "closure_note": "Đã xử lý xong toàn bộ finding", }, ) self.assertEqual(close_response.status_code, 200) close_payload = close_response.json() self.assertEqual(close_payload["status"], "closed") self.assertEqual(close_payload["closed_at"], "2026-01-01T00:00:04Z") self.assertEqual(close_payload["actions"][-1]["action_type"], "session_closed") @patch("app.api.routes.PostgresClient", return_value=FakePostgresRoutesClient()) def test_list_reviews_route_returns_schema_compatible_payload( self, _mock_postgres ) -> None: response = self.client.get("/v1/dossiers/reviews") self.assertEqual(response.status_code, 200) payload = response.json() self.assertEqual(payload["source"], "postgres") self.assertFalse(payload["used_fallback"]) self.assertEqual(len(payload["items"]), 1) self.assertEqual( payload["items"][0]["review_code"], "RVW-PRJ-001-LEGAL_AGENT-20260101000000" ) if __name__ == "__main__": unittest.main()