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

feat(task-005): complete article intake dashboard detail flow

Browse Source
This commit is contained in:
huncode
2026-05-21 18:54:11 +03:00
parent 7972e9a95f
commit cf84400118
28 changed files with 1714 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/tests/integration/test_article_intake_public_api.py
+217
View File
@@ -0,0 +1,217 @@
from __future__ import annotations
import sys
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.presentation.main import app # noqa: E402
DEMO_EDITOR_EMAIL = "editor@example.com"
DEMO_USER_EMAIL_HEADER = "X-Demo-User-Email"
class ArticleIntakePublicApiTest(unittest.TestCase):
def setUp(self) -> None:
self.client = TestClient(app)
def test_editor_creates_article_and_reads_dashboard_detail_timeline(self) -> None:
headers = {DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL}
editor_response = self.client.get("/api/me", headers=headers)
self.assertEqual(200, editor_response.status_code, editor_response.text)
editor_user_id = editor_response.json()["user"]["id"]
sites_response = self.client.get("/api/sites", headers=headers)
self.assertEqual(200, sites_response.status_code, sites_response.text)
target_site = sites_response.json()[0]["site"]
brief_description = (
"Create an editorial control guide for AI content intake "
f"{uuid4()}."
)
create_response = self.client.post(
"/api/articles",
headers=headers,
json={
"target_site_id": target_site["id"],
"brief_description": brief_description,
"working_title": "AI Content Intake Controls",
"content_type": "longform_guide",
"primary_keyword": "AI content intake",
},
)
self.assertEqual(201, create_response.status_code, create_response.text)
created_article = create_response.json()["article"]
self.assertEqual("ARTICLE_BRIEF_CREATED", created_article["status"])
self.assertEqual(target_site["id"], created_article["target_site_id"])
self.assertEqual(brief_description, created_article["brief_description"])
article_id = created_article["id"]
list_response = self.client.get("/api/articles", headers=headers)
detail_response = self.client.get(f"/api/articles/{article_id}", headers=headers)
failures: list[str] = []
if list_response.status_code != 200:
failures.append(
"GET /api/articles should return the persisted dashboard list; "
f"got {list_response.status_code}: {list_response.text}"
)
else:
article = self._find_article(list_response.json(), article_id)
if article is None:
failures.append(
"Created article is missing from GET /api/articles dashboard list"
)
else:
self._expect_equal(
failures,
"dashboard article status",
"ARTICLE_BRIEF_CREATED",
article.get("status"),
)
self._expect_equal(
failures,
"dashboard article target_site_id",
target_site["id"],
article.get("target_site_id"),
)
if detail_response.status_code != 200:
failures.append(
"GET /api/articles/{article_id} should return the persisted detail "
f"shell; got {detail_response.status_code}: {detail_response.text}"
)
else:
detail = detail_response.json()
if not isinstance(detail, dict):
failures.append("Article detail should return a JSON object")
detail = {}
detail_article = detail.get("article", {})
self._expect_equal(
failures,
"detail article id",
article_id,
detail_article.get("id"),
)
self._expect_equal(
failures,
"detail article status",
"ARTICLE_BRIEF_CREATED",
detail_article.get("status"),
)
detail_target_site = detail.get("target_site")
if not isinstance(detail_target_site, dict):
failures.append("Article detail should include target_site summary")
else:
self._expect_equal(
failures,
"detail target_site id",
target_site["id"],
detail_target_site.get("id"),
)
self._expect_equal(
failures,
"detail target_site slug",
target_site["slug"],
detail_target_site.get("slug"),
)
workflow_events = detail.get("workflow_events")
if not isinstance(workflow_events, list):
failures.append("Article detail should include workflow_events timeline")
else:
created_events = [
event
for event in workflow_events
if event.get("event_type") == "ARTICLE_CREATED"
]
if not created_events:
failures.append(
"Article detail workflow_events should contain ARTICLE_CREATED"
)
else:
created_event = created_events[0]
self._expect_equal(
failures,
"ARTICLE_CREATED article_id",
article_id,
created_event.get("article_id"),
)
self._expect_equal(
failures,
"ARTICLE_CREATED to_status",
"ARTICLE_BRIEF_CREATED",
created_event.get("to_status"),
)
self._expect_equal(
failures,
"ARTICLE_CREATED actor_user_id",
editor_user_id,
created_event.get("actor_user_id"),
)
if not created_event.get("created_at"):
failures.append("ARTICLE_CREATED should include created_at")
self.assertEqual([], failures)
def test_editor_cannot_create_article_with_unknown_target_site(self) -> None:
response = self.client.post(
"/api/articles",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
json={
"target_site_id": "00000000-0000-0000-0000-000000000000",
"brief_description": "Intake target-site guardrail check.",
"content_type": "longform_guide",
"primary_keyword": "integration",
},
)
self.assertEqual(400, response.status_code, response.text)
self.assertEqual("Invalid target_site_id", response.json()["detail"])
def _find_article(
self,
response_body: Any,
article_id: str,
) -> dict[str, Any] | None:
if not isinstance(response_body, dict):
return None
articles = response_body.get("articles")
if not isinstance(articles, list):
return None
return next(
(
article
for article in articles
if isinstance(article, dict) and article.get("id") == article_id
),
None,
)
def _expect_equal(
self,
failures: list[str],
label: str,
expected: object,
actual: object,
) -> None:
if actual != expected:
failures.append(f"{label}: expected {expected!r}, got {actual!r}")
if __name__ == "__main__":
unittest.main()