task015 implement seo and language review flow
This commit is contained in:
@@ -0,0 +1,431 @@
|
||||
from __future__ import annotations
|
||||
|
||||
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 SeoLanguageReviewPublicApiTest(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) / 'seo-language-review.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_seo_report_includes_required_fields(self) -> None:
|
||||
article_id = self._prepare_article_with_assembled_draft()
|
||||
self._run_seo_review(article_id)
|
||||
|
||||
report = self._get_seo_report(article_id)
|
||||
self.assertIn("score", report)
|
||||
self.assertIn("issues", report)
|
||||
self.assertIn("recommended_slug", report)
|
||||
self.assertIn("recommended_title", report)
|
||||
self.assertIn("schema_json", report)
|
||||
self.assertGreaterEqual(len(report["issues"]), 1)
|
||||
|
||||
first_issue = report["issues"][0]
|
||||
self.assertIn("severity", first_issue)
|
||||
self.assertIn("location", first_issue)
|
||||
self.assertIn("message", first_issue)
|
||||
self.assertIn("suggested_fix", first_issue)
|
||||
self.assertIn("suggestion_id", first_issue)
|
||||
|
||||
def test_language_report_includes_required_fields(self) -> None:
|
||||
article_id = self._prepare_article_with_assembled_draft()
|
||||
self._seed_language_problematic_draft(article_id)
|
||||
self._run_language_review(article_id)
|
||||
|
||||
report = self._get_language_report(article_id)
|
||||
self.assertGreaterEqual(len(report["issues"]), 1)
|
||||
self.assertTrue(any(issue.get("suggested_rewrite") for issue in report["issues"]))
|
||||
first_issue = report["issues"][0]
|
||||
self.assertIn("severity", first_issue)
|
||||
self.assertIn("location", first_issue)
|
||||
self.assertIn("message", first_issue)
|
||||
self.assertIn("suggested_rewrite", first_issue)
|
||||
|
||||
def test_editor_actions_accept_reject_edit_and_immutable_drafts(self) -> None:
|
||||
article_id = self._prepare_article_with_assembled_draft()
|
||||
self._seed_language_problematic_draft(article_id)
|
||||
|
||||
self._run_seo_review(article_id)
|
||||
self._run_language_review(article_id)
|
||||
|
||||
seo_report = self._get_seo_report(article_id)
|
||||
language_report = self._get_language_report(article_id)
|
||||
base_latest = self._latest_draft(article_id)
|
||||
|
||||
seo_patch_issue = next(
|
||||
issue for issue in seo_report["issues"] if issue["location"] == "meta_description"
|
||||
)
|
||||
accept_response = self.client.post(
|
||||
f"/api/articles/{article_id}/seo/suggestions/{seo_patch_issue['suggestion_id']}/accept",
|
||||
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
|
||||
)
|
||||
self.assertEqual(200, accept_response.status_code, accept_response.text)
|
||||
accepted = accept_response.json()
|
||||
self.assertEqual("ACCEPTED", accepted["suggestion"]["status"])
|
||||
self.assertEqual(base_latest["version"] + 1, accepted["draft"]["version"])
|
||||
|
||||
seo_non_patch_issue = next(
|
||||
issue for issue in seo_report["issues"] if issue["location"] == "body_markdown"
|
||||
)
|
||||
reject_response = self.client.post(
|
||||
f"/api/articles/{article_id}/seo/suggestions/{seo_non_patch_issue['suggestion_id']}/reject",
|
||||
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
|
||||
)
|
||||
self.assertEqual(200, reject_response.status_code, reject_response.text)
|
||||
self.assertEqual("REJECTED", reject_response.json()["suggestion"]["status"])
|
||||
self.assertIsNone(reject_response.json()["draft"])
|
||||
|
||||
language_rewrite_issue = next(
|
||||
issue for issue in language_report["issues"] if "Sentence exceeds" in issue["message"]
|
||||
)
|
||||
edit_response = self.client.post(
|
||||
f"/api/articles/{article_id}/language/suggestions/{language_rewrite_issue['suggestion_id']}/edit",
|
||||
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
|
||||
json={"applied_text": "This sentence now states the same idea clearly."},
|
||||
)
|
||||
self.assertEqual(200, edit_response.status_code, edit_response.text)
|
||||
edited = edit_response.json()
|
||||
self.assertEqual("EDITED", edited["suggestion"]["status"])
|
||||
self.assertEqual(
|
||||
accepted["draft"]["version"] + 1,
|
||||
edited["draft"]["version"],
|
||||
)
|
||||
self.assertIn("states the same idea clearly", edited["draft"]["body_markdown"])
|
||||
|
||||
def test_target_site_overrides_are_applied(self) -> None:
|
||||
site = self.client.get(
|
||||
"/api/sites",
|
||||
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
|
||||
).json()[0]["site"]
|
||||
patch_response = self.client.patch(
|
||||
f"/api/sites/{site['id']}",
|
||||
headers={DEMO_USER_EMAIL_HEADER: DEMO_ADMIN_EMAIL},
|
||||
json={
|
||||
"seo_rules": {
|
||||
"seo_review": {
|
||||
"meta_description_min": 10,
|
||||
"meta_description_max": 40,
|
||||
"schema_type": "HowTo",
|
||||
},
|
||||
"language_review": {
|
||||
"max_sentence_words": 8,
|
||||
"forbidden_phrases": ["precision tone"],
|
||||
"brand_tone": "strict",
|
||||
},
|
||||
}
|
||||
},
|
||||
)
|
||||
self.assertEqual(200, patch_response.status_code, patch_response.text)
|
||||
|
||||
article_id = self._prepare_article_with_assembled_draft(target_site_id=UUID(site["id"]))
|
||||
self._seed_language_problematic_draft(
|
||||
article_id,
|
||||
extra_body="This precision tone phrase should be avoided!",
|
||||
)
|
||||
self._run_seo_review(article_id)
|
||||
self._run_language_review(article_id)
|
||||
|
||||
seo_report = self._get_seo_report(article_id)
|
||||
self.assertEqual("HowTo", seo_report["schema_json"]["@type"])
|
||||
|
||||
language_report = self._get_language_report(article_id)
|
||||
self.assertEqual(8, language_report["rules_snapshot"]["max_sentence_words"])
|
||||
self.assertEqual("strict", language_report["rules_snapshot"]["brand_tone"])
|
||||
self.assertTrue(
|
||||
any("8 words" in issue["message"] for issue in language_report["issues"])
|
||||
)
|
||||
self.assertTrue(
|
||||
any("Forbidden phrase" in issue["message"] for issue in language_report["issues"])
|
||||
)
|
||||
|
||||
def test_final_review_shows_unresolved_issues(self) -> None:
|
||||
article_id = self._prepare_article_with_assembled_draft()
|
||||
self._seed_language_problematic_draft(article_id)
|
||||
self._run_seo_review(article_id)
|
||||
self._run_language_review(article_id)
|
||||
|
||||
seo_report = self._get_seo_report(article_id)
|
||||
issue_to_accept = seo_report["issues"][0]
|
||||
accept_response = self.client.post(
|
||||
f"/api/articles/{article_id}/seo/suggestions/{issue_to_accept['suggestion_id']}/accept",
|
||||
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
|
||||
)
|
||||
self.assertEqual(200, accept_response.status_code, accept_response.text)
|
||||
|
||||
unresolved_response = self.client.get(
|
||||
f"/api/articles/{article_id}/final-review/issues",
|
||||
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
|
||||
)
|
||||
self.assertEqual(200, unresolved_response.status_code, unresolved_response.text)
|
||||
unresolved = unresolved_response.json()
|
||||
self.assertGreaterEqual(unresolved["unresolved_count"], 1)
|
||||
self.assertTrue(unresolved["unresolved_issues"])
|
||||
self.assertTrue(
|
||||
all(item["status"] == "PENDING" for item in unresolved["unresolved_issues"])
|
||||
)
|
||||
|
||||
def test_fake_runner_is_deterministic(self) -> None:
|
||||
article_id = self._prepare_article_with_assembled_draft()
|
||||
self._seed_language_problematic_draft(article_id)
|
||||
|
||||
self._run_seo_review(article_id)
|
||||
first = self._get_seo_report(article_id)
|
||||
self._run_seo_review(article_id)
|
||||
second = self._get_seo_report(article_id)
|
||||
|
||||
self.assertEqual(first["score"], second["score"])
|
||||
self.assertEqual(first["recommended_slug"], second["recommended_slug"])
|
||||
self.assertEqual(first["recommended_title"], second["recommended_title"])
|
||||
|
||||
first_issue_signature = [
|
||||
(
|
||||
issue["id"],
|
||||
issue["severity"],
|
||||
issue["location"],
|
||||
issue["message"],
|
||||
issue["suggested_fix"],
|
||||
issue["suggested_rewrite"],
|
||||
)
|
||||
for issue in first["issues"]
|
||||
]
|
||||
second_issue_signature = [
|
||||
(
|
||||
issue["id"],
|
||||
issue["severity"],
|
||||
issue["location"],
|
||||
issue["message"],
|
||||
issue["suggested_fix"],
|
||||
issue["suggested_rewrite"],
|
||||
)
|
||||
for issue in second["issues"]
|
||||
]
|
||||
self.assertEqual(first_issue_signature, second_issue_signature)
|
||||
|
||||
def _run_seo_review(self, article_id: UUID) -> dict[str, Any]:
|
||||
response = self.client.post(
|
||||
f"/api/articles/{article_id}/seo/review",
|
||||
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
|
||||
)
|
||||
self.assertEqual(201, response.status_code, response.text)
|
||||
return response.json()["report"]
|
||||
|
||||
def _run_language_review(self, article_id: UUID) -> dict[str, Any]:
|
||||
response = self.client.post(
|
||||
f"/api/articles/{article_id}/language/review",
|
||||
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
|
||||
)
|
||||
self.assertEqual(201, response.status_code, response.text)
|
||||
return response.json()["report"]
|
||||
|
||||
def _get_seo_report(self, article_id: UUID) -> dict[str, Any]:
|
||||
response = self.client.get(
|
||||
f"/api/articles/{article_id}/seo/report",
|
||||
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
|
||||
)
|
||||
self.assertEqual(200, response.status_code, response.text)
|
||||
return response.json()["report"]
|
||||
|
||||
def _get_language_report(self, article_id: UUID) -> dict[str, Any]:
|
||||
response = self.client.get(
|
||||
f"/api/articles/{article_id}/language/report",
|
||||
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
|
||||
)
|
||||
self.assertEqual(200, response.status_code, response.text)
|
||||
return response.json()["report"]
|
||||
|
||||
def _latest_draft(self, article_id: UUID) -> dict[str, Any]:
|
||||
response = self.client.get(
|
||||
f"/api/articles/{article_id}/drafts",
|
||||
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
|
||||
)
|
||||
self.assertEqual(200, response.status_code, response.text)
|
||||
return response.json()["drafts"][0]
|
||||
|
||||
def _seed_language_problematic_draft(
|
||||
self,
|
||||
article_id: UUID,
|
||||
*,
|
||||
extra_body: str = "",
|
||||
) -> None:
|
||||
latest = self._latest_draft(article_id)
|
||||
body = "\n".join(
|
||||
[
|
||||
latest["body_markdown"],
|
||||
(
|
||||
"This sentence is intentionally extremely long and includes many additional words "
|
||||
"that make it exceed any practical editorial threshold for readability while still "
|
||||
"describing the same guidance in an unnecessarily verbose and repetitive way for testing."
|
||||
),
|
||||
"Our approach is the best ever for everyone.",
|
||||
extra_body,
|
||||
]
|
||||
)
|
||||
patch_response = self.client.patch(
|
||||
f"/api/articles/{article_id}/drafts/{latest['id']}",
|
||||
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
|
||||
json={"body_markdown": body},
|
||||
)
|
||||
self.assertEqual(200, patch_response.status_code, patch_response.text)
|
||||
|
||||
def _prepare_article_with_assembled_draft(
|
||||
self,
|
||||
*,
|
||||
target_site_id: UUID | None = None,
|
||||
) -> UUID:
|
||||
article_id, _ = self._prepare_article_with_approved_plan(target_site_id=target_site_id)
|
||||
self._ensure_evidence_ready(article_id)
|
||||
section_jobs = self._start_parallel_production(article_id)
|
||||
self._complete_all_section_jobs_success(section_jobs)
|
||||
self._assemble_draft(article_id)
|
||||
return article_id
|
||||
|
||||
def _prepare_article_with_approved_plan(
|
||||
self,
|
||||
*,
|
||||
target_site_id: UUID | None = None,
|
||||
) -> tuple[UUID, dict[str, Any]]:
|
||||
if target_site_id is None:
|
||||
site = self.client.get(
|
||||
"/api/sites",
|
||||
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
|
||||
).json()[0]["site"]
|
||||
target_site_id = UUID(site["id"])
|
||||
article = self.client.post(
|
||||
"/api/articles",
|
||||
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
|
||||
json={
|
||||
"target_site_id": str(target_site_id),
|
||||
"brief_description": "SEO and language review implementation coverage.",
|
||||
"working_title": "Short",
|
||||
"content_type": "longform_guide",
|
||||
"primary_keyword": "seo language review",
|
||||
},
|
||||
).json()["article"]
|
||||
article_id = UUID(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"]
|
||||
|
||||
approve_plan_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_plan_response.status_code, approve_plan_response.text)
|
||||
return article_id, plan
|
||||
|
||||
def _ensure_evidence_ready(self, article_id: UUID) -> 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 _start_parallel_production(self, article_id: UUID) -> 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)
|
||||
jobs = response.json()["jobs"]
|
||||
section_jobs = [job for job in jobs if job["job_type"] == "SECTION_SCAFFOLD"]
|
||||
self.assertTrue(section_jobs)
|
||||
return section_jobs
|
||||
|
||||
def _complete_all_section_jobs_success(self, section_jobs: list[dict[str, Any]]) -> None:
|
||||
for index, job in enumerate(section_jobs, start=1):
|
||||
complete_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": {
|
||||
"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}.",
|
||||
},
|
||||
},
|
||||
},
|
||||
)
|
||||
self.assertEqual(200, complete_response.status_code, complete_response.text)
|
||||
self.assertEqual("SUCCEEDED", complete_response.json()["job"]["status"])
|
||||
|
||||
def _assemble_draft(self, article_id: UUID) -> dict[str, Any]:
|
||||
response = self.client.post(
|
||||
f"/api/articles/{article_id}/draft/assemble",
|
||||
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
|
||||
)
|
||||
self.assertEqual(201, response.status_code, response.text)
|
||||
return response.json()["draft"]
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user