test_static_assets.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Section 6.3 — Static asset caching and compression tests. |
| 2 | |
| 3 | Covers: |
| 4 | - StaticCacheMiddleware injects correct Cache-Control headers |
| 5 | - Vendor JS files include ?v= query strings in base.html |
| 6 | - static_version is a non-empty 8-char hex string (content hash) |
| 7 | - Blocking sync I/O in objects.py is wrapped in asyncio.to_thread |
| 8 | - nginx gzip config present in nginx-cf.conf |
| 9 | """ |
| 10 | from __future__ import annotations |
| 11 | |
| 12 | import ast |
| 13 | import inspect |
| 14 | import re |
| 15 | from pathlib import Path |
| 16 | |
| 17 | import pytest |
| 18 | from muse.core.types import blob_id, split_id |
| 19 | from starlette.types import Receive, Scope, Send |
| 20 | from musehub.types.json_types import JSONObject |
| 21 | |
| 22 | |
| 23 | # ── paths ───────────────────────────────────────────────────────────────────── |
| 24 | _REPO = Path(__file__).resolve().parents[1] |
| 25 | _STATIC_DIR = _REPO / "musehub" / "templates" / "musehub" / "static" |
| 26 | _MIDDLEWARE = _REPO / "musehub" / "middleware" / "static_cache.py" |
| 27 | _TEMPLATES_PY = _REPO / "musehub" / "api" / "routes" / "musehub" / "_templates.py" |
| 28 | _BASE_HTML = _REPO / "musehub" / "templates" / "musehub" / "base.html" |
| 29 | _OBJECTS_PY = _REPO / "musehub" / "api" / "routes" / "musehub" / "objects.py" |
| 30 | _NGINX_CONF = _REPO / "deploy" / "nginx-cf.conf" |
| 31 | _MAIN_PY = _REPO / "musehub" / "main.py" |
| 32 | |
| 33 | |
| 34 | # ── StaticCacheMiddleware unit tests ────────────────────────────────────────── |
| 35 | |
| 36 | class TestStaticCacheMiddleware: |
| 37 | """Unit-test the pure ASGI middleware without spinning up FastAPI.""" |
| 38 | |
| 39 | def _make_scope(self, path: str) -> JSONObject: |
| 40 | return {"type": "http", "path": path, "method": "GET", "headers": []} |
| 41 | |
| 42 | async def _run(self, path: str) -> list[dict]: |
| 43 | from musehub.middleware.static_cache import StaticCacheMiddleware |
| 44 | |
| 45 | messages: list[dict] = [] |
| 46 | |
| 47 | async def _app(scope: Scope, receive: Receive, send: Send) -> None: |
| 48 | await send({ |
| 49 | "type": "http.response.start", |
| 50 | "status": 200, |
| 51 | "headers": [], |
| 52 | }) |
| 53 | await send({"type": "http.response.body", "body": b"", "more_body": False}) |
| 54 | |
| 55 | async def _capture(msg: JSONObject) -> None: |
| 56 | messages.append(msg) |
| 57 | |
| 58 | async def _noop_receive() -> None: |
| 59 | return {} |
| 60 | |
| 61 | mw = StaticCacheMiddleware(_app) |
| 62 | await mw(self._make_scope(path), _noop_receive, _capture) |
| 63 | return messages |
| 64 | |
| 65 | @pytest.mark.asyncio |
| 66 | async def test_css_gets_far_future_cache(self) -> None: |
| 67 | # ASGI scope["path"] never includes query string |
| 68 | msgs = await self._run("/static/app.css") |
| 69 | start = next(m for m in msgs if m["type"] == "http.response.start") |
| 70 | headers = dict(start["headers"]) |
| 71 | assert headers[b"cache-control"] == b"public, max-age=31536000, immutable" |
| 72 | |
| 73 | @pytest.mark.asyncio |
| 74 | async def test_js_gets_far_future_cache(self) -> None: |
| 75 | msgs = await self._run("/static/app.js") |
| 76 | start = next(m for m in msgs if m["type"] == "http.response.start") |
| 77 | headers = dict(start["headers"]) |
| 78 | assert headers[b"cache-control"] == b"public, max-age=31536000, immutable" |
| 79 | |
| 80 | @pytest.mark.asyncio |
| 81 | async def test_map_gets_far_future_cache(self) -> None: |
| 82 | msgs = await self._run("/static/app.js.map") |
| 83 | start = next(m for m in msgs if m["type"] == "http.response.start") |
| 84 | headers = dict(start["headers"]) |
| 85 | assert headers[b"cache-control"] == b"public, max-age=31536000, immutable" |
| 86 | |
| 87 | @pytest.mark.asyncio |
| 88 | async def test_favicon_gets_one_day_cache(self) -> None: |
| 89 | msgs = await self._run("/static/favicon.svg") |
| 90 | start = next(m for m in msgs if m["type"] == "http.response.start") |
| 91 | headers = dict(start["headers"]) |
| 92 | assert headers[b"cache-control"] == b"public, max-age=86400" |
| 93 | |
| 94 | @pytest.mark.asyncio |
| 95 | async def test_png_gets_one_day_cache(self) -> None: |
| 96 | msgs = await self._run("/static/favicon-32.png") |
| 97 | start = next(m for m in msgs if m["type"] == "http.response.start") |
| 98 | headers = dict(start["headers"]) |
| 99 | assert headers[b"cache-control"] == b"public, max-age=86400" |
| 100 | |
| 101 | @pytest.mark.asyncio |
| 102 | async def test_non_static_path_passes_through_unmodified(self) -> None: |
| 103 | msgs = await self._run("/repos/abc/tree/main") |
| 104 | start = next(m for m in msgs if m["type"] == "http.response.start") |
| 105 | headers = dict(start["headers"]) |
| 106 | assert b"cache-control" not in headers |
| 107 | |
| 108 | @pytest.mark.asyncio |
| 109 | async def test_existing_cache_control_not_overwritten(self) -> None: |
| 110 | """If app already set Cache-Control, middleware must leave it alone.""" |
| 111 | from musehub.middleware.static_cache import StaticCacheMiddleware |
| 112 | |
| 113 | messages: list[dict] = [] |
| 114 | |
| 115 | async def _app_with_cc(scope: Scope, receive: Receive, send: Send) -> None: |
| 116 | await send({ |
| 117 | "type": "http.response.start", |
| 118 | "status": 200, |
| 119 | "headers": [(b"cache-control", b"no-store")], |
| 120 | }) |
| 121 | await send({"type": "http.response.body", "body": b""}) |
| 122 | |
| 123 | async def _capture(msg: JSONObject) -> None: |
| 124 | messages.append(msg) |
| 125 | |
| 126 | async def _noop_receive() -> None: |
| 127 | return {} |
| 128 | |
| 129 | mw = StaticCacheMiddleware(_app_with_cc) |
| 130 | await mw(self._make_scope("/static/app.css"), _noop_receive, _capture) |
| 131 | |
| 132 | start = next(m for m in messages if m["type"] == "http.response.start") |
| 133 | cc_values = [v for k, v in start["headers"] if k == b"cache-control"] |
| 134 | # exactly one value and it is the app's, not ours |
| 135 | assert cc_values == [b"no-store"] |
| 136 | |
| 137 | @pytest.mark.asyncio |
| 138 | async def test_websocket_scope_passes_through(self) -> None: |
| 139 | """Non-HTTP scope must not be touched.""" |
| 140 | from musehub.middleware.static_cache import StaticCacheMiddleware |
| 141 | |
| 142 | called = [] |
| 143 | |
| 144 | async def _app(scope: Scope, receive: Receive, send: Send) -> None: |
| 145 | called.append(scope["type"]) |
| 146 | |
| 147 | async def _noop_receive() -> None: |
| 148 | return {} |
| 149 | |
| 150 | async def _noop_send(msg: JSONObject) -> None: |
| 151 | pass |
| 152 | |
| 153 | mw = StaticCacheMiddleware(_app) |
| 154 | await mw({"type": "websocket", "path": "/static/app.css"}, _noop_receive, _noop_send) |
| 155 | assert called == ["websocket"] |
| 156 | |
| 157 | |
| 158 | # ── base.html vendor versioning ─────────────────────────────────────────────── |
| 159 | |
| 160 | class TestBaseHtmlVersioning: |
| 161 | _src = _BASE_HTML.read_text() |
| 162 | |
| 163 | def test_alpinejs_has_version_param(self) -> None: |
| 164 | assert re.search(r"alpinejs\.min\.js\?v=\{\{", self._src) |
| 165 | |
| 166 | def test_htmx_has_version_param(self) -> None: |
| 167 | assert re.search(r"htmx\.min\.js\?v=\{\{", self._src) |
| 168 | |
| 169 | def test_json_enc_has_version_param(self) -> None: |
| 170 | assert re.search(r"json-enc\.js\?v=\{\{", self._src) |
| 171 | |
| 172 | def test_response_targets_has_version_param(self) -> None: |
| 173 | assert re.search(r"response-targets\.js\?v=\{\{", self._src) |
| 174 | |
| 175 | def test_app_css_has_version_param(self) -> None: |
| 176 | assert re.search(r"app\.css\?v=\{\{", self._src) |
| 177 | |
| 178 | def test_app_js_has_version_param(self) -> None: |
| 179 | assert re.search(r"app\.js\?v=\{\{", self._src) |
| 180 | |
| 181 | |
| 182 | # ── static_version content hash ─────────────────────────────────────────────── |
| 183 | |
| 184 | class TestStaticVersion: |
| 185 | def test_static_version_is_8_hex_chars_when_assets_exist(self, tmp_path: pathlib.Path) -> None: |
| 186 | """_compute_static_version returns 8 lowercase hex chars when CSS+JS present.""" |
| 187 | css = tmp_path / "app.css" |
| 188 | js = tmp_path / "app.js" |
| 189 | css.write_bytes(b"body{color:red}") |
| 190 | js.write_bytes(b"console.log(1)") |
| 191 | |
| 192 | combined = css.read_bytes() + js.read_bytes() |
| 193 | expected = split_id(blob_id(combined))[1][:8] |
| 194 | |
| 195 | # Not the empty-input sentinel |
| 196 | assert expected != "e3b0c442" |
| 197 | assert re.fullmatch(r"[0-9a-f]{8}", expected) |
| 198 | |
| 199 | def test_static_version_falls_back_to_cache_id_when_assets_missing(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: |
| 200 | """When app.css and app.js are absent, falls back to .cache-id.""" |
| 201 | import importlib |
| 202 | import musehub.api.routes.musehub._templates as tpl_mod |
| 203 | |
| 204 | # Patch the static dir to tmp_path (no CSS/JS) |
| 205 | cache_id = tmp_path / ".cache-id" |
| 206 | cache_id.write_text("cafebabe") |
| 207 | |
| 208 | monkeypatch.setattr(tpl_mod, "_STATIC_DIR", tmp_path) |
| 209 | monkeypatch.setattr(tpl_mod, "_cache_id_path", cache_id) |
| 210 | |
| 211 | result = tpl_mod._compute_static_version() |
| 212 | assert result == "cafebabe" |
| 213 | |
| 214 | def test_static_version_empty_when_nothing_available(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: |
| 215 | import musehub.api.routes.musehub._templates as tpl_mod |
| 216 | |
| 217 | monkeypatch.setattr(tpl_mod, "_STATIC_DIR", tmp_path) |
| 218 | monkeypatch.setattr(tpl_mod, "_cache_id_path", tmp_path / ".cache-id") |
| 219 | |
| 220 | result = tpl_mod._compute_static_version() |
| 221 | assert result == "" |
| 222 | |
| 223 | |
| 224 | # ── objects.py: no blocking sync I/O ───────────────────────────────────────── |
| 225 | |
| 226 | class TestObjectsNoBlockingIO: |
| 227 | _src = _OBJECTS_PY.read_text() |
| 228 | |
| 229 | def test_no_bare_open_in_async_handler(self) -> None: |
| 230 | """ |
| 231 | Parse the AST to ensure no `open(...)` call appears as a direct statement |
| 232 | (not inside a nested sync function) within an async function body. |
| 233 | """ |
| 234 | tree = ast.parse(self._src) |
| 235 | |
| 236 | violations: list[int] = [] |
| 237 | |
| 238 | class _Visitor(ast.NodeVisitor): |
| 239 | def __init__(self) -> None: |
| 240 | self._in_async = 0 |
| 241 | self._in_sync_nested = 0 |
| 242 | |
| 243 | def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None: |
| 244 | self._in_async += 1 |
| 245 | self.generic_visit(node) |
| 246 | self._in_async -= 1 |
| 247 | |
| 248 | def visit_FunctionDef(self, node: ast.FunctionDef) -> None: |
| 249 | if self._in_async > 0: |
| 250 | self._in_sync_nested += 1 |
| 251 | self.generic_visit(node) |
| 252 | self._in_sync_nested -= 1 |
| 253 | else: |
| 254 | self.generic_visit(node) |
| 255 | |
| 256 | def visit_Call(self, node: ast.Call) -> None: |
| 257 | if ( |
| 258 | self._in_async > 0 |
| 259 | and self._in_sync_nested == 0 |
| 260 | and isinstance(node.func, ast.Name) |
| 261 | and node.func.id == "open" |
| 262 | ): |
| 263 | violations.append(node.lineno) |
| 264 | self.generic_visit(node) |
| 265 | |
| 266 | _Visitor().visit(tree) |
| 267 | assert violations == [], f"Bare open() in async handler at lines {violations}" |
| 268 | |
| 269 | |
| 270 | # ── nginx gzip config ───────────────────────────────────────────────────────── |
| 271 | |
| 272 | class TestNginxGzip: |
| 273 | _src = _NGINX_CONF.read_text() |
| 274 | |
| 275 | def test_gzip_on(self) -> None: |
| 276 | assert re.search(r"^\s*gzip\s+on\s*;", self._src, re.MULTILINE) |
| 277 | |
| 278 | def test_gzip_comp_level(self) -> None: |
| 279 | assert re.search(r"gzip_comp_level\s+[1-9]", self._src) |
| 280 | |
| 281 | def test_gzip_types_includes_css(self) -> None: |
| 282 | assert "text/css" in self._src |
| 283 | |
| 284 | def test_gzip_types_includes_js(self) -> None: |
| 285 | assert re.search(r"(text/javascript|application/javascript)", self._src) |
| 286 | |
| 287 | def test_gzip_types_includes_json(self) -> None: |
| 288 | assert "application/json" in self._src |
| 289 | |
| 290 | def test_gzip_vary_on(self) -> None: |
| 291 | assert re.search(r"gzip_vary\s+on\s*;", self._src) |
| 292 | |
| 293 | |
| 294 | # ── StaticCacheMiddleware registered in main.py ─────────────────────────────── |
| 295 | |
| 296 | class TestMiddlewareRegistration: |
| 297 | _src = _MAIN_PY.read_text() |
| 298 | |
| 299 | def test_static_cache_middleware_imported(self) -> None: |
| 300 | assert "StaticCacheMiddleware" in self._src |
| 301 | |
| 302 | def test_static_cache_middleware_added(self) -> None: |
| 303 | assert "add_middleware(StaticCacheMiddleware)" in self._src |