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

feat(task-023): bind workflow templates into demo flow

Browse Source
This commit is contained in:
huncode
2026-05-22 19:38:29 +03:00
parent baff9211e8
commit 5f3e543ea2
24 changed files with 2671 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/tests/integration/test_auth_authorization_public_api.py
+28 -1
View File
@@ -316,8 +316,35 @@ class AuthAuthorizationPublicApiTest(unittest.TestCase):
)
article_id = article_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)
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_id = plan_response.json()["plan"]["id"]
review_response = self.client.post(
f"/api/articles/{article_id}/plans/00000000-0000-0000-0000-000000000123/approve",
f"/api/articles/{article_id}/plans/{plan_id}/approve",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
apps/backend/tests/integration/test_workflow_template_binding_public_api.py
+288
View File
@@ -0,0 +1,288 @@
from __future__ import annotations
import sys
import tempfile
import unittest
from pathlib import Path
from typing import Any
from uuid import uuid4
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_ADMIN_EMAIL = "admin@example.com"
DEMO_EDITOR_EMAIL = "editor@example.com"
DEMO_USER_EMAIL_HEADER = "X-Demo-User-Email"
class WorkflowTemplateBindingPublicApiTest(unittest.TestCase):
def setUp(self) -> None:
self.tmp_dir = tempfile.TemporaryDirectory()
dsn = f"sqlite:///{Path(self.tmp_dir.name) / 'workflow-template-binding.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_article_creation_stores_active_workflow_template_snapshot(self) -> None:
workflow = self._create_active_workflow("article-binding-active")
target_site = self._first_target_site()
create_response = self.client.post(
"/api/articles",
headers=self._editor_headers(),
json={
"target_site_id": target_site["id"],
"workflow_template_id": workflow["id"],
"brief_description": (
"Create an article that is governed by an Admin-managed "
f"workflow template {uuid4()}."
),
"working_title": "Workflow-bound article",
"content_type": "longform_guide",
"primary_keyword": "workflow template binding",
},
)
self.assertEqual(201, create_response.status_code, create_response.text)
article = create_response.json()["article"]
self.assertEqual(workflow["id"], article["workflow_template_id"])
self.assertEqual(workflow["version"], article["workflow_template_version"])
archive_response = self.client.post(
f"/api/admin/workflows/{workflow['id']}/archive",
headers=self._admin_headers(),
)
self.assertEqual(200, archive_response.status_code, archive_response.text)
detail_response = self.client.get(
f"/api/articles/{article['id']}",
headers=self._editor_headers(),
)
self.assertEqual(200, detail_response.status_code, detail_response.text)
detail = detail_response.json()
expected_stage_summary = [
{
"id": stage["id"],
"stable_key": stage["stable_key"],
"display_name": stage["display_name"],
"position": stage["position"],
"owner_role": stage["owner_role"],
"runner_profile_key": stage["runner_profile_key"],
"requires_human_approval": stage["requires_human_approval"],
}
for stage in workflow["stages"]
]
self.assertEqual(
{
"id": workflow["id"],
"name": workflow["name"],
"slug": workflow["slug"],
"version": workflow["version"],
"stage_summary": expected_stage_summary,
},
detail["workflow_template_snapshot"],
)
def test_article_creation_rejects_archived_workflow_template_reference(self) -> None:
workflow = self._create_active_workflow("article-binding-archived")
archive_response = self.client.post(
f"/api/admin/workflows/{workflow['id']}/archive",
headers=self._admin_headers(),
)
self.assertEqual(200, archive_response.status_code, archive_response.text)
response = self._create_article_with_workflow_template(workflow["id"])
self.assertEqual(400, response.status_code, response.text)
self.assertEqual("Workflow template must be active", response.json()["detail"])
def test_article_creation_rejects_missing_workflow_template_reference(self) -> None:
response = self._create_article_with_workflow_template(str(uuid4()))
self.assertEqual(400, response.status_code, response.text)
self.assertEqual("Invalid workflow_template_id", response.json()["detail"])
def test_seeded_active_workflow_template_is_available_for_editor_articles(self) -> None:
workflows_response = self.client.get(
"/api/admin/workflows",
headers=self._editor_headers(),
)
self.assertEqual(200, workflows_response.status_code, workflows_response.text)
seeded_workflow = next(
workflow
for workflow in workflows_response.json()["workflows"]
if workflow["slug"] == "demo-article-production"
)
self.assertEqual("ACTIVE", seeded_workflow["status"])
self.assertGreaterEqual(len(seeded_workflow["stages"]), 7)
create_response = self._create_article_with_workflow_template(
seeded_workflow["id"]
)
self.assertEqual(201, create_response.status_code, create_response.text)
article = create_response.json()["article"]
self.assertEqual(seeded_workflow["id"], article["workflow_template_id"])
self.assertEqual(
seeded_workflow["version"],
article["workflow_template_version"],
)
def _create_article_with_workflow_template(self, workflow_template_id: str) -> Any:
target_site = self._first_target_site()
return self.client.post(
"/api/articles",
headers=self._editor_headers(),
json={
"target_site_id": target_site["id"],
"workflow_template_id": workflow_template_id,
"brief_description": (
"Create an article using the selected workflow template "
f"{uuid4()}."
),
"content_type": "longform_guide",
"primary_keyword": "workflow binding guardrail",
},
)
def _create_active_workflow(self, slug_prefix: str) -> dict[str, Any]:
create_response = self.client.post(
"/api/admin/workflows",
headers=self._admin_headers(),
json={
"name": "Article production workflow",
"slug": f"{slug_prefix}-{uuid4()}",
"description": "Reusable editorial workflow for article production.",
},
)
self.assertEqual(201, create_response.status_code, create_response.text)
workflow = create_response.json()["workflow"]
workflow_id = workflow["id"]
for stage_payload in (
self._intake_stage_payload(),
self._draft_stage_payload(),
):
stage_response = self.client.post(
f"/api/admin/workflows/{workflow_id}/stages",
headers=self._admin_headers(),
json=stage_payload,
)
self.assertEqual(201, stage_response.status_code, stage_response.text)
activate_response = self.client.post(
f"/api/admin/workflows/{workflow_id}/activate",
headers=self._admin_headers(),
)
self.assertEqual(200, activate_response.status_code, activate_response.text)
workflow = activate_response.json()["workflow"]
self.assertEqual("ACTIVE", workflow["status"])
self.assertEqual(2, workflow["version"])
return workflow
def _first_target_site(self) -> dict[str, Any]:
response = self.client.get("/api/sites", headers=self._editor_headers())
self.assertEqual(200, response.status_code, response.text)
return response.json()[0]["site"]
def _intake_stage_payload(self) -> dict[str, object]:
return {
"stable_key": "intake-boundary-questions",
"display_name": "Boundary question intake",
"description": "Collect required editorial context before planning.",
"position": 1,
"owner_role": "EDITOR",
"runner_profile_key": "boundary-question-agent-v1",
"required_inputs": [
"article_brief",
"target_site",
],
"expected_outputs": [
"answered_boundary_questions",
],
"acceptance_criteria": [
"All required boundary questions are answered.",
],
"requires_human_approval": True,
"retry_policy": {
"max_attempts": 1,
"backoff_seconds": 0,
},
"parts": [
{
"key": "required-context-checklist",
"type": "checklist",
"title": "Required context checklist",
"payload": {
"prompt": "Identify missing audience and source constraints.",
},
"acceptance_criteria": [
"Missing required context is listed explicitly.",
],
}
],
}
def _draft_stage_payload(self) -> dict[str, object]:
return {
"stable_key": "draft-assembly",
"display_name": "Draft assembly",
"description": "Generate a first draft from the approved plan.",
"position": 2,
"owner_role": "EDITOR",
"runner_profile_key": "draft-writer-v1",
"required_inputs": [
"approved_plan",
],
"expected_outputs": [
"article_draft",
],
"acceptance_criteria": [
"Draft follows the approved plan.",
],
"requires_human_approval": False,
"retry_policy": {
"max_attempts": 2,
"backoff_seconds": 60,
},
"parts": [
{
"key": "draft-outline",
"type": "outline",
"title": "Draft outline",
"payload": {
"prompt": "Build a concise outline before drafting.",
},
"acceptance_criteria": [
"Every section has a purpose.",
],
}
],
}
def _admin_headers(self) -> dict[str, str]:
return {DEMO_USER_EMAIL_HEADER: DEMO_ADMIN_EMAIL}
def _editor_headers(self) -> dict[str, str]:
return {DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL}
if __name__ == "__main__":
unittest.main()