Task 004: add demo auth and role checks

This commit is contained in:
2026-05-21 18:31:40 +03:00
parent 7a11d50aa7
commit 7972e9a95f
26 changed files with 2207 additions and 23 deletions
@@ -0,0 +1,227 @@
from __future__ import annotations
import sys
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.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 AuthAuthorizationPublicApiTest(unittest.TestCase):
def setUp(self) -> None:
self.client = TestClient(app)
def test_me_returns_selected_demo_user_and_role(self) -> None:
response = self.client.get(
"/api/me",
headers={DEMO_USER_EMAIL_HEADER: DEMO_ADMIN_EMAIL},
)
self.assertEqual(200, response.status_code, response.text)
self.assertEqual(DEMO_ADMIN_EMAIL, response.json()["user"]["email"])
self.assertEqual("ADMIN", response.json()["user"]["role"])
def test_missing_and_unknown_demo_users_are_rejected_consistently(self) -> None:
missing_response = self.client.get("/api/me")
unknown_response = self.client.get(
"/api/me",
headers={DEMO_USER_EMAIL_HEADER: "missing@example.com"},
)
self.assertEqual(401, missing_response.status_code, missing_response.text)
self.assertEqual(401, unknown_response.status_code, unknown_response.text)
self.assertEqual({"detail": "Unauthorized"}, missing_response.json())
self.assertEqual({"detail": "Unauthorized"}, unknown_response.json())
def test_editor_cannot_create_target_site_config(self) -> None:
response = self.client.post(
"/api/sites",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
json=self._site_payload("editorial_ops_blog"),
)
self.assertEqual(403, response.status_code, response.text)
self.assertEqual({"detail": "Forbidden"}, response.json())
def test_admin_can_create_and_update_target_site_config(self) -> None:
create_response = self.client.post(
"/api/sites",
headers={DEMO_USER_EMAIL_HEADER: DEMO_ADMIN_EMAIL},
json=self._site_payload("admin_ops_blog"),
)
self.assertEqual(201, create_response.status_code, create_response.text)
site = create_response.json()["site"]
self.assertEqual("admin_ops_blog", site["slug"])
update_response = self.client.patch(
f"/api/sites/{site['id']}",
headers={DEMO_USER_EMAIL_HEADER: DEMO_ADMIN_EMAIL},
json={"brand_voice": "Updated practical editorial operations guidance."},
)
self.assertEqual(200, update_response.status_code, update_response.text)
self.assertEqual(
"Updated practical editorial operations guidance.",
update_response.json()["site"]["brand_voice"],
)
def test_editor_cannot_update_site_or_mutate_script_versions(self) -> None:
create_response = self.client.post(
"/api/sites",
headers={DEMO_USER_EMAIL_HEADER: DEMO_ADMIN_EMAIL},
json=self._site_payload("script_editor_denied_blog"),
)
self.assertEqual(201, create_response.status_code, create_response.text)
site_id = create_response.json()["site"]["id"]
version_response = self.client.post(
f"/api/sites/{site_id}/publishing-config/versions",
headers={DEMO_USER_EMAIL_HEADER: DEMO_ADMIN_EMAIL},
json=self._script_version_payload(),
)
self.assertEqual(201, version_response.status_code, version_response.text)
version_id = version_response.json()["version"]["id"]
site_update_response = self.client.patch(
f"/api/sites/{site_id}",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
json={"brand_voice": "Editor should not mutate Admin config."},
)
version_create_response = self.client.post(
f"/api/sites/{site_id}/publishing-config/versions",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
json=self._script_version_payload(),
)
version_activate_response = self.client.post(
f"/api/sites/{site_id}/publishing-config/versions/{version_id}/activate",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
self.assertEqual(403, site_update_response.status_code, site_update_response.text)
self.assertEqual(
403,
version_create_response.status_code,
version_create_response.text,
)
self.assertEqual(
403,
version_activate_response.status_code,
version_activate_response.text,
)
def test_admin_can_create_and_activate_script_config_version(self) -> None:
site_response = self.client.post(
"/api/sites",
headers={DEMO_USER_EMAIL_HEADER: DEMO_ADMIN_EMAIL},
json=self._site_payload("script_admin_blog"),
)
self.assertEqual(201, site_response.status_code, site_response.text)
site_id = site_response.json()["site"]["id"]
create_response = self.client.post(
f"/api/sites/{site_id}/publishing-config/versions",
headers={DEMO_USER_EMAIL_HEADER: DEMO_ADMIN_EMAIL},
json=self._script_version_payload(),
)
self.assertEqual(201, create_response.status_code, create_response.text)
self.assertEqual("DRAFT", create_response.json()["version"]["status"])
version_id = create_response.json()["version"]["id"]
activate_response = self.client.post(
f"/api/sites/{site_id}/publishing-config/versions/{version_id}/activate",
headers={DEMO_USER_EMAIL_HEADER: DEMO_ADMIN_EMAIL},
)
self.assertEqual(200, activate_response.status_code, activate_response.text)
self.assertEqual("ACTIVE", activate_response.json()["version"]["status"])
def test_editor_can_create_articles_and_approve_plan_review(self) -> None:
sites_response = self.client.get(
"/api/sites",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
self.assertEqual(200, sites_response.status_code, sites_response.text)
target_site_id = sites_response.json()[0]["site"]["id"]
article_response = self.client.post(
"/api/articles",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
json={
"target_site_id": target_site_id,
"brief_description": "Write about editorial AI workflow controls.",
"working_title": "Editorial AI Workflow Controls",
},
)
self.assertEqual(201, article_response.status_code, article_response.text)
self.assertEqual(
"ARTICLE_BRIEF_CREATED",
article_response.json()["article"]["status"],
)
article_id = article_response.json()["article"]["id"]
review_response = self.client.post(
f"/api/articles/{article_id}/plans/00000000-0000-0000-0000-000000000123/approve",
headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
)
self.assertEqual(200, review_response.status_code, review_response.text)
self.assertEqual("PLAN", review_response.json()["review"]["review_type"])
self.assertEqual("APPROVED", review_response.json()["review"]["status"])
def test_unauthorized_requests_are_rejected_for_protected_mutations(self) -> None:
response = self.client.post(
"/api/articles",
json={
"target_site_id": "00000000-0000-0000-0000-000000000001",
"brief_description": "No user selected.",
},
)
self.assertEqual(401, response.status_code, response.text)
self.assertEqual({"detail": "Unauthorized"}, response.json())
def _site_payload(self, slug: str) -> dict[str, object]:
return {
"name": "Editorial Ops Blog",
"slug": slug,
"publishing_type": "git_next",
"default_language": "en",
"brand_voice": "Practical editorial operations guidance.",
"audience": "Editorial teams running AI-assisted workflows.",
"seo_rules": {"primary_keyword_required": True},
"visual_rules": {"hero_style": "product editorial"},
"source_rules": {"minimum_sources": 3},
"publishing_rules": {
"repository_url": "git@github.com:example/editorial-ops.git",
"production_branch": "main",
"content_format": "mdx",
"content_path_template": "content/articles/{slug}.mdx",
"asset_path_template": "public/articles/{slug}/{filename}",
"frontmatter_mapping": {"title": "title"},
"dry_run_renderer": "next-mdx",
},
}
def _script_version_payload(self) -> dict[str, object]:
return {
"publishing_yaml": "target: editorial_ops\nrepository:\n branch: main\n",
"transform_script": "export function transformArticle(article) { return article; }\n",
"diff": {"summary": "Task 004 authorization fixture."},
}
if __name__ == "__main__":
unittest.main()