test_server_fastpaths.py (7687B)
1 """Offline checks for the navigation fast paths (no network, no Hugging Face). 2 3 Run from the repo root: HF_BUCKET_ID=ci/placeholder python tests/test_server_fastpaths.py 4 5 Covers: HTML caching + gzip (app/fast_html.py), skipping the wasted HEAD request when the parent 6 folder's listing is cached (storage.get_path_info), and one Hugging Face call per folder even when 7 several requests ask at once (storage.list_directory). 8 """ 9 10 import asyncio 11 import gzip 12 import os 13 import sys 14 15 os.environ.setdefault("HF_BUCKET_ID", "ci/placeholder") 16 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) 17 18 from fastapi.testclient import TestClient # noqa: E402 19 from starlette.requests import Request # noqa: E402 20 from starlette.responses import HTMLResponse # noqa: E402 21 22 import app.main as m # noqa: E402 23 from app import storage # noqa: E402 24 from app.fast_html import HTML_MAX_AGE, optimise_html # noqa: E402 25 26 # A tiny fake bucket. Directories have no `size` attribute, like the real tree items. 27 FILES = { 28 "top.txt": 7, 29 "bsc/README.md": 3000, 30 "bsc/semester_1/notes.txt": 12, 31 "bsc/semester_1/deep/a.bin": 5, 32 "_logs/hidden.log": 1, 33 } 34 calls = {"tree": 0, "head": 0} 35 36 37 class Dir: 38 def __init__(self, path): 39 self.path = path 40 41 42 class File: 43 def __init__(self, path, size): 44 self.path, self.size = path, size 45 46 47 def fake_fetch_tree(path): 48 calls["tree"] += 1 49 path = path.strip("/") 50 prefix = path + "/" if path else "" 51 out = {} 52 for fp, size in FILES.items(): 53 if not fp.startswith(prefix): 54 continue 55 rest = fp[len(prefix):] 56 if "/" in rest: 57 d = prefix + rest.split("/")[0] 58 out[d] = Dir(d) 59 else: 60 out[fp] = File(fp, size) 61 return list(out.values()) 62 63 64 async def fake_get_file_info(path): 65 calls["head"] += 1 66 return {"exists": path in FILES, "size": FILES.get(path), "content_type": None} 67 68 69 async def no_stats(): 70 return None 71 72 73 storage._fetch_tree = fake_fetch_tree 74 storage.get_file_info = fake_get_file_info 75 m.repo_stats = no_stats 76 77 78 def reset(): 79 storage._cache.clear() 80 calls["tree"] = calls["head"] = 0 81 82 83 def check(cond, msg): 84 if not cond: 85 raise AssertionError(msg) 86 print("ok -", msg) 87 88 89 # --- storage ---------------------------------------------------------------------------------------- 90 91 reset() 92 loop = asyncio.new_event_loop() 93 94 async def burst(): 95 return await asyncio.gather(*[storage.list_directory("bsc") for _ in range(6)]) 96 97 results = loop.run_until_complete(burst()) 98 check(calls["tree"] == 1, "six simultaneous listings of one folder make a single Hugging Face call") 99 check(all(r == results[0] for r in results), "...and all six get the same listing") 100 names = [(e["name"], e["is_dir"]) for e in results[0]] 101 check(names == [("semester_1", True), ("README.md", False)], "listing is folders-first, hidden paths excluded") 102 check(results[0][1]["size"] == 3000 and results[0][1]["size_str"] == "2.93 KB", "entries keep the raw size too") 103 104 reset() 105 loop.run_until_complete(storage.list_directory("bsc")) 106 calls["head"] = 0 107 info = loop.run_until_complete(storage.get_path_info("bsc/semester_1")) 108 check(info["exists"] is False and calls["head"] == 0, "folder is recognised from the parent listing, no HEAD") 109 info = loop.run_until_complete(storage.get_path_info("bsc/README.md")) 110 check(info["exists"] is True and info["size"] == 3000 and calls["head"] == 0, "file + size come from the parent listing, no HEAD") 111 info = loop.run_until_complete(storage.get_path_info("bsc/missing")) 112 check(info["exists"] is False and calls["head"] == 1, "a path the parent doesn't list falls back to the HEAD check") 113 reset() 114 info = loop.run_until_complete(storage.get_path_info("bsc/README.md")) 115 check(info["exists"] is True and calls["head"] == 1, "with no cached parent listing it falls back to the HEAD check") 116 loop.close() 117 118 # --- HTML optimisation ------------------------------------------------------------------------------- 119 120 def req(method="GET", accept="gzip"): 121 return Request({"type": "http", "method": method, "headers": [(b"accept-encoding", accept.encode())]}) 122 123 big = "<html>" + "x" * 5000 + "</html>" 124 r = optimise_html(req(), HTMLResponse(big)) 125 check(r.headers["content-encoding"] == "gzip" and gzip.decompress(r.body).decode() == big, "gzip round-trips") 126 check(int(r.headers["content-length"]) == len(r.body) < len(big), "content-length matches the compressed body") 127 check(r.headers["cache-control"] == f"private, max-age={HTML_MAX_AGE}", "pages are cached privately") 128 check("accept-encoding" in r.headers["vary"].lower(), "Vary: Accept-Encoding is set") 129 r = optimise_html(req(accept="identity"), HTMLResponse(big)) 130 check("content-encoding" not in r.headers and r.body.decode() == big, "no gzip unless the client accepts it") 131 r = optimise_html(req(), HTMLResponse("tiny")) 132 check("content-encoding" not in r.headers, "tiny bodies are left uncompressed") 133 r = optimise_html(req(), HTMLResponse(big, status_code=404)) 134 check("cache-control" not in r.headers and "content-encoding" not in r.headers, "non-200 responses are untouched") 135 r = optimise_html(req(method="POST"), HTMLResponse(big)) 136 check("cache-control" not in r.headers, "non-GET requests are untouched") 137 r = HTMLResponse(big, headers={"Cache-Control": "no-store"}) 138 check(optimise_html(req(), r).headers["cache-control"] == "no-store", "an explicit Cache-Control is respected") 139 140 # --- end to end through the real app ----------------------------------------------------------------- 141 142 reset() 143 c = TestClient(m.app) 144 gz = {"Accept-Encoding": "gzip"} 145 146 r = c.get("/bsc", headers=gz) 147 check(r.status_code == 200 and r.headers.get("content-encoding") == "gzip", "/bsc listing is served gzipped") 148 check("semester_1" in r.text and "README.md" in r.text, "...and decodes to the real page") 149 check(r.headers["cache-control"] == f"private, max-age={HTML_MAX_AGE}", "...with the browser-cache header") 150 151 calls["head"] = 0 152 r = c.get("/bsc/semester_1", headers=gz) 153 check(r.status_code == 200 and "notes.txt" in r.text and calls["head"] == 0, "opening a folder from its parent makes no HEAD request") 154 r = c.get("/bsc/README.md", headers=gz) 155 check(r.status_code == 200 and "2.93 KB" in r.text and calls["head"] == 0, "opening a file from its parent makes no HEAD request") 156 157 r = c.get("/bsc", headers={"Accept-Encoding": "identity"}) 158 check(r.status_code == 200 and "content-encoding" not in r.headers, "identity clients get plain HTML") 159 160 for path in ["/new", "/find", "/history", "/shorten", "/documentation", "/gh-sync"]: 161 r = c.get(path, headers=gz) 162 check(r.status_code == 200 and r.headers.get("content-encoding") == "gzip" 163 and "max-age" in r.headers["cache-control"], f"{path} is cached and gzipped") 164 165 r = c.get("/no/such/folder", headers=gz) 166 check(r.status_code == 404 and "max-age" not in r.headers.get("cache-control", ""), "404 pages are not cached") 167 r = c.get("/static/sw.js", headers=gz) 168 check(r.headers["cache-control"].startswith("no-store") and "content-encoding" not in r.headers, "service worker is still no-store and untouched") 169 for name in ("app.css", "app.js"): 170 r = c.get(f"/static/{name}", headers=gz) 171 plain = (m.STATIC_DIR / name).read_bytes() 172 check(r.headers.get("content-encoding") == "gzip" and r.content == plain and len(r.headers.get("content-length", "0")) > 0 173 and r.headers["vary"] == "Accept-Encoding", f"/static/{name} is gzipped and decodes to the original") 174 r = c.get(f"/static/{name}", headers={"accept-encoding": "identity"}) 175 check("content-encoding" not in r.headers and r.content == plain, f"/static/{name} stays plain for clients without gzip") 176 177 print("\nall server fast-path checks passed")