"""Unit tests: the frontend asset version token (phase 33, app/core/caching.py). Covers the four token paths: * git repo -> ``git rev-parse --short HEAD`` (stable, cached). * fallback -> 12-hex content hash; stable for an unchanged tree, flips on a size or mtime change once the per-process cache is cleared. * failure -> a ``.git`` present but git broken (missing / timeout / non-zero exit) falls back to the content hash without raising. * empty -> a missing or empty static dir yields ``"dev"``. Plus the asset-reference rewrite (``rewrite_asset_refs``) and the CachingMiddleware fallback branches (task 02), and the phase-54 conditional-request pins on a ``StaticFiles(html=True)``-backed app (known pages 200 on conditional requests and publish no validators; ``/assets/*`` and ``/api/*`` keep their conditional behavior). """ from __future__ import annotations import asyncio import os import re import subprocess import uuid from collections.abc import AsyncIterator, Iterator from pathlib import Path import pytest from fastapi import FastAPI, Request, Response from fastapi.responses import HTMLResponse, JSONResponse from fastapi.testclient import TestClient from sqlalchemy import text from sqlalchemy.orm import Session from starlette.responses import FileResponse from starlette.staticfiles import StaticFiles import app.core.caching as caching from app.config import Settings from app.core import theming from app.core.caching import asset_version, rewrite_asset_refs from app.core.security_headers import CSP from app.models import UiSettings TOKEN = "abc123" @pytest.fixture(autouse=True) def _clear_token_cache() -> Iterator[None]: # The token is lru_cached per process — every case starts from a clean # slate and never leaks its entry into the next case. asset_version.cache_clear() yield asset_version.cache_clear() def _git(repo: Path, *args: str) -> None: subprocess.run(["git", "-C", str(repo), *args], check=True, capture_output=True) def _make_git_repo(base: Path) -> tuple[Path, str]: """A repo with a committed ``frontend/`` file; returns (frontend, short SHA).""" repo = base / "proj" frontend = repo / "frontend" frontend.mkdir(parents=True) (frontend / "styles.css").write_text("body { margin: 0 }\n") _git(repo, "init", "-q") _git(repo, "add", "frontend") _git( repo, "-c", "user.name=test", "-c", "user.email=test@example.com", "-c", "commit.gpgsign=false", # the fixture commit never signs (env gpg) "commit", "-q", "-m", "init", ) proc = subprocess.run( ["git", "-C", str(repo), "rev-parse", "--short", "HEAD"], check=True, capture_output=True, text=True, ) return frontend, proc.stdout.strip() def _make_plain_frontend(tmp_path: Path) -> Path: frontend = tmp_path / "frontend" frontend.mkdir() (frontend / "app.js").write_text("console.log('hi')\n") (frontend / "styles.css").write_text("body { margin: 0 }\n") return frontend def test_git_repo_token_matches_short_sha_and_is_cached(tmp_path) -> None: frontend, short_sha = _make_git_repo(tmp_path) assert asset_version(str(frontend)) == short_sha # Second call: same value, served from the per-process cache. assert asset_version(str(frontend)) == short_sha assert asset_version.cache_info().hits >= 1 def test_default_static_dir_comes_from_settings(tmp_path, monkeypatch) -> None: frontend, short_sha = _make_git_repo(tmp_path) monkeypatch.setattr( caching, "get_settings", lambda: Settings(static_dir=str(frontend)) ) assert asset_version() == short_sha # no argument -> settings default def test_fallback_token_is_12_hex_and_stable_for_unchanged_tree(tmp_path) -> None: frontend = _make_plain_frontend(tmp_path) token = asset_version(str(frontend)) assert re.fullmatch(r"[0-9a-f]{12}", token) # Unchanged tree: same token even after the cache entry is dropped # (i.e. the hash itself is deterministic, not just the cache). asset_version.cache_clear() assert asset_version(str(frontend)) == token def test_fallback_token_changes_on_size_change(tmp_path) -> None: frontend = _make_plain_frontend(tmp_path) before = asset_version(str(frontend)) (frontend / "app.js").write_text("console.log('a longer payload')\n") asset_version.cache_clear() assert asset_version(str(frontend)) != before def test_fallback_token_changes_on_mtime_only(tmp_path) -> None: frontend = _make_plain_frontend(tmp_path) before = asset_version(str(frontend)) path = frontend / "app.js" now = os.stat(path).st_mtime_ns os.utime(path, ns=(now + 5_000, now + 5_000)) # mtime change, same size asset_version.cache_clear() assert asset_version(str(frontend)) != before @pytest.mark.parametrize( ("error",), [ (FileNotFoundError("git not found"),), (subprocess.TimeoutExpired(cmd="git", timeout=5),), (subprocess.CalledProcessError(returncode=128, cmd="git"),), ], ids=["git-missing", "git-timeout", "git-nonzero-exit"], ) def test_git_failure_falls_back_to_content_hash(tmp_path, monkeypatch, error: Exception) -> None: # A ``.git`` exists, but git itself is broken -> content hash, no raise. repo = tmp_path / "proj" frontend = repo / "frontend" frontend.mkdir(parents=True) (repo / ".git").mkdir() (frontend / "styles.css").write_text("body { margin: 0 }\n") def _boom(*args: object, **kwargs: object) -> subprocess.CompletedProcess[str]: raise error monkeypatch.setattr(caching.subprocess, "run", _boom) token = asset_version(str(frontend)) assert re.fullmatch(r"[0-9a-f]{12}", token) def test_git_success_with_empty_output_falls_back(tmp_path, monkeypatch) -> None: # git exits 0 but prints nothing (defensive guard) -> content hash. repo = tmp_path / "proj" frontend = repo / "frontend" frontend.mkdir(parents=True) (repo / ".git").mkdir() (frontend / "styles.css").write_text("body { margin: 0 }\n") monkeypatch.setattr( caching.subprocess, "run", lambda *a, **k: subprocess.CompletedProcess( args=["git"], returncode=0, stdout=" \n", stderr="" ), ) token = asset_version(str(frontend)) assert re.fullmatch(r"[0-9a-f]{12}", token) def test_missing_static_dir_is_dev(tmp_path) -> None: assert asset_version(str(tmp_path / "does-not-exist")) == "dev" def test_empty_static_dir_is_dev(tmp_path) -> None: empty = tmp_path / "frontend" empty.mkdir() assert asset_version(str(empty)) == "dev" # --------------------------------------------------------------------------- # HTML_PAGES registration # --------------------------------------------------------------------------- def test_html_pages_include_history() -> None: """Phase 50: the History page is registered in HTML_PAGES — without this entry it would serve unversioned asset refs, which the immutable-for-a-year asset caching would pin to stale CSS after a deploy (the phase-35 git-sources lesson). The entry is ADDED — every pre-phase-50 page stays registered.""" for path in ( "/", "/index.html", "/sources.html", "/document.html", "/login.html", "/tuning.html", "/git-sources.html", "/history.html", "/tokens.html", # phase 79 task 06: the admin tokens page (shell route) "/theme.html", # phase 91 task 04: the admin theme page (shell route) "/shared.html", # phase 51: the shared page's static path "/doc-edit.html", # phase 59: the doc edit screen (task 06) ): assert path in caching.HTML_PAGES, f"{path} must be in HTML_PAGES" # --------------------------------------------------------------------------- # rewrite_asset_refs (task 02) # --------------------------------------------------------------------------- def test_rewrite_versions_href_with_leading_slash() -> None: html = '' assert rewrite_asset_refs(html, TOKEN) == ( '' ) def test_rewrite_versions_src_without_leading_slash() -> None: html = '' assert rewrite_asset_refs(html, TOKEN) == ( '' ) def test_rewrite_versions_module_script_src() -> None: html = '' assert rewrite_asset_refs(html, TOKEN) == ( '' ) def test_rewrite_versions_every_ref_in_one_pass() -> None: html = ( '' "" '' ) assert rewrite_asset_refs(html, TOKEN) == ( '' '' '' ) def test_rewrite_is_idempotent_for_already_versioned_refs() -> None: html = '' assert rewrite_asset_refs(html, "deadbeef") == html def test_rewrite_leaves_query_and_fragment_refs_alone() -> None: html = '' assert rewrite_asset_refs(html, TOKEN) == html def test_rewrite_leaves_non_asset_refs_untouched() -> None: html = ( 'sources' '' 'login' ) assert rewrite_asset_refs(html, TOKEN) == html def test_rewrite_returns_html_unchanged_when_no_asset_refs() -> None: html = "

