feat: implement final approval gate workflow and checklist

2026-05-22 00:31:48 +03:00
parent 6dcbe0e36c
commit d3990b4019
15 changed files with 1851 additions and 14 deletions
@@ -0,0 +1,427 @@
+from __future__ import annotations
+
+import base64
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from typing import Any
+from uuid import UUID
+
+from fastapi.testclient import TestClient
+
+
+BACKEND_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(BACKEND_ROOT))
+
+from src.application.seed_data import seed_reference_data  # noqa: E402
+from src.infrastructure.repositories import open_backend_repository  # noqa: E402
+from src.presentation.dependencies import get_repository  # noqa: E402
+from src.presentation.main import app  # noqa: E402
+
+
+DEMO_EDITOR_EMAIL = "editor@example.com"
+DEMO_ADMIN_EMAIL = "admin@example.com"
+DEMO_USER_EMAIL_HEADER = "X-Demo-User-Email"
+
+
+class FinalApprovalGatePublicApiTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self.tmp_dir = tempfile.TemporaryDirectory()
+        os.environ["OBJECT_STORAGE_LOCAL_ROOT"] = str(Path(self.tmp_dir.name) / "objects")
+        dsn = f"sqlite:///{Path(self.tmp_dir.name) / 'final-approval-gate.db'}"
+        self.repository = open_backend_repository(dsn)
+        self.repository.setup()
+        seed_reference_data(self.repository)
+        app.dependency_overrides[get_repository] = lambda: self.repository
+        self.client = TestClient(app)
+
+    def tearDown(self) -> None:
+        app.dependency_overrides.clear()
+        os.environ.pop("OBJECT_STORAGE_LOCAL_ROOT", None)
+        self.tmp_dir.cleanup()
+
+    def test_high_risk_unsupported_claims_block_final_approval(self) -> None:
+        article_id, draft = self._prepare_article_ready_for_final_approval(
+            include_high_risk_unsupported=True,
+        )
+
+        response = self.client.post(
+            f"/api/articles/{article_id}/final-approval",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+            json=self._final_approval_payload(draft_version=draft["version"]),
+        )
+        self.assertEqual(409, response.status_code, response.text)
+        self.assertIn("high-risk unsupported claims", response.text.lower())
+
+    def test_final_approval_requires_exact_draft_version(self) -> None:
+        article_id, draft = self._prepare_article_ready_for_final_approval()
+
+        response = self.client.post(
+            f"/api/articles/{article_id}/final-approval",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+            json=self._final_approval_payload(draft_version=draft["version"] + 1),
+        )
+        self.assertEqual(409, response.status_code, response.text)
+        self.assertIn("latest draft version", response.text.lower())
+
+    def test_missing_required_publishing_settings_block_final_approval(self) -> None:
+        article_id, draft = self._prepare_article_ready_for_final_approval()
+
+        response = self.client.post(
+            f"/api/articles/{article_id}/final-approval",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+            json={
+                "draft_version": draft["version"],
+                "publishing_settings": {
+                    "content_path": "",
+                    "author": "",
+                    "publishing_mode": "",
+                    "frontmatter": {},
+                },
+            },
+        )
+        self.assertEqual(409, response.status_code, response.text)
+        self.assertIn("required publishing settings are missing", response.text.lower())
+
+    def test_unapproved_required_assets_block_final_approval(self) -> None:
+        article_id, draft = self._prepare_article_ready_for_final_approval(approve_assets=False)
+
+        response = self.client.post(
+            f"/api/articles/{article_id}/final-approval",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+            json=self._final_approval_payload(draft_version=draft["version"]),
+        )
+        self.assertEqual(409, response.status_code, response.text)
+        self.assertIn("required assets", response.text.lower())
+
+    def test_successful_final_approval_writes_event_and_moves_to_publish_dry_run_required(self) -> None:
+        article_id, draft = self._prepare_article_ready_for_final_approval()
+        actor = self.client.get(
+            "/api/me",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        ).json()["user"]
+        payload = self._final_approval_payload(draft_version=draft["version"])
+
+        response = self.client.post(
+            f"/api/articles/{article_id}/final-approval",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+            json=payload,
+        )
+        self.assertEqual(200, response.status_code, response.text)
+        body = response.json()
+        self.assertEqual("PUBLISH_DRY_RUN_REQUIRED", body["article"]["status"])
+        self.assertTrue(body["checklist"]["can_approve"])
+        self.assertFalse(body["checklist"]["blockers"])
+
+        detail = self.client.get(
+            f"/api/articles/{article_id}",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        ).json()
+        events = [
+            event
+            for event in detail["workflow_events"]
+            if event["event_type"] == "FINAL_APPROVAL_GRANTED"
+        ]
+        self.assertTrue(events, "Expected FINAL_APPROVAL_GRANTED event")
+        latest = events[-1]
+        self.assertEqual(actor["id"], latest["actor_user_id"])
+        self.assertTrue(latest["created_at"])
+        self.assertEqual(draft["version"], latest["payload"]["draft_version"])
+        self.assertEqual(payload["publishing_settings"], latest["payload"]["publishing_settings"])
+
+    def test_revision_request_writes_event_and_moves_article_to_final_revision_required(self) -> None:
+        article_id, draft = self._prepare_article_ready_for_final_approval()
+        actor = self.client.get(
+            "/api/me",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        ).json()["user"]
+
+        response = self.client.post(
+            f"/api/articles/{article_id}/final-revision-request",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+            json={
+                "reason": "Need factual clarification in section 2.",
+                "draft_version": draft["version"],
+            },
+        )
+        self.assertEqual(200, response.status_code, response.text)
+        body = response.json()
+        self.assertEqual("FINAL_REVISION_REQUIRED", body["article"]["status"])
+
+        detail = self.client.get(
+            f"/api/articles/{article_id}",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        ).json()
+        events = [
+            event
+            for event in detail["workflow_events"]
+            if event["event_type"] == "FINAL_REVISION_REQUESTED"
+        ]
+        self.assertTrue(events, "Expected FINAL_REVISION_REQUESTED event")
+        latest = events[-1]
+        self.assertEqual(actor["id"], latest["actor_user_id"])
+        self.assertEqual("Need factual clarification in section 2.", latest["payload"]["reason"])
+        self.assertEqual(draft["version"], latest["payload"]["draft_version"])
+
+    def test_checklist_endpoint_returns_blockers_and_completed_items(self) -> None:
+        article_id, _ = self._prepare_article_ready_for_final_approval(approve_assets=False)
+
+        response = self.client.get(
+            f"/api/articles/{article_id}/final-approval/checklist",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(200, response.status_code, response.text)
+        checklist = response.json()
+        self.assertFalse(checklist["can_approve"])
+        self.assertTrue(checklist["blockers"])
+        self.assertTrue(checklist["completed_items"])
+        self.assertTrue(any(item["completed"] for item in checklist["items"]))
+        self.assertTrue(any(not item["completed"] for item in checklist["items"]))
+
+    def _final_approval_payload(self, *, draft_version: int) -> dict[str, Any]:
+        return {
+            "draft_version": draft_version,
+            "publishing_settings": {
+                "content_path": "/guides/final-approval-gate",
+                "author": "Editorial Team",
+                "publishing_mode": "MANUAL",
+                "frontmatter": {
+                    "category": "Guides",
+                    "tags": ["workflow", "approval"],
+                },
+            },
+        }
+
+    def _prepare_article_ready_for_final_approval(
+        self,
+        *,
+        include_high_risk_unsupported: bool = False,
+        approve_assets: bool = True,
+    ) -> tuple[str, dict[str, Any]]:
+        article_id = self._create_article_with_approved_plan()
+        self._ensure_evidence_ready(article_id)
+        self._approve_all_evidence(article_id)
+
+        section_jobs = self._start_parallel_production(article_id)
+        self.assertGreaterEqual(len(section_jobs), 1)
+        first_job = section_jobs[0]
+
+        if include_high_risk_unsupported:
+            failed = self._complete_job(
+                first_job["id"],
+                output={
+                    "status": "SUCCEEDED",
+                    "output_files": [{"path": "outputs/high-risk.md"}],
+                    "payload": {
+                        "used_evidence_ids": first_job["payload"]["used_evidence_ids"],
+                        "unsupported_claims": [
+                            {
+                                "claim_text": "Unverified migration benchmark for market leadership.",
+                                "risk_level": "high",
+                            }
+                        ],
+                        "draft_markdown": "Section draft with high-risk unsupported claim.",
+                    },
+                },
+            )
+            self.assertEqual("FAILED", failed["status"])
+            retry_response = self.client.post(
+                f"/api/agent-jobs/{first_job['id']}/retry",
+                headers={DEMO_USER_EMAIL_HEADER: DEMO_ADMIN_EMAIL},
+            )
+            self.assertEqual(201, retry_response.status_code, retry_response.text)
+            retry_job = retry_response.json()["job"]
+            retried = self._complete_job(
+                retry_job["id"],
+                output={
+                    "status": "SUCCEEDED",
+                    "output_files": [{"path": "outputs/retry-clean.md"}],
+                    "payload": {
+                        "used_evidence_ids": retry_job["payload"]["used_evidence_ids"],
+                        "unsupported_claims": [],
+                        "draft_markdown": (
+                            f"## {retry_job['payload']['heading']}\n\n"
+                            "Clean retry section with [internal guide](/guides/internal)."
+                        ),
+                    },
+                },
+            )
+            self.assertEqual("SUCCEEDED", retried["status"])
+        else:
+            completed = self._complete_job(
+                first_job["id"],
+                output={
+                    "status": "SUCCEEDED",
+                    "output_files": [{"path": "outputs/section-1.md"}],
+                    "payload": {
+                        "used_evidence_ids": first_job["payload"]["used_evidence_ids"],
+                        "unsupported_claims": [],
+                        "draft_markdown": (
+                            f"## {first_job['payload']['heading']}\n\n"
+                            "Use [internal guide](/guides/internal) for implementation details."
+                        ),
+                    },
+                },
+            )
+            self.assertEqual("SUCCEEDED", completed["status"])
+
+        for index, job in enumerate(section_jobs[1:], start=2):
+            completed = self._complete_job(
+                job["id"],
+                output={
+                    "status": "SUCCEEDED",
+                    "output_files": [{"path": f"outputs/section-{index}.md"}],
+                    "payload": {
+                        "used_evidence_ids": job["payload"]["used_evidence_ids"],
+                        "unsupported_claims": [],
+                        "draft_markdown": f"## {job['payload']['heading']}\n\nSection {index} ready.",
+                    },
+                },
+            )
+            self.assertEqual("SUCCEEDED", completed["status"])
+
+        assemble_response = self.client.post(
+            f"/api/articles/{article_id}/draft/assemble",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(201, assemble_response.status_code, assemble_response.text)
+        draft = assemble_response.json()["draft"]
+
+        assets_response = self.client.post(
+            f"/api/articles/{article_id}/assets/generate-specs",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(201, assets_response.status_code, assets_response.text)
+        assets = assets_response.json()["assets"]
+        self.assertTrue(assets)
+
+        if approve_assets:
+            for asset in assets:
+                upload_response = self.client.post(
+                    f"/api/articles/{article_id}/assets/{asset['id']}/upload",
+                    headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+                    json={
+                        "filename": f"{asset['id']}.png",
+                        "content_base64": base64.b64encode(b"asset-binary").decode("utf-8"),
+                        "content_type": "image/png",
+                    },
+                )
+                self.assertEqual(200, upload_response.status_code, upload_response.text)
+                approve_response = self.client.post(
+                    f"/api/articles/{article_id}/assets/{asset['id']}/approve",
+                    headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+                )
+                self.assertEqual(200, approve_response.status_code, approve_response.text)
+
+        return article_id, draft
+
+    def _approve_all_evidence(self, article_id: str) -> None:
+        evidence_response = self.client.get(
+            f"/api/articles/{article_id}/evidence",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(200, evidence_response.status_code, evidence_response.text)
+        evidence = evidence_response.json()["evidence"]
+        self.assertTrue(evidence)
+        for item in evidence:
+            patch_response = self.client.patch(
+                f"/api/articles/{article_id}/evidence/{item['id']}",
+                headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+                json={"review_status": "APPROVED"},
+            )
+            self.assertEqual(200, patch_response.status_code, patch_response.text)
+
+    def _complete_job(self, job_id: str, *, output: dict[str, Any]) -> dict[str, Any]:
+        response = self.client.post(
+            f"/internal/agent-jobs/{job_id}/complete",
+            json={
+                "workspace_path": f"/tmp/{job_id}",
+                "stdout": "fake section scaffolding runner\n",
+                "stderr": "",
+                "exit_code": 0,
+                "duration_ms": 1,
+                "output": output,
+            },
+        )
+        self.assertEqual(200, response.status_code, response.text)
+        return response.json()["job"]
+
+    def _start_parallel_production(self, article_id: str) -> list[dict[str, Any]]:
+        response = self.client.post(
+            f"/api/articles/{article_id}/draft/start",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(202, response.status_code, response.text)
+        return [job for job in response.json()["jobs"] if job["job_type"] == "SECTION_SCAFFOLD"]
+
+    def _ensure_evidence_ready(self, article_id: str) -> None:
+        research_response = self.client.post(
+            f"/api/articles/{article_id}/research/start",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(201, research_response.status_code, research_response.text)
+
+        evidence_response = self.client.get(
+            f"/api/articles/{article_id}/evidence",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(200, evidence_response.status_code, evidence_response.text)
+        self.assertEqual("EVIDENCE_MATRIX_READY", evidence_response.json()["article"]["status"])
+
+    def _create_article_with_approved_plan(self) -> str:
+        site = self.client.get(
+            "/api/sites",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        ).json()[0]["site"]
+        article = self.client.post(
+            "/api/articles",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+            json={
+                "target_site_id": site["id"],
+                "brief_description": "Validate final approval gate checklist and transitions.",
+                "working_title": "Final Approval Gate",
+                "content_type": "longform_guide",
+                "primary_keyword": "final approval gate",
+            },
+        ).json()["article"]
+        article_id = article["id"]
+
+        questions = self.client.post(
+            f"/api/articles/{article_id}/boundary-questions/generate",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        ).json()["questions"]
+        for question in questions:
+            if question["is_required"]:
+                patch_response = self.client.patch(
+                    f"/api/articles/{article_id}/boundary-questions/{question['id']}",
+                    headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+                    json={"answer": f"Answer for {question['category']}"},
+                )
+                self.assertEqual(200, patch_response.status_code, patch_response.text)
+
+        submit_response = self.client.post(
+            f"/api/articles/{article_id}/boundary-questions/submit",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(200, submit_response.status_code, submit_response.text)
+
+        plan_response = self.client.post(
+            f"/api/articles/{article_id}/plan/generate",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(201, plan_response.status_code, plan_response.text)
+        plan = plan_response.json()["plan"]
+        self.assertGreaterEqual(len(plan["sections"]), 1)
+
+        approve_response = self.client.post(
+            f"/api/articles/{article_id}/plans/{plan['id']}/approve",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(200, approve_response.status_code, approve_response.text)
+        return article_id
+
+
+if __name__ == "__main__":
+    unittest.main()