ptarrant/chicken_babies_site
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: phase 4 admin CMS — dashboard, editor, media, CSRF

Browse Source 
Head Hen CMS end-to-end: dashboard lists all posts (drafts + published),
Markdown editor with live preview + drag-drop image upload, Pillow media
pipeline re-encoding every upload to JPEG, post CRUD + publish toggle +
hard delete, About page edit, and double-submit CSRF cookie enforced on
every admin mutating endpoint (Phase 3's TODO markers resolved).

Slug auto-generated on create and server-locked once a post has been
published. Unpublish preserves `published_at` so re-publish keeps
original date ordering. Every admin write invalidates the read-side
Post/Page TTL caches and records an `auth_events` audit row.

CSRF middleware is narrow by design — issues/refreshes the `cb_csrf`
cookie only on `GET /admin*`, and mutating endpoints opt in via
`require_csrf_form` or `require_csrf_header` Depends. Public routes,
healthz, and pre-auth login stay untouched.

64 new tests cover slugs, CSRF, media, admin posts/pages services, and
end-to-end CMS routes. Tests never mock the DB — real temp SQLite files
per the CLAUDE.md mandate.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Phillip Tarrant
2026-04-21 20:42:01 -05:00
parent 76875a455e
commit 9a8506970c
30 changed files with 3831 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

411
tests/test_admin_cms_routes.py Normal file
View File

@@ -0,0 +1,411 @@
"""End-to-end HTTP tests for the Phase 4 admin CMS.
Exercises the real :class:`FastAPI` app constructed by
:func:`app.main.create_app` against a temp-file SQLite database plus
a temp media directory. Uses the email-capture trick from
``tests/test_admin_routes.py`` to complete the magic-link login so we
have a real authenticated session + CSRF cookie.
"""
from __future__ import annotations
import importlib
import io
import re
from pathlib import Path
from typing import Iterator
import pytest
from fastapi.testclient import TestClient
from PIL import Image
from sqlalchemy import text
_CSRF_META_RE = re.compile(
r'<meta name="csrf-token" content="([^"]*)"'
)
@pytest.fixture
def cms_app(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> Iterator[tuple[TestClient, dict, Path]]:
"""Build a fresh authed app + temp media dir + captured email URL."""
media_root = tmp_path / "media"
media_root.mkdir()
monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path}/cms.db")
monkeypatch.setenv("ADMIN_EMAILS", "headhen@example.com")
monkeypatch.setenv("APP_ENV", "development")
monkeypatch.setenv(
"SECRET_KEY", "test-only-secret-key-0123456789abcdef-XYZ"
)
monkeypatch.setenv("RESEND_API_KEY", "")
monkeypatch.setenv("PUBLIC_BASE_URL", "http://testserver")
monkeypatch.setenv("MEDIA_ROOT", str(media_root))
from app import config as _config
_config.get_settings.cache_clear()
import app.main as main_module
importlib.reload(main_module)
app = main_module.app
captured: dict[str, str] = {}
def _capture(**kw):
captured["url"] = kw["url"]
captured["to"] = kw["to"]
app.state.email_service.send_magic_link = _capture # type: ignore[assignment]
from app.services.rate_limit import limiter
limiter.reset()
with TestClient(app) as client:
yield client, captured, media_root
_config.get_settings.cache_clear()
def _login(client: TestClient, captured: dict) -> None:
"""Walk the magic-link flow so subsequent requests are authed."""
client.post("/admin/login", data={"email": "headhen@example.com"})
assert "url" in captured
consume_path = captured["url"].replace("http://testserver", "")
resp = client.get(consume_path, follow_redirects=False)
assert resp.status_code == 303
def _dashboard_csrf(client: TestClient) -> str:
"""GET /admin and extract the csrf-token meta tag."""
resp = client.get("/admin")
assert resp.status_code == 200
match = _CSRF_META_RE.search(resp.text)
assert match, "csrf-token meta tag missing"
return match.group(1)
# ---------------------------------------------------------------------------
# unauthed
# ---------------------------------------------------------------------------
def test_dashboard_redirects_unauthed_to_login(cms_app) -> None:
"""An unauthenticated GET /admin yields 303 to /admin/login."""
client, _, _ = cms_app
resp = client.get("/admin", follow_redirects=False)
assert resp.status_code == 303
assert resp.headers["location"] == "/admin/login"
def test_post_create_without_auth_is_redirected(cms_app) -> None:
"""Create POST is gated behind require_admin."""
client, _, _ = cms_app
resp = client.post(
"/admin/posts",
data={"title": "x", "body_md": "y", "status": "draft", "csrf_token": ""},
follow_redirects=False,
)
# Unauthenticated -> require_admin raises 303.
assert resp.status_code == 303
# ---------------------------------------------------------------------------
# dashboard
# ---------------------------------------------------------------------------
def test_dashboard_renders_with_seeded_post(cms_app) -> None:
"""The seeded welcome post shows up in the admin table."""
client, captured, _ = cms_app
_login(client, captured)
resp = client.get("/admin")
assert resp.status_code == 200
assert "Dashboard" in resp.text
assert "Welcome to the Farm" in resp.text
assert "New post" in resp.text
assert "Edit About" in resp.text
# ---------------------------------------------------------------------------
# create + edit + delete
# ---------------------------------------------------------------------------
def test_create_post_happy_path(cms_app) -> None:
"""POST /admin/posts with CSRF creates a row and 303s to the dashboard."""
client, captured, _ = cms_app
_login(client, captured)
csrf = _dashboard_csrf(client)
resp = client.post(
"/admin/posts",
data={
"title": "First Admin Post",
"body_md": "Hello **world**.",
"status": "draft",
"csrf_token": csrf,
},
follow_redirects=False,
)
assert resp.status_code == 303
assert resp.headers["location"].startswith("/admin?msg=created")
# The new post is visible on the dashboard.
dash = client.get("/admin")
assert "First Admin Post" in dash.text
def test_create_post_without_csrf_is_forbidden(cms_app) -> None:
"""Missing csrf_token on create returns 403."""
client, captured, _ = cms_app
_login(client, captured)
# Touch /admin to materialize the cb_csrf cookie.
client.get("/admin")
resp = client.post(
"/admin/posts",
data={
"title": "Sneaky",
"body_md": "x",
"status": "draft",
"csrf_token": "", # wrong
},
follow_redirects=False,
)
assert resp.status_code == 403
def test_update_and_publish_and_delete(cms_app) -> None:
"""Full CRUD + publish toggle + delete cycle."""
client, captured, _ = cms_app
_login(client, captured)
csrf = _dashboard_csrf(client)
# create
client.post(
"/admin/posts",
data={
"title": "Toggle Target",
"body_md": "body-v1",
"status": "draft",
"csrf_token": csrf,
},
)
# look up the post id via the engine
with client.app.state.engine.connect() as conn:
row = conn.execute(
text("SELECT id FROM posts WHERE title = :t"),
{"t": "Toggle Target"},
).mappings().first()
assert row is not None
post_id = int(row["id"])
# edit form renders
edit_resp = client.get(f"/admin/posts/{post_id}/edit")
assert edit_resp.status_code == 200
assert "Toggle Target" in edit_resp.text
# CSRF for subsequent POSTs should come off the edit page.
match = _CSRF_META_RE.search(edit_resp.text)
assert match
csrf = match.group(1)
# update
upd = client.post(
f"/admin/posts/{post_id}",
data={
"title": "Toggle Target Edited",
"body_md": "body-v2",
"csrf_token": csrf,
},
follow_redirects=False,
)
assert upd.status_code == 303
# publish
pub = client.post(
f"/admin/posts/{post_id}/publish",
data={"csrf_token": csrf},
follow_redirects=False,
)
assert pub.status_code == 303
assert pub.headers["location"] == "/admin?msg=published"
# unpublish
unpub = client.post(
f"/admin/posts/{post_id}/publish",
data={"csrf_token": csrf},
follow_redirects=False,
)
assert unpub.status_code == 303
assert unpub.headers["location"] == "/admin?msg=unpublished"
# delete
delete = client.post(
f"/admin/posts/{post_id}/delete",
data={"csrf_token": csrf},
follow_redirects=False,
)
assert delete.status_code == 303
assert delete.headers["location"] == "/admin?msg=deleted"
# row is gone
with client.app.state.engine.connect() as conn:
gone = conn.execute(
text("SELECT id FROM posts WHERE id = :i"),
{"i": post_id},
).first()
assert gone is None
def test_create_post_rejects_blank_title(cms_app) -> None:
"""Blank title re-renders the form at 400."""
client, captured, _ = cms_app
_login(client, captured)
csrf = _dashboard_csrf(client)
resp = client.post(
"/admin/posts",
data={
"title": " ",
"body_md": "x",
"status": "draft",
"csrf_token": csrf,
},
)
assert resp.status_code == 400
assert "Title is required" in resp.text
# ---------------------------------------------------------------------------
# About page
# ---------------------------------------------------------------------------
def test_about_update(cms_app) -> None:
"""Editing the About page updates the stored copy."""
client, captured, _ = cms_app
_login(client, captured)
form_resp = client.get("/admin/pages/about/edit")
assert form_resp.status_code == 200
match = _CSRF_META_RE.search(form_resp.text)
assert match
csrf = match.group(1)
resp = client.post(
"/admin/pages/about",
data={
"title": "Re-titled About",
"body_md": "Edited **copy**.",
"csrf_token": csrf,
},
follow_redirects=False,
)
assert resp.status_code == 303
assert resp.headers["location"] == "/admin?msg=saved"
# Public site reflects the change (cache was invalidated).
public = client.get("/about")
assert public.status_code == 200
assert "Re-titled About" in public.text
assert "<strong>copy</strong>" in public.text
# ---------------------------------------------------------------------------
# media upload
# ---------------------------------------------------------------------------
def _jpeg_bytes() -> bytes:
img = Image.new("RGB", (32, 32), "blue")
buf = io.BytesIO()
img.save(buf, format="JPEG")
return buf.getvalue()
def test_media_upload_happy_path(cms_app) -> None:
"""Valid JPEG upload with header CSRF returns JSON with a /media URL."""
client, captured, media_root = cms_app
_login(client, captured)
csrf = _dashboard_csrf(client)
data = _jpeg_bytes()
files = {"file": ("photo.jpg", data, "image/jpeg")}
resp = client.post(
"/admin/media/upload",
files=files,
headers={"X-CSRF-Token": csrf},
)
assert resp.status_code == 200, resp.text
payload = resp.json()
assert payload["url"].startswith("/media/")
assert payload["filename"].endswith(".jpg")
# The file actually exists under the media root.
stored = Path(media_root)
assert any(stored.rglob("*.jpg"))
def test_media_upload_missing_csrf_is_forbidden(cms_app) -> None:
"""Without the X-CSRF-Token header the upload is 403."""
client, captured, _ = cms_app
_login(client, captured)
# Warm cookie.
client.get("/admin")
files = {"file": ("x.jpg", _jpeg_bytes(), "image/jpeg")}
resp = client.post("/admin/media/upload", files=files)
assert resp.status_code == 403
def test_media_upload_rejects_non_image(cms_app) -> None:
"""Plain text rejected with a JSON error payload."""
client, captured, _ = cms_app
_login(client, captured)
csrf = _dashboard_csrf(client)
files = {"file": ("fake.jpg", b"not an image at all, really not", "image/jpeg")}
resp = client.post(
"/admin/media/upload",
files=files,
headers={"X-CSRF-Token": csrf},
)
assert resp.status_code == 400
assert "error" in resp.json()
# ---------------------------------------------------------------------------
# preview
# ---------------------------------------------------------------------------
def test_preview_renders_sanitized_html(cms_app) -> None:
"""Preview endpoint returns the same sanitized HTML as server-side."""
client, captured, _ = cms_app
_login(client, captured)
csrf = _dashboard_csrf(client)
resp = client.post(
"/admin/preview",
data={"markdown": "Hello **there**."},
headers={"X-CSRF-Token": csrf},
)
assert resp.status_code == 200
assert "<strong>there</strong>" in resp.text
def test_preview_strips_raw_html(cms_app) -> None:
"""Raw HTML in the markdown input is sanitized away."""
client, captured, _ = cms_app
_login(client, captured)
csrf = _dashboard_csrf(client)
resp = client.post(
"/admin/preview",
data={"markdown": "<script>alert(1)</script>\n\nOK."},
headers={"X-CSRF-Token": csrf},
)
assert resp.status_code == 200
assert "<script>" not in resp.text
def test_preview_without_csrf_is_forbidden(cms_app) -> None:
"""Preview endpoint enforces CSRF via header."""
client, captured, _ = cms_app
_login(client, captured)
client.get("/admin") # warm cookie
resp = client.post(
"/admin/preview",
data={"markdown": "hi"},
)
assert resp.status_code == 403