no assets here

" assert rewrite_asset_refs(html, TOKEN) == html # --------------------------------------------------------------------------- # CachingMiddleware fallback branches (task 02) # --------------------------------------------------------------------------- def _page_app() -> FastAPI: """A bare app with a ``text/html`` route at ``/`` + the middleware.""" app = FastAPI() @app.get("/", response_class=HTMLResponse) def index() -> str: return '' @app.get("/api/health") def health() -> JSONResponse: return JSONResponse({"status": "ok"}) caching.configure_caching(app) return app def test_middleware_success_path_rewrites_page_and_sets_no_cache() -> None: client = TestClient(_page_app()) r = client.get("/") assert r.status_code == 200 assert r.headers["cache-control"] == "no-cache" token = caching.asset_version() assert f'href="/assets/styles.css?v={token}"' in r.text assert 'href="/assets/styles.css">' not in r.text def test_middleware_buffer_failure_keeps_body_and_sets_no_cache(monkeypatch) -> None: """If buffering/token resolution fails, the ORIGINAL streaming body is served unmodified — with ``no-cache`` — never an empty page.""" def _boom() -> str: raise RuntimeError("token blew up") monkeypatch.setattr(caching, "asset_version", _boom) client = TestClient(_page_app()) r = client.get("/") assert r.status_code == 200 assert r.headers["cache-control"] == "no-cache" assert r.text == ( "" ) def test_middleware_non_html_body_on_page_path_gets_no_cache_only() -> None: """A page path whose response is not ``text/html`` (e.g. the 404 JSON when the static dir is missing) is revalidated but never rewritten.""" app = FastAPI() @app.get("/") def not_found() -> JSONResponse: return JSONResponse({"detail": "Not Found"}, status_code=404) caching.configure_caching(app) client = TestClient(app) r = client.get("/") assert r.status_code == 404 assert r.headers["cache-control"] == "no-cache" assert r.json() == {"detail": "Not Found"} # body untouched def test_middleware_rewrite_failure_reserves_original_bytes(monkeypatch) -> None: """If the body buffered fine but the rewrite itself fails, the ORIGINAL bytes are re-served (never an empty page) with ``no-cache``.""" def _boom(html: str, token: str) -> str: raise ValueError("rewrite blew up") monkeypatch.setattr(caching, "rewrite_asset_refs", _boom) client = TestClient(_page_app()) r = client.get("/") assert r.status_code == 200 assert r.headers["cache-control"] == "no-cache" assert r.text == ( "" ) def test_middleware_leaves_api_responses_byte_identical() -> None: """``/api/*`` gets no injected headers at all (no cache-control).""" client = TestClient(_page_app()) r = client.get("/api/health") assert r.status_code == 200 assert "cache-control" not in r.headers assert r.json() == {"status": "ok"} def test_read_body_buffers_plain_response_bytes() -> None: from starlette.responses import Response resp = Response(content=b"hi", media_type="text/html") assert asyncio.run(caching._read_body(resp)) == b"hi" def test_read_body_copies_plain_response_memoryview() -> None: from starlette.responses import Response resp = Response(content=memoryview(b"hi"), media_type="text/html") assert asyncio.run(caching._read_body(resp)) == b"hi" def test_read_body_drains_streaming_response() -> None: from starlette.responses import StreamingResponse async def gen() -> AsyncIterator[bytes]: yield b"" yield b"" resp = StreamingResponse(content=gen(), media_type="text/html") assert asyncio.run(caching._read_body(resp)) == b"" # --------------------------------------------------------------------------- # Phase 51: the dynamic share page — the ``/shared/`` prefix contract # --------------------------------------------------------------------------- def _shared_page_app() -> FastAPI: """A bare app with the phase-51 route pair + the middleware: ``/shared/`` (``text/html`` — the page the route serves) and ``/api/shared/`` (the JSON read), plus an unknown path.""" app = FastAPI() @app.get("/shared/{token}", response_class=HTMLResponse) def shared_page(token: str) -> str: return ( "" '' "shared" ) @app.get("/api/shared/{token}") def shared_api(token: str) -> JSONResponse: return JSONResponse({"title": "Shared", "messages": []}) @app.get("/some/unknown/path") def unknown() -> JSONResponse: return JSONResponse({"ok": True}) caching.configure_caching(app) return app def test_middleware_treats_shared_page_path_as_known_html_page() -> None: """Phase 51: ``/shared/`` joins the HTML_PAGES contract — ``no-cache`` + ``?v=`` asset rewrite on the ``text/html`` body (the FileResponse body is drained by the existing ``_read_body`` path).""" client = TestClient(_shared_page_app()) r = client.get(f"/shared/{uuid.uuid4()}") assert r.status_code == 200 assert r.headers["cache-control"] == "no-cache" token = caching.asset_version() assert f'href="/assets/styles.css?v={token}"' in r.text assert 'href="/assets/styles.css">' not in r.text def test_middleware_leaves_api_shared_read_untouched() -> None: """``/api/shared/`` — the JSON read — starts with ``/api/``, not ``/shared/``: byte-identical pass-through, no injected headers.""" client = TestClient(_shared_page_app()) r = client.get(f"/api/shared/{uuid.uuid4()}") assert r.status_code == 200 assert "cache-control" not in r.headers assert r.json() == {"title": "Shared", "messages": []} def test_middleware_leaves_unknown_paths_untouched() -> None: """A path that is neither a known page, ``/shared/*``, nor ``/assets/*`` passes through byte-identical, no headers.""" client = TestClient(_shared_page_app()) r = client.get("/some/unknown/path") assert r.status_code == 200 assert "cache-control" not in r.headers assert r.json() == {"ok": True} # --------------------------------------------------------------------------- # Phase 54: conditional requests on page paths (StaticFiles-backed app) # --------------------------------------------------------------------------- # # The 304 hole: the validators published for an HTML page describe the # static FILE, but the bytes the browser receives are the REWRITTEN body # this process built from its own token. Starlette's ``StaticFiles`` # honours ``If-None-Match`` / ``If-Modified-Since`` *before* the middleware # can see a response, so a matching conditional request used to 304 out of # the rewrite — the browser kept HTML whose ``?v=`` pinned the previous # commit's immutable assets. The middleware now strips those headers on # page paths (inbound) and drops the validators (outbound); ``/assets/*`` # and ``/api/*`` keep their conditional behavior. def _file_validators(path: Path) -> dict[str, str]: """The ``etag`` / ``last-modified`` the StaticFiles mount will publish for ``path`` — derived straight from the file. The post-fix page response publishes no validators to capture, so the browser-side etag is reconstructed the way the mount computes it (starlette 1.x's ``FileResponse`` defers the stat to ``__call__`` unless ``stat_result`` is passed).""" file_response = FileResponse(path, stat_result=os.stat(path)) return { "etag": file_response.headers["etag"], "last-modified": file_response.headers["last-modified"], } def _static_page_app( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> tuple[TestClient, Path]: """A ``StaticFiles(html=True)``-backed app + the middleware — the same mounting shape as ``app/main.py`` (routes + the catch-all static mount) — so conditional requests are exercised against REAL ``FileResponse``/``StaticFiles`` validators, the exact layer that produced the pre-fix 304. Returns ``(TestClient(app), frontend root)``. The version token is pinned to ``"unit-token"`` so the assertions are exact: ``asset_version`` is the module-level call the middleware makes, and its ``functools.cache``d result can only be a git short SHA, a 12-hex content hash, or ``"dev"`` — so it is monkeypatched directly. The ``get_settings`` stub (mirroring ``test_default_static_dir_comes_from_settings``) steers any unpatched token computation to the tmp tree instead of the checkout. """ frontend = tmp_path / "frontend" assets = frontend / "assets" assets.mkdir(parents=True) (frontend / "index.html").write_text( "" '' '' "static index" ) (assets / "app.js").write_text("x") (assets / "app.css").write_text("y") monkeypatch.setattr( caching, "get_settings", lambda: Settings(static_dir=str(frontend)) ) monkeypatch.setattr(caching, "asset_version", lambda: "unit-token") app = FastAPI() @app.get("/api/ping") def ping(request: Request) -> Response: # Honours a conditional header, exactly like StaticFiles does — # proves the inbound strip does NOT widen to /api/*. if "if-none-match" in request.headers: return Response(status_code=304) return JSONResponse({"status": "ok"}) @app.get("/shared/{token}") def shared_page(request: Request, token: str) -> Response: # Like the phase-51 route: serves text/html AND honours a # conditional header — without the inbound strip this route would # 304 a conditional GET, so the strip is observable here. if "if-none-match" in request.headers: return Response(status_code=304) return HTMLResponse( "" '' "shared" ) app.mount("/", StaticFiles(directory=frontend, html=True), name="static") caching.configure_caching(app) return TestClient(app), frontend def test_conditional_get_page_returns_200_with_rewrite( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """THE phase-54 regression: a conditional GET / that matches the static file's etag used to 304 out of the rewrite (StaticFiles honours If-None-Match before the middleware can see a body), leaving the browser on HTML whose ``?v=`` pinned the previous commit's immutable assets. Post-fix the same request gets a full 200 with the current ``?v=`` refs — and the page publishes no validators. (Fails with a 304 on the pre-fix code — the defect reproduction.)""" client, index_html = _static_page_app(tmp_path, monkeypatch) first = client.get("/") assert first.status_code == 200 assert "?v=unit-token" in first.text assert "etag" not in first.headers # the fix drops it # Derive the *file* validator the way a browser would have captured # it — the post-fix page response publishes no etag to capture. etag = _file_validators(index_html)["etag"] r = client.get("/", headers={"if-none-match": etag}) assert r.status_code == 200 # never 304 on a page path assert 'href="/assets/app.js?v=unit-token"' in r.text assert 'src="/assets/app.js?v=unit-token"' in r.text assert r.headers["cache-control"] == "no-cache" assert "etag" not in r.headers assert "last-modified" not in r.headers def test_conditional_get_page_with_if_modified_since_returns_200( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """Same pin via If-Modified-Since: round-tripping the mount's last-modified used to 304 (pre-fix); the inbound strip now guarantees the full rewritten 200.""" client, index_html = _static_page_app(tmp_path, monkeypatch) last_modified = _file_validators(index_html)["last-modified"] r = client.get("/", headers={"if-modified-since": last_modified}) assert r.status_code == 200 # never 304 on a page path assert 'href="/assets/app.js?v=unit-token"' in r.text assert r.headers["cache-control"] == "no-cache" assert "etag" not in r.headers assert "last-modified" not in r.headers def test_page_response_publishes_no_validators( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """Every known page (the static mount's ``/`` and the dynamic ``/shared/``) publishes NO validators — combined with ``Cache-Control: no-cache`` the browser can never revalidate a page against a validator this process published.""" client, _ = _static_page_app(tmp_path, monkeypatch) for path in ("/", "/shared/abc123"): r = client.get(path) assert r.status_code == 200 assert r.headers["cache-control"] == "no-cache" assert "etag" not in r.headers assert "last-modified" not in r.headers @pytest.mark.parametrize("status_code", [304, 204], ids=["304", "204"]) def test_downstream_bodiless_status_on_page_is_passed_bodiless(status_code: int) -> None: """A downstream that still returns a bodiless status on a page path is passed through BODIESS — starlette forbids a body on 204/304, and the pre-guard rewrite path would build exactly that (``Response(content=…, status_code=304)``). Belt-and-braces: after the inbound strip StaticFiles cannot 304 page paths, but a future route or proxy could — the passthrough keeps no-cache and no validators.""" app = FastAPI() @app.get("/") def index(request: Request) -> Response: # The inbound strip already removed if-none-match / # if-modified-since from the scope, so a conditional route can no # longer 304 a page path — trigger the bodiless status with a # header the strip does not touch, and publish validators to pin # the outbound drop. if request.headers.get("x-bodiless") == "1": return Response( status_code=status_code, headers={ "etag": '"stale"', "last-modified": "Wed, 01 Jan 2024 00:00:00 GMT", }, ) return HTMLResponse( "" '' "" ) caching.configure_caching(app) client = TestClient(app) # The plain path still gets the full rewritten page (the guard only # kicks in for bodiless statuses). full = client.get("/") assert full.status_code == 200 assert 'href="/assets/styles.css?v=' in full.text assert full.headers["cache-control"] == "no-cache" r = client.get("/", headers={"x-bodiless": "1", "if-none-match": "whatever"}) assert r.status_code == status_code assert r.content == b"" # never a body on a bodiless status assert r.headers["cache-control"] == "no-cache" assert "etag" not in r.headers assert "last-modified" not in r.headers def test_shared_page_path_ignores_conditional_headers( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """The dynamic /shared/ page (phase 51) gets the same inbound strip: the conditional route in ``_static_page_app`` would 304 if it saw if-none-match — the strip hides it, so the page always 200s with the current ``?v=`` refs.""" client, _ = _static_page_app(tmp_path, monkeypatch) r = client.get("/shared/abc123", headers={"if-none-match": "whatever"}) assert r.status_code == 200 # the route would have 304'd pre-strip assert 'href="/assets/app.css?v=unit-token"' in r.text assert r.headers["cache-control"] == "no-cache" assert "etag" not in r.headers def test_api_path_keeps_conditional_headers( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """The strip must NOT widen to /api/*: /api/ping still sees if-none-match (it 304s on it, like any conditional route), and the middleware passes that 304 through byte-identical — no cache-control injected.""" client, _ = _static_page_app(tmp_path, monkeypatch) plain = client.get("/api/ping") assert plain.status_code == 200 assert plain.json() == {"status": "ok"} assert "cache-control" not in plain.headers r = client.get("/api/ping", headers={"if-none-match": "whatever"}) assert r.status_code == 304 # the route saw the header — no strip assert r.content == b"" assert "cache-control" not in r.headers def test_assets_path_keeps_validators_and_304( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """/assets/* keeps its conditional behavior completely intact: the immutable caching is header-only, the validators stay, and a 304 on a versioned asset URL is SAFE (the URL itself carries ``?v=``).""" client, _ = _static_page_app(tmp_path, monkeypatch) r = client.get("/assets/app.css?v=unit-token") assert r.status_code == 200 assert r.headers["cache-control"] == caching.ASSET_CACHE_CONTROL assert "etag" in r.headers assert "last-modified" in r.headers etag = r.headers["etag"] r304 = client.get("/assets/app.css?v=unit-token", headers={"if-none-match": etag}) assert r304.status_code == 304 # versioned-URL 304s stay safe assert r304.content == b"" assert r304.headers["cache-control"] == caching.ASSET_CACHE_CONTROL # --------------------------------------------------------------------------- # Phase 91 (task 02): the pre-paint inline theme tag # --------------------------------------------------------------------------- # # The middleware's rewrite branch now ALSO builds the theme tag from the # effective ``ui_settings`` row (task 01's resolver — one short-lived # session per response, no process cache) and inserts it before the # first ````. Unset/defaults → ``tag == ""`` → the served bytes # are EXACTLY the phase-33/54 rewrite-only output (B4's byte-identical # contract); a DB blip is the same no-op (the page never breaks). def _theme_page(name: str) -> str: """The ``text/html`` body the fixture routes below serve.""" return ( f"{name}" '' f"
{name}
" ) def _theme_page_app() -> FastAPI: """A bare app with the middleware: the shell page at ``/`` plus a second known page (``/document.html``) and the phase-51 dynamic ``/shared/`` route (the prefix branch) — every served ``text/html`` with one versionable asset ref.""" app = FastAPI() @app.get("/", response_class=HTMLResponse) def index() -> str: return _theme_page("index") @app.get("/document.html", response_class=HTMLResponse) def document() -> str: return _theme_page("document") @app.get("/shared/{token}", response_class=HTMLResponse) def shared(token: str) -> str: return _theme_page(f"shared-{token}") caching.configure_caching(app) return app def test_middleware_unset_page_is_byte_identical_to_rewrite_only(db: Session) -> None: """THE byte-identical contract (B4): with NO ``ui_settings`` row the served body is EXACTLY the phase-33/54 rewrite-only output — not a single byte differs, no ``#bor-theme`` anywhere.""" db.execute(text("DELETE FROM ui_settings")) db.commit() client = TestClient(_theme_page_app()) token = caching.asset_version() for path, name in (("/", "index"), ("/document.html", "document")): r = client.get(path) assert r.status_code == 200 assert r.headers["cache-control"] == "no-cache" expected = caching.rewrite_asset_refs(_theme_page(name), token) assert r.content == expected.encode("utf-8") # byte-identical assert "bor-theme" not in r.text # No tag → no style-src exemption: the response carries no CSP # of its own (this bare app has no security-headers layer), so # the outer middleware's plain A1 string stands untouched. assert "content-security-policy" not in r.headers def test_middleware_themed_injects_tag_before_head_on_every_page(db: Session) -> None: """A ``ui_settings`` row with ONE changed color: every HTML page — ``/``, the non-shell ``/document.html``, and the dynamic ``/shared/`` (the prefix branch) — carries EXACTLY ONE ``', r.text) assert declared is not None names = re.findall(r"--([a-z-]+):", declared.group(1)) assert names == [k.replace("_", "-") for k in theming.COLOR_FIELDS] assert "--brand:#818cf8;" in r.text # Phase 91 (task 05): the inline tag is blocked by the # phase-82 CSP in a real browser unless this response also # carries the style-src exemption — the A1 string plus a # sha256 hash of the EXACT tag content (the current theme # is the only inline style ever permitted; no # 'unsafe-inline'). assert r.headers["content-security-policy"] == ( f"{CSP}; style-src 'self' '{theming.theme_csp_hash(tag)}'" ) assert "unsafe-inline" not in r.headers["content-security-policy"] # The phase-33/54 asset rewrite is untouched and applies too. assert f'href="/assets/styles.css?v={token}"' in r.text finally: db.execute(text("DELETE FROM ui_settings")) db.commit() def test_middleware_grid_only_change_tag_carries_grid_line(db: Session) -> None: """Phase 92 (task 01): the 9th identity variable — the OTHER 8 colors at built-in + ONLY ``grid_line`` set still breaks the no-op contract: the tag is NON-empty and carries ALL 9 vars (``--grid-line:`` with the changed value, the rest their built-ins, ``COLOR_FIELDS`` order) with the matching style-src CSP hash.""" db.execute(text("DELETE FROM ui_settings")) db.add(UiSettings(id=1, grid_line="#123123")) db.commit() try: colors = dict(theming.BUILTIN_COLORS) colors["grid_line"] = "#123123" # one changed color, rest built-in tag = theming.theme_style_tag(colors) assert tag != "" # the no-op contract holds ONLY for all-built-in assert "--grid-line:#123123;" in tag client = TestClient(_theme_page_app()) r = client.get("/") assert r.status_code == 200 assert r.text.count('id="bor-theme"') == 1 assert "\n" + tag + "" in r.text # All 9 vars, COLOR_FIELDS order (grid_line between line and brand). declared = re.search(r'', r.text) assert declared is not None names = re.findall(r"--([a-z-]+):", declared.group(1)) assert names == [k.replace("_", "-") for k in theming.COLOR_FIELDS] assert names.index("grid-line") == 5 assert r.headers["content-security-policy"] == ( f"{CSP}; style-src 'self' '{theming.theme_csp_hash(tag)}'" ) finally: db.execute(text("DELETE FROM ui_settings")) db.commit() @pytest.mark.parametrize( ("what",), [("session",), ("resolver",)], ids=["session-open-fails", "resolver-fails"], ) def test_middleware_db_failure_serves_page_without_tag( monkeypatch: pytest.MonkeyPatch, what: str ) -> None: """A DB blip must NEVER break the page (loadHealth house style): whether the session fails to open or the row read raises, the page still 200s with the byte-identical rewrite-only body (no tag) and the no-cache contract intact — a pre-migration boot is the same path.""" if what == "session": def _boom_session() -> object: raise RuntimeError("db down") monkeypatch.setattr(caching, "SessionLocal", _boom_session) else: def _boom_resolver(session: object) -> dict[str, str]: raise RuntimeError("select failed") monkeypatch.setattr(caching.theming, "effective_settings", _boom_resolver) client = TestClient(_theme_page_app()) r = client.get("/") assert r.status_code == 200 assert r.headers["cache-control"] == "no-cache" token = caching.asset_version() assert r.content == caching.rewrite_asset_refs( _theme_page("index"), token ).encode("utf-8") assert "bor-theme" not in r.text # The DB-failure fallback is the UNSET shape: no tag, no style-src # exemption (the plain A1 policy stands — the page degrades to the # built-in palette, never to an inline-style exemption for a tag # that is not there). assert "content-security-policy" not in r.headers