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

Task 008 implement boundary questions loop

Browse Source
This commit is contained in:
huncode
2026-05-21 19:46:26 +03:00
parent 8aa92ea728
commit b7e2fa34b2
27 changed files with 1883 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/src/application/articles.py
+2
View File
@@ -62,10 +62,12 @@ def get_article_detail(
article = repository.articles.get(article_id)
target_site = repository.target_sites.get_by_id(article.target_site_id)
workflow_events = repository.articles.list_workflow_events(article_id)
boundary_questions = repository.boundary_questions.list_for_article(article_id)
return ArticleDetailResponse(
article=article,
target_site=target_site,
workflow_events=workflow_events,
boundary_questions=boundary_questions,
)
apps/backend/src/application/boundary_questions.py
+265
View File
@@ -0,0 +1,265 @@
from __future__ import annotations
from datetime import UTC, datetime
from uuid import UUID
from src.domain.contracts import (
AgentJobErrorCategory,
AgentJobListResponse,
AgentJobStatus,
AgentJobType,
ArticleSummary,
ArticleWorkflowStatus,
BoundaryQuestionListResponse,
BoundaryQuestionResponse,
BoundaryQuestionUpdateRequest,
CurrentUser,
)
BOUNDARY_QUESTION_FIXTURE: tuple[dict[str, object], ...] = (
{
"category": "audience",
"label": "Audience",
"is_required": True,
},
{
"category": "purpose",
"label": "Purpose",
"is_required": True,
},
{
"category": "reader_outcome",
"label": "Reader outcome",
"is_required": True,
},
{
"category": "depth",
"label": "Depth",
"is_required": True,
},
{
"category": "tone",
"label": "Tone",
"is_required": True,
},
{
"category": "excluded_topics",
"label": "Excluded topics",
"is_required": True,
},
{
"category": "primary_keyword",
"label": "Primary keyword",
"is_required": True,
},
{
"category": "competitor_angle",
"label": "Competitor angle",
"is_required": True,
},
{
"category": "evidence_standard",
"label": "Evidence standard",
"is_required": True,
},
{
"category": "visual_expectations",
"label": "Visual expectations",
"is_required": False,
},
)
def generate_boundary_questions(
repository: object,
*,
article_id: UUID,
current_user: CurrentUser,
) -> BoundaryQuestionListResponse:
article = repository.articles.get(article_id)
target_site = repository.target_sites.get_by_id(article.target_site_id)
now = _now()
previous_status = article.status
job = repository.agent_jobs.create(
article_id=article_id,
parent_job_id=None,
attempt=1,
job_type=AgentJobType.BOUNDARY_QUESTIONS,
agent_profile="fake-boundary-questions",
status=AgentJobStatus.QUEUED,
input_files=[
{
"path": "inputs/article-brief.json",
"content_hash": None,
},
{
"path": "inputs/target-site-config.json",
"content_hash": None,
},
],
queued_at=now,
)
job = repository.agent_jobs.complete(
job_id=job.id,
status=AgentJobStatus.SUCCEEDED,
workspace_path=None,
output_files=[{"path": "outputs/boundary-questions.json"}],
error_category=None,
error_message=None,
stdout="fake boundary question fixture generated\n",
stderr="",
exit_code=0,
duration_ms=0,
finished_at=now,
)
questions = repository.boundary_questions.replace_for_article(
article_id=article_id,
questions=_build_questions(article=article, target_site=target_site),
created_at=now,
)
article = repository.articles.update_status(
article_id=article_id,
status=ArticleWorkflowStatus.BOUNDARY_QUESTIONS_GENERATED,
updated_at=now,
)
repository.articles.create_workflow_event(
article_id=article.id,
event_type="BOUNDARY_QUESTIONS_GENERATED",
from_status=previous_status,
to_status=ArticleWorkflowStatus.BOUNDARY_QUESTIONS_GENERATED,
actor_user_id=current_user.id,
payload={"agent_job_id": str(job.id), "question_count": len(questions)},
created_at=now,
)
return BoundaryQuestionListResponse(questions=questions, job=job)
def list_boundary_questions(
repository: object,
*,
article_id: UUID,
) -> BoundaryQuestionListResponse:
repository.articles.get(article_id)
return BoundaryQuestionListResponse(
questions=repository.boundary_questions.list_for_article(article_id),
)
def update_boundary_question(
repository: object,
*,
article_id: UUID,
question_id: UUID,
request: BoundaryQuestionUpdateRequest,
) -> BoundaryQuestionResponse:
repository.articles.get(article_id)
answer = request.answer.strip() if request.answer is not None else None
question = repository.boundary_questions.update_answer(
article_id=article_id,
question_id=question_id,
answer=answer or None,
updated_at=_now(),
)
return BoundaryQuestionResponse(question=question)
def submit_boundary_questions(
repository: object,
*,
article_id: UUID,
current_user: CurrentUser,
) -> ArticleSummary:
article = repository.articles.get(article_id)
questions = repository.boundary_questions.list_for_article(article_id)
if not questions:
raise ValueError("Boundary questions must be generated before submission")
missing_required = [
question.category
for question in questions
if question.is_required and not (question.answer or "").strip()
]
if missing_required:
raise ValueError(
"All required boundary questions must be answered: "
+ ", ".join(missing_required)
)
now = _now()
updated_article = repository.articles.update_status(
article_id=article_id,
status=ArticleWorkflowStatus.BOUNDARY_ANSWERS_SUBMITTED,
updated_at=now,
)
repository.articles.create_workflow_event(
article_id=article_id,
event_type="BOUNDARY_ANSWERS_SUBMITTED",
from_status=article.status,
to_status=ArticleWorkflowStatus.BOUNDARY_ANSWERS_SUBMITTED,
actor_user_id=current_user.id,
payload={"required_answer_count": len(questions) - len(missing_required)},
created_at=now,
)
return updated_article
def create_plan_generation_job(
repository: object,
*,
article_id: UUID,
) -> AgentJobListResponse:
article = repository.articles.get(article_id)
if article.status != ArticleWorkflowStatus.BOUNDARY_ANSWERS_SUBMITTED:
raise PermissionError("Boundary answers must be submitted before plan generation")
job = repository.agent_jobs.create(
article_id=article_id,
parent_job_id=None,
attempt=1,
job_type=AgentJobType.PLAN_GENERATION,
agent_profile="fake-plan-generation",
status=AgentJobStatus.QUEUED,
input_files=[{"path": "inputs/boundary-answers.json", "content_hash": None}],
queued_at=_now(),
)
return AgentJobListResponse(jobs=[job])
def _build_questions(*, article: ArticleSummary, target_site: object) -> list[dict[str, object]]:
title = article.working_title or article.primary_keyword or article.content_type
site_audience = getattr(target_site, "audience")
brand_voice = getattr(target_site, "brand_voice")
primary_keyword = article.primary_keyword or "the primary topic"
brief = article.brief_description
prompts = {
"audience": f"Who exactly is this article for on {site_audience}?",
"purpose": f"What business or editorial purpose should '{title}' serve?",
"reader_outcome": "What should the reader be able to decide or do after reading?",
"depth": f"How deep should coverage go for this {article.content_type} brief?",
"tone": f"What tone should adapt the site voice '{brand_voice}' for this brief?",
"excluded_topics": "Which topics, claims, or comparisons must be excluded?",
"primary_keyword": f"How should the keyword '{primary_keyword}' be prioritized?",
"competitor_angle": "Which competitor angle should the article address or avoid?",
"evidence_standard": "What evidence standard is required for claims in this article?",
"visual_expectations": "What diagrams, tables, or images would clarify the brief?",
}
return [
{
"sort_order": index,
"category": str(item["category"]),
"question": f"{item['label']}: {prompts[str(item['category'])]} Brief: {brief}",
"answer": None,
"is_required": bool(item["is_required"]),
}
for index, item in enumerate(BOUNDARY_QUESTION_FIXTURE, start=1)
]
def _now() -> datetime:
return datetime.now(UTC)
apps/backend/src/domain/contracts/__init__.py
+8
View File
@@ -25,6 +25,10 @@ from .models import (
ArticleDetailResponse,
ArticleListResponse,
ArticleSummary,
BoundaryQuestionListResponse,
BoundaryQuestionResponse,
BoundaryQuestionSummary,
BoundaryQuestionUpdateRequest,
WorkflowEventSummary,
AssetSummary,
ClaimSummary,
@@ -73,6 +77,10 @@ __all__ = [
"ArticleDetailResponse",
"ArticleListResponse",
"ArticleSummary",
"BoundaryQuestionListResponse",
"BoundaryQuestionResponse",
"BoundaryQuestionSummary",
"BoundaryQuestionUpdateRequest",
"WorkflowEventSummary",
"ArticleWorkflowStatus",
"AssetStatus",
apps/backend/src/domain/contracts/models.py
+26
View File
@@ -183,6 +183,31 @@ class WorkflowEventSummary(ContractModel):
created_at: datetime
class BoundaryQuestionSummary(ContractModel):
id: UUID
article_id: UUID
sort_order: int = Field(ge=1)
category: str = Field(min_length=1)
question: str = Field(min_length=1)
answer: str | None = None
is_required: bool = True
created_at: datetime
updated_at: datetime
class BoundaryQuestionUpdateRequest(ContractModel):
answer: str | None = None
class BoundaryQuestionResponse(ContractModel):
question: BoundaryQuestionSummary
class BoundaryQuestionListResponse(ContractModel):
questions: list[BoundaryQuestionSummary]
job: AgentJobSummary | None = None
class PlanSectionSummary(ContractModel):
id: UUID
article_plan_id: UUID
@@ -351,6 +376,7 @@ class ArticleDetailResponse(ContractModel):
article: ArticleSummary
target_site: TargetSiteConfig | None = None
workflow_events: list[WorkflowEventSummary] = Field(default_factory=list)
boundary_questions: list[BoundaryQuestionSummary] = Field(default_factory=list)
plan: PlanSummary | None = None
draft: DraftSummary | None = None
evidence: list[EvidenceSummary] = Field(default_factory=list)
apps/backend/src/domain/contracts/openapi.py
+8
View File
@@ -32,6 +32,10 @@ from .models import (
ArticleDetailResponse,
ArticleListResponse,
ArticleSummary,
BoundaryQuestionListResponse,
BoundaryQuestionResponse,
BoundaryQuestionSummary,
BoundaryQuestionUpdateRequest,
AssetSummary,
ClaimSummary,
CurrentUser,
@@ -95,6 +99,10 @@ CONTRACT_SCHEMA_MODELS: tuple[type[BaseModel], ...] = (
ArticleCreateResponse,
ArticleListResponse,
WorkflowEventSummary,
BoundaryQuestionSummary,
BoundaryQuestionUpdateRequest,
BoundaryQuestionResponse,
BoundaryQuestionListResponse,
PlanSectionSummary,
PlanSummary,
ResearchArtifactSummary,
apps/backend/src/infrastructure/repositories.py
+175
View File
@@ -14,6 +14,7 @@ from src.domain.contracts import (
AgentJobSummary,
AgentJobType,
ArticleSummary,
BoundaryQuestionSummary,
PublishingRules,
PublishingStatus,
ArticleWorkflowStatus,
@@ -40,6 +41,7 @@ class BackendRepository:
self.users = UsersRepository(self)
self.target_sites = TargetSitesRepository(self)
self.articles = ArticlesRepository(self)
self.boundary_questions = BoundaryQuestionsRepository(self)
self.agent_jobs = AgentJobsRepository(self)
self.script_config_versions = ScriptConfigVersionsRepository(self)
self.script_config_version_events = ScriptConfigVersionAuditEventsRepository(self)
@@ -519,6 +521,26 @@ class ArticlesRepository:
raise LookupError(f"Article not found: {article_id}")
return _article_summary_from_row(row)
def update_status(
self,
*,
article_id: UUID,
status: ArticleWorkflowStatus,
updated_at: datetime,
) -> ArticleSummary:
placeholder = self._repository.placeholder()
with self._repository.connection() as connection:
connection.execute(
f"""
UPDATE articles
SET status = {placeholder}, updated_at = {placeholder}
WHERE id = {placeholder}
""",
(status.value, _datetime_value(updated_at), str(article_id)),
)
return self.get(article_id)
def create_workflow_event(
self,
*,
@@ -593,6 +615,141 @@ class ArticlesRepository:
return [_workflow_event_from_row(row) for row in rows]
class BoundaryQuestionsRepository:
def __init__(self, repository: BackendRepository) -> None:
self._repository = repository
def replace_for_article(
self,
*,
article_id: UUID,
questions: list[JsonObject],
created_at: datetime,
) -> list[BoundaryQuestionSummary]:
placeholder = self._repository.placeholder()
with self._repository.connection() as connection:
connection.execute(
f"DELETE FROM boundary_questions WHERE article_id = {placeholder}",
(str(article_id),),
)
for question in questions:
connection.execute(
f"""
INSERT INTO boundary_questions (
id,
article_id,
sort_order,
category,
question,
answer,
is_required,
created_at,
updated_at
)
VALUES (
{placeholder},
{placeholder},
{placeholder},
{placeholder},
{placeholder},
{placeholder},
{placeholder},
{placeholder},
{placeholder}
)
""",
(
str(uuid4()),
str(article_id),
question["sort_order"],
question["category"],
question["question"],
question.get("answer"),
_bool_value(question.get("is_required", True)),
_datetime_value(created_at),
_datetime_value(created_at),
),
)
return self.list_for_article(article_id)
def list_for_article(self, article_id: UUID) -> list[BoundaryQuestionSummary]:
placeholder = self._repository.placeholder()
with self._repository.connection() as connection:
rows = connection.execute(
f"""
SELECT
id,
article_id,
sort_order,
category,
question,
answer,
is_required,
created_at,
updated_at
FROM boundary_questions
WHERE article_id = {placeholder}
ORDER BY sort_order
""",
(str(article_id),),
).fetchall()
return [_boundary_question_from_row(row) for row in rows]
def get(self, *, article_id: UUID, question_id: UUID) -> BoundaryQuestionSummary:
placeholder = self._repository.placeholder()
with self._repository.connection() as connection:
row = connection.execute(
f"""
SELECT
id,
article_id,
sort_order,
category,
question,
answer,
is_required,
created_at,
updated_at
FROM boundary_questions
WHERE article_id = {placeholder} AND id = {placeholder}
""",
(str(article_id), str(question_id)),
).fetchone()
if row is None:
raise LookupError(f"Boundary question not found: {question_id}")
return _boundary_question_from_row(row)
def update_answer(
self,
*,
article_id: UUID,
question_id: UUID,
answer: str | None,
updated_at: datetime,
) -> BoundaryQuestionSummary:
self.get(article_id=article_id, question_id=question_id)
placeholder = self._repository.placeholder()
with self._repository.connection() as connection:
connection.execute(
f"""
UPDATE boundary_questions
SET answer = {placeholder}, updated_at = {placeholder}
WHERE article_id = {placeholder} AND id = {placeholder}
""",
(
answer,
_datetime_value(updated_at),
str(article_id),
str(question_id),
),
)
return self.get(article_id=article_id, question_id=question_id)
class AgentJobsRepository:
def __init__(self, repository: BackendRepository) -> None:
self._repository = repository
@@ -1247,6 +1404,20 @@ def _workflow_event_from_row(row: Any) -> WorkflowEventSummary:
)
def _boundary_question_from_row(row: Any) -> BoundaryQuestionSummary:
return BoundaryQuestionSummary(
id=_row_value(row, "id"),
article_id=_row_value(row, "article_id"),
sort_order=_row_value(row, "sort_order"),
category=_row_value(row, "category"),
question=_row_value(row, "question"),
answer=_row_value(row, "answer"),
is_required=bool(_row_value(row, "is_required")),
created_at=_row_value(row, "created_at"),
updated_at=_row_value(row, "updated_at"),
)
def _agent_job_summary_from_row(row: Any) -> AgentJobSummary:
return AgentJobSummary(
id=_row_value(row, "id"),
@@ -1331,6 +1502,10 @@ def _datetime_value(value: datetime | None) -> str | None:
return value.isoformat()
def _bool_value(value: bool) -> bool:
return bool(value)
def _article_status_value(value: ArticleWorkflowStatus | None) -> str | None:
if value is None:
return None
apps/backend/src/infrastructure/schema.py
+3
View File
@@ -116,6 +116,7 @@ POSTGRES_SCHEMA_STATEMENTS: tuple[str, ...] = (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
article_id UUID NOT NULL REFERENCES articles(id) ON DELETE CASCADE,
sort_order INTEGER NOT NULL,
category TEXT NOT NULL DEFAULT 'general',
question TEXT NOT NULL,
answer TEXT,
is_required BOOLEAN NOT NULL DEFAULT true,
@@ -124,6 +125,7 @@ POSTGRES_SCHEMA_STATEMENTS: tuple[str, ...] = (
UNIQUE (article_id, sort_order)
)
""",
"ALTER TABLE boundary_questions ADD COLUMN IF NOT EXISTS category TEXT NOT NULL DEFAULT 'general'",
f"""
CREATE TABLE IF NOT EXISTS article_plans (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
@@ -413,6 +415,7 @@ SQLITE_SCHEMA_STATEMENTS: tuple[str, ...] = (
id TEXT PRIMARY KEY,
article_id TEXT NOT NULL,
sort_order INTEGER NOT NULL,
category TEXT NOT NULL DEFAULT 'general',
question TEXT NOT NULL,
answer TEXT,
is_required INTEGER NOT NULL DEFAULT 1,
apps/backend/src/presentation/main.py
+2
View File
@@ -11,12 +11,14 @@ from src.presentation.routes.agent_jobs import (
from src.presentation.routes.agent_jobs import router as agent_jobs_router
from src.presentation.routes.articles import router as articles_router
from src.presentation.routes.auth import router as auth_router
from src.presentation.routes.boundary_questions import router as boundary_questions_router
from src.presentation.routes.sites import router as sites_router
app = FastAPI(title="AI Content Pipeline Backend")
app.include_router(auth_router)
app.include_router(articles_router)
app.include_router(boundary_questions_router)
app.include_router(agent_jobs_router)
app.include_router(internal_agent_jobs_router)
app.include_router(sites_router)
apps/backend/src/presentation/routes/boundary_questions.py
+146
View File
@@ -0,0 +1,146 @@
from __future__ import annotations
from uuid import UUID
from fastapi import APIRouter, Depends, HTTPException, status
from src.application.boundary_questions import (
create_plan_generation_job,
generate_boundary_questions,
list_boundary_questions,
submit_boundary_questions,
update_boundary_question,
)
from src.domain.auth import EDITOR_OR_ADMIN_ROLES
from src.domain.contracts import (
AgentJobListResponse,
ArticleCreateResponse,
BoundaryQuestionListResponse,
BoundaryQuestionResponse,
BoundaryQuestionUpdateRequest,
CurrentUser,
)
from src.infrastructure.repositories import BackendRepository
from src.presentation.dependencies import get_repository, require_roles
router = APIRouter(prefix="/api", tags=["boundary-questions"])
@router.post(
"/articles/{article_id}/boundary-questions/generate",
response_model=BoundaryQuestionListResponse,
status_code=status.HTTP_201_CREATED,
)
def post_generate_boundary_questions(
article_id: UUID,
current_user: CurrentUser = Depends(require_roles(EDITOR_OR_ADMIN_ROLES)),
repository: BackendRepository = Depends(get_repository),
) -> BoundaryQuestionListResponse:
try:
return generate_boundary_questions(
repository,
article_id=article_id,
current_user=current_user,
)
except LookupError as error:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Not Found",
) from error
@router.get(
"/articles/{article_id}/boundary-questions",
response_model=BoundaryQuestionListResponse,
)
def get_boundary_questions(
article_id: UUID,
_: CurrentUser = Depends(require_roles(EDITOR_OR_ADMIN_ROLES)),
repository: BackendRepository = Depends(get_repository),
) -> BoundaryQuestionListResponse:
try:
return list_boundary_questions(repository, article_id=article_id)
except LookupError as error:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Not Found",
) from error
@router.patch(
"/articles/{article_id}/boundary-questions/{question_id}",
response_model=BoundaryQuestionResponse,
)
def patch_boundary_question(
article_id: UUID,
question_id: UUID,
request: BoundaryQuestionUpdateRequest,
_: CurrentUser = Depends(require_roles(EDITOR_OR_ADMIN_ROLES)),
repository: BackendRepository = Depends(get_repository),
) -> BoundaryQuestionResponse:
try:
return update_boundary_question(
repository,
article_id=article_id,
question_id=question_id,
request=request,
)
except LookupError as error:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Not Found",
) from error
@router.post(
"/articles/{article_id}/boundary-questions/submit",
response_model=ArticleCreateResponse,
)
def post_submit_boundary_questions(
article_id: UUID,
current_user: CurrentUser = Depends(require_roles(EDITOR_OR_ADMIN_ROLES)),
repository: BackendRepository = Depends(get_repository),
) -> ArticleCreateResponse:
try:
article = submit_boundary_questions(
repository,
article_id=article_id,
current_user=current_user,
)
except LookupError as error:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Not Found",
) from error
except ValueError as error:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=str(error),
) from error
return ArticleCreateResponse(article=article)
@router.post(
"/articles/{article_id}/plans/generate",
response_model=AgentJobListResponse,
status_code=status.HTTP_202_ACCEPTED,
)
def post_generate_plan(
article_id: UUID,
_: CurrentUser = Depends(require_roles(EDITOR_OR_ADMIN_ROLES)),
repository: BackendRepository = Depends(get_repository),
) -> AgentJobListResponse:
try:
return create_plan_generation_job(repository, article_id=article_id)
except LookupError as error:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Not Found",
) from error
except PermissionError as error:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=str(error),
) from error
apps/backend/tests/integration/test_boundary_questions_public_api.py
+196
View File
@@ -0,0 +1,196 @@
from __future__ import annotations
import sys
import tempfile
import unittest
from pathlib import Path
from typing import Any
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 BoundaryQuestionsPublicApiTest(unittest.TestCase):
def setUp(self) -> None:
self.tmp_dir = tempfile.TemporaryDirectory()
dsn = f"sqlite:///{Path(self.tmp_dir.name) / 'boundary-questions.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_required_unanswered_questions_block_submission_then_submit_updates_workflow(
self,
) -> None:
article_id = self._create_article()
blocked_plan_response = self.client.post(
f"/api/articles/{article_id}/plans/generate",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
self.assertEqual(409, blocked_plan_response.status_code, blocked_plan_response.text)
generate_response = self.client.post(
f"/api/articles/{article_id}/boundary-questions/generate",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
self.assertEqual(201, generate_response.status_code, generate_response.text)
generated = generate_response.json()
questions = generated["questions"]
self.assertGreaterEqual(len(questions), 5)
self.assertLessEqual(len(questions), 10)
self.assertEqual("BOUNDARY_QUESTIONS", generated["job"]["job_type"])
self.assertEqual("SUCCEEDED", generated["job"]["status"])
categories = {question["category"] for question in questions}
for category in (
"audience",
"purpose",
"reader_outcome",
"depth",
"tone",
"excluded_topics",
"primary_keyword",
"competitor_angle",
"evidence_standard",
"visual_expectations",
):
self.assertIn(category, categories)
required_question = next(
question for question in questions if question["is_required"]
)
optional_question = next(
question for question in questions if not question["is_required"]
)
patch_response = self.client.patch(
f"/api/articles/{article_id}/boundary-questions/{required_question['id']}",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
json={"answer": "Senior editors evaluating automated SEO content."},
)
self.assertEqual(200, patch_response.status_code, patch_response.text)
self.assertEqual(
"Senior editors evaluating automated SEO content.",
patch_response.json()["question"]["answer"],
)
optional_patch_response = self.client.patch(
f"/api/articles/{article_id}/boundary-questions/{optional_question['id']}",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
json={"answer": "No custom visual constraints."},
)
self.assertEqual(200, optional_patch_response.status_code, optional_patch_response.text)
submit_blocked_response = self.client.post(
f"/api/articles/{article_id}/boundary-questions/submit",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
self.assertEqual(409, submit_blocked_response.status_code, submit_blocked_response.text)
self.assertIn("required", submit_blocked_response.json()["detail"])
list_response = self.client.get(
f"/api/articles/{article_id}/boundary-questions",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
self.assertEqual(200, list_response.status_code, list_response.text)
saved_required = self._find_question(
list_response.json()["questions"],
required_question["id"],
)
self.assertEqual(
"Senior editors evaluating automated SEO content.",
saved_required["answer"],
)
for question in list_response.json()["questions"]:
if question["is_required"] and not question["answer"]:
answer_response = self.client.patch(
f"/api/articles/{article_id}/boundary-questions/{question['id']}",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
json={"answer": f"Accepted answer for {question['category']}."},
)
self.assertEqual(200, answer_response.status_code, answer_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)
self.assertEqual(
"BOUNDARY_ANSWERS_SUBMITTED",
submit_response.json()["article"]["status"],
)
detail_response = self.client.get(
f"/api/articles/{article_id}",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
self.assertEqual(200, detail_response.status_code, detail_response.text)
detail = detail_response.json()
self.assertEqual("BOUNDARY_ANSWERS_SUBMITTED", detail["article"]["status"])
self.assertTrue(
any(
event["event_type"] == "BOUNDARY_ANSWERS_SUBMITTED"
for event in detail["workflow_events"]
)
)
allowed_plan_response = self.client.post(
f"/api/articles/{article_id}/plans/generate",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
self.assertEqual(202, allowed_plan_response.status_code, allowed_plan_response.text)
def _create_article(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"]
response = self.client.post(
"/api/articles",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
json={
"target_site_id": target_site["id"],
"brief_description": "Build a precise comparison of AI content systems.",
"working_title": "AI Content Systems",
"content_type": "longform_guide",
"primary_keyword": "AI content systems",
},
)
self.assertEqual(201, response.status_code, response.text)
return response.json()["article"]["id"]
def _find_question(
self,
questions: list[dict[str, Any]],
question_id: str,
) -> dict[str, Any]:
for question in questions:
if question["id"] == question_id:
return question
self.fail(f"Question not found: {question_id}")
if __name__ == "__main__":
unittest.main()
apps/backend/tests/integration/test_schema_storage_contracts.py
+4
View File
@@ -87,6 +87,10 @@ class SchemaStorageContractsIntegrationTest(unittest.TestCase):
"output_files",
}.issubset(repository.schema.list_columns("agent_jobs"))
)
self.assertIn(
"category",
repository.schema.list_columns("boundary_questions"),
)
self.assertTrue(
{
"s3_prefix",