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