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

Task 011 implement evidence matrix and claim gate

Browse Source
This commit is contained in:
huncode
2026-05-21 22:52:41 +03:00
parent 7d4a79acb5
commit 7e1b05067c
23 changed files with 2090 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/tests/integration/test_evidence_matrix_public_api.py
+193
View File
@@ -0,0 +1,193 @@
from __future__ import annotations
import os
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 EvidenceMatrixPublicApiTest(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) / 'evidence.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_insufficient_evidence_returns_to_plan_revision_and_blocks_draft(self) -> None:
article_id = self._create_researched_article("insufficient evidence coverage")
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)
body = evidence_response.json()
self.assertEqual("PLAN_REVISION_REQUIRED", body["article"]["status"])
self.assertTrue(body["insufficient_evidence_reasons"])
unsupported = [
claim
for claim in body["claims"]
if claim["support_status"] == "UNSUPPORTED"
]
self.assertTrue(unsupported)
self.assertTrue(
any(claim["risk_level"] == "high" for claim in unsupported)
)
draft_response = self.client.post(
f"/api/articles/{article_id}/draft/start",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
self.assertEqual(409, draft_response.status_code, draft_response.text)
final_response = self.client.post(
f"/api/articles/{article_id}/final/approve",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
self.assertEqual(409, final_response.status_code, final_response.text)
evidence_item = body["evidence"][0]
patch_response = self.client.patch(
f"/api/articles/{article_id}/evidence/{evidence_item['id']}",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
json={"review_status": "REJECTED"},
)
self.assertEqual(200, patch_response.status_code, patch_response.text)
self.assertEqual("REJECTED", patch_response.json()["evidence"]["review_status"])
def test_sufficient_evidence_maps_claims_and_allows_draft(self) -> None:
article_id = self._create_researched_article("sufficient evidence coverage")
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)
body = evidence_response.json()
self.assertEqual("RESEARCH_RUNNING", body["article"]["status"])
self.assertFalse(body["insufficient_evidence_reasons"])
self.assertTrue(body["evidence"])
self.assertTrue(body["claims"])
self.assertTrue(
all(claim["evidence_item_ids"] for claim in body["claims"])
)
first_evidence = body["evidence"][0]
self.assertTrue(first_evidence["source_url"])
self.assertTrue(first_evidence["source_title"])
self.assertTrue(first_evidence["source_type"])
self.assertTrue(first_evidence["summary"])
self.assertGreaterEqual(first_evidence["source_quality_score"], 0)
self.assertTrue(first_evidence["retrieved_at"])
self.assertTrue(first_evidence["artifact_manifest_id"])
add_response = self.client.post(
f"/api/articles/{article_id}/evidence",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
json={
"source_title": "Manual source",
"source_url": "https://manual.example/source",
"source_type": "manual",
"summary": "Manual evidence note",
"source_quality_score": 0.74,
},
)
self.assertEqual(201, add_response.status_code, add_response.text)
manual_evidence = add_response.json()["evidence"]
self.assertEqual("PENDING", manual_evidence["review_status"])
delete_response = self.client.delete(
f"/api/articles/{article_id}/evidence/{manual_evidence['id']}",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
self.assertEqual(200, delete_response.status_code, delete_response.text)
refreshed = self.client.get(
f"/api/articles/{article_id}/evidence",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
).json()
self.assertNotIn(
manual_evidence["id"],
[item["id"] for item in refreshed["evidence"]],
)
draft_response = self.client.post(
f"/api/articles/{article_id}/draft/start",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
self.assertEqual(202, draft_response.status_code, draft_response.text)
def _create_researched_article(self, brief_suffix: str) -> str:
site = self.client.get(
"/api/sites",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
).json()[0]["site"]
article_id = self.client.post(
"/api/articles",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
json={
"target_site_id": site["id"],
"brief_description": f"Build an article with {brief_suffix}.",
"working_title": "Evidence Matrix Article",
"content_type": "longform_guide",
"primary_keyword": brief_suffix,
},
).json()["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"]:
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.client.post(
f"/api/articles/{article_id}/boundary-questions/submit",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
plan = self.client.post(
f"/api/articles/{article_id}/plan/generate",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
).json()["plan"]
self.client.post(
f"/api/articles/{article_id}/plans/{plan['id']}/approve",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
self.client.post(
f"/api/articles/{article_id}/research/start",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
return article_id
if __name__ == "__main__":
unittest.main()
apps/backend/tests/integration/test_schema_storage_contracts.py
+4
View File
@@ -120,6 +120,10 @@ class SchemaStorageContractsIntegrationTest(unittest.TestCase):
"artifacts",
}.issubset(repository.schema.list_columns("research_run_manifests"))
)
self.assertIn(
"review_status",
repository.schema.list_columns("evidence_items"),
)
self.assertTrue(
{
"repository_url",