huncode/content-factory
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Task 009 implement plan generation review gate

Browse Source
This commit is contained in:
huncode
2026-05-21 19:56:58 +03:00
parent b7e2fa34b2
commit 5b27d1a3ca
24 changed files with 2422 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/tests/integration/test_plan_generation_review_public_api.py
+174
View File
@@ -0,0 +1,174 @@
from __future__ import annotations
import sys
import tempfile
import unittest
from pathlib import Path
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_USER_EMAIL_HEADER = "X-Demo-User-Email"
class PlanGenerationReviewPublicApiTest(unittest.TestCase):
def setUp(self) -> None:
self.tmp_dir = tempfile.TemporaryDirectory()
dsn = f"sqlite:///{Path(self.tmp_dir.name) / 'plans.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()
self.tmp_dir.cleanup()
def test_research_start_is_blocked_until_plan_approval(self) -> None:
article_id = self._create_article_with_boundary_answers()
blocked_research_response = self.client.post(
f"/api/articles/{article_id}/research/start",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
self.assertEqual(409, blocked_research_response.status_code, blocked_research_response.text)
generate_response = self.client.post(
f"/api/articles/{article_id}/plan/generate",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
self.assertEqual(201, generate_response.status_code, generate_response.text)
plan = generate_response.json()["plan"]
self.assertEqual(1, plan["version"])
self.assertEqual("PENDING_REVIEW", plan["status"])
self.assertGreaterEqual(len(plan["sections"]), 4)
self.assertTrue(plan["recommended_title"])
self.assertTrue(plan["reader_persona"])
self.assertTrue(plan["evidence_needs"])
self.assertTrue(plan["visual_needs"])
for section in plan["sections"]:
self.assertTrue(section["purpose"])
self.assertTrue(section["key_points"])
self.assertTrue(section["evidence_needs"])
edit_response = self.client.patch(
f"/api/articles/{article_id}/plans/{plan['id']}",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
json={
"recommended_title": "Edited plan title",
"tone": "Direct and evidence-led",
"audience": "Technical editors",
"seo_notes": ["Prioritize comparison intent"],
"sections": [
{
"heading": "Edited section",
"purpose": "Set the context",
"key_points": ["Point A", "Point B"],
"evidence_needs": ["Vendor docs"],
"target_word_count": 300,
}
],
},
)
self.assertEqual(201, edit_response.status_code, edit_response.text)
edited_plan = edit_response.json()["plan"]
self.assertEqual(2, edited_plan["version"])
self.assertEqual("Edited plan title", edited_plan["recommended_title"])
list_response = self.client.get(
f"/api/articles/{article_id}/plans",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
self.assertEqual(200, list_response.status_code, list_response.text)
self.assertEqual([1, 2], [item["version"] for item in list_response.json()["plans"]])
revision_response = self.client.post(
f"/api/articles/{article_id}/plans/{edited_plan['id']}/request-revision",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
json={"notes": "Narrow the evidence standard."},
)
self.assertEqual(200, revision_response.status_code, revision_response.text)
self.assertEqual("CHANGES_REQUESTED", revision_response.json()["review"]["status"])
approve_response = self.client.post(
f"/api/articles/{article_id}/plans/{edited_plan['id']}/approve",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
self.assertEqual(200, approve_response.status_code, approve_response.text)
self.assertEqual("APPROVED", approve_response.json()["review"]["status"])
allowed_research_response = self.client.post(
f"/api/articles/{article_id}/research/start",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
self.assertEqual(202, allowed_research_response.status_code, allowed_research_response.text)
self.assertEqual("RESEARCH", allowed_research_response.json()["jobs"][0]["job_type"])
detail_response = self.client.get(
f"/api/articles/{article_id}",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
events = detail_response.json()["workflow_events"]
approval_events = [
event for event in events if event["event_type"] == "PLAN_APPROVED"
]
self.assertEqual(1, len(approval_events))
self.assertEqual(2, approval_events[0]["payload"]["plan_version"])
def _create_article_with_boundary_answers(self) -> str:
site_response = self.client.get(
"/api/sites",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
self.assertEqual(200, site_response.status_code, site_response.text)
target_site = site_response.json()[0]["site"]
create_response = self.client.post(
"/api/articles",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
json={
"target_site_id": target_site["id"],
"brief_description": "Create an implementation plan for AI content governance.",
"working_title": "AI Content Governance",
"content_type": "longform_guide",
"primary_keyword": "AI content governance",
},
)
self.assertEqual(201, create_response.status_code, create_response.text)
article_id = create_response.json()["article"]["id"]
questions_response = self.client.post(
f"/api/articles/{article_id}/boundary-questions/generate",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
self.assertEqual(201, questions_response.status_code, questions_response.text)
for question in questions_response.json()["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)
return article_id
if __name__ == "__main__":
unittest.main()
apps/backend/tests/integration/test_schema_storage_contracts.py
+18
View File
@@ -91,6 +91,24 @@ class SchemaStorageContractsIntegrationTest(unittest.TestCase):
"category",
repository.schema.list_columns("boundary_questions"),
)
self.assertTrue(
{
"title_options",
"reader_persona",
"evidence_needs",
"visual_needs",
"seo_notes",
"source_requirements",
"excluded_sources",
"tone",
"audience",
}.issubset(repository.schema.list_columns("article_plans"))
)
self.assertTrue(
{"key_points", "evidence_needs"}.issubset(
repository.schema.list_columns("plan_sections")
)
)
self.assertTrue(
{
"s3_prefix",