"""The image recipe and the locked dependency set, offline -- the code audit of 2026-09-21: optimization 9 (the model prefetch moved before the data COPYs) and missing item 10 (the image installs requirements.lock, the full hashed set, instead of re-resolving the transitive dependencies at every build; tools/test_in_docker.sh runs the suite on the shipped 3.12). Docker is not run here: the Dockerfile is read as text, its layer order and install line checked, and the lock checked against requirements.txt. python -m unittest tests.test_deploy_image """ import re import unittest from pathlib import Path ROOT = Path(__file__).resolve().parent.parent PIN = re.compile(r"^([A-Za-z0-9][A-Za-z0-9._-]*)==([^\s;\\#]+)") def instructions(): """[(KEYWORD, arguments)] of the Dockerfile, continuation lines joined.""" out, pending = [], "" for line in (ROOT / "Dockerfile").read_text(encoding="utf-8").splitlines(): if not pending and (not line.strip() or line.lstrip().startswith("#")): continue pending += line.rstrip("\\").strip() + " " if not line.rstrip().endswith("\\"): word, _, rest = pending.strip().partition(" ") out.append((word.upper(), rest)) pending = "" return out def pins(path): out = {} for line in (ROOT / path).read_text(encoding="utf-8").splitlines(): m = PIN.match(line.strip()) if not line[:1].isspace() else None if m: out[re.sub(r"[-_.]+", "-", m.group(1)).lower()] = m.group(2) return out class LayerOrderTests(unittest.TestCase): """Optimization 9: the 331 MB model prefetch is rebuilt only when canlex/ or the lock changes, not on every corpus, curated or embeddings change.""" @classmethod def setUpClass(cls): cls.steps = instructions() def index(self, keyword, needle): hits = [i for i, (k, rest) in enumerate(self.steps) if k == keyword and needle in rest] self.assertEqual(len(hits), 1, f"{keyword} {needle}: {hits}") return hits[0] def test_the_prefetch_follows_the_code_and_precedes_every_data_copy(self): code = self.index("COPY", "canlex/ ./canlex/") prefetch = self.index("RUN", "Embedder()") self.assertIn("Reranker()", self.steps[prefetch][1]) data = [i for i, (k, rest) in enumerate(self.steps) if k == "COPY" and "data/" in rest] self.assertEqual(len(data), 3) # before the audit: code 6, data 7-9, prefetch 12 -- after every data COPY self.assertLess(code, prefetch) self.assertLess(prefetch, min(data)) between = [k for k, _ in self.steps[code + 1:prefetch]] self.assertEqual(between, ["USER", "ENV"]) # nothing else rides along def test_the_prefetch_runs_as_app_with_the_cache_in_its_home(self): prefetch = self.index("RUN", "Embedder()") self.assertLess(self.index("USER", "app"), prefetch) env = self.index("ENV", "HF_HOME=/app/.hf_cache") self.assertLess(env, prefetch) self.assertIn("HOME=/app", self.steps[env][1]) def test_offline_mode_stays_after_the_data_and_the_image_is_otherwise_unchanged(self): offline = self.index("ENV", "HF_HUB_OFFLINE=1") data = max(i for i, (k, rest) in enumerate(self.steps) if k == "COPY" and "data/" in rest) self.assertGreater(offline, data) self.assertEqual([k for k, _ in self.steps[offline:]], ["ENV", "EXPOSE", "CMD"]) self.assertEqual(self.steps[-1], ("CMD", '["python", "-m", "canlex.server"]')) env = self.steps[self.index("ENV", "HF_HOME=/app/.hf_cache")][1] for var in ("CANLEX_HTTP=1", "CANLEX_LOG_DIR=/tmp/canlex-logs", "PORT=7860", "PYTHONUNBUFFERED=1"): self.assertIn(var, env) for i, (k, rest) in enumerate(self.steps): if k == "COPY" and "requirements" not in rest: self.assertTrue(rest.startswith("--chown=app:app "), rest) class LockTests(unittest.TestCase): """Missing item 10: the image installs the whole locked set, hashed.""" def test_the_image_installs_the_lock_with_hashes(self): steps = instructions() self.assertIn(("COPY", "requirements.lock ."), steps) self.assertIn(("RUN", "pip install --no-cache-dir --require-hashes -r requirements.lock"), steps) self.assertFalse(any("requirements.txt" in rest for _, rest in steps)) def test_the_lock_pins_every_direct_dependency_at_its_version(self): direct, locked = pins("requirements.txt"), pins("requirements.lock") self.assertGreaterEqual(len(direct), 8) for name, version in direct.items(): self.assertEqual(locked.get(name), version, name) self.assertGreater(len(locked), len(direct)) # the transitives too for name in ("starlette", "anyio", "uvicorn", "pydantic", "httpx"): self.assertIn(name, locked) def test_every_locked_package_carries_a_hash_and_the_lock_names_its_recipe(self): text = (ROOT / "requirements.lock").read_text(encoding="utf-8") blocks = re.split(r"\n(?=[A-Za-z0-9])", text.split("\n", 2)[2]) self.assertEqual(len(blocks), len(pins("requirements.lock"))) for block in blocks: self.assertIn("--hash=sha256:", block, block.splitlines()[0]) self.assertIn("uv pip compile requirements.txt --python-platform linux " "--python-version 3.12 --generate-hashes -o requirements.lock", text.splitlines()[1]) def test_mcp_is_past_the_websocket_advisory(self): # GHSA-vj7q-gjh5-988w, fixed in 1.28.1; 1.30.0 checked against the hooks # canlex/calllog.py and canlex/server.py rely on (requirements.txt) version = tuple(int(x) for x in pins("requirements.txt")["mcp"].split(".")) self.assertGreaterEqual(version, (1, 28, 1)) class DockerRecipeTests(unittest.TestCase): def test_the_suite_recipe_uses_the_shipped_interpreter_and_the_lock(self): text = (ROOT / "tools" / "test_in_docker.sh").read_text(encoding="utf-8") self.assertTrue(text.startswith("#!/usr/bin/env sh")) self.assertIn("python:3.12-slim", text) self.assertIn("pip install -q --require-hashes -r requirements.lock", text) self.assertIn("HF_HUB_OFFLINE=1", text) self.assertIn(":/hf:ro", text) # the model cache, read-only self.assertIn(":/src:ro", text) # the checkout, read-only self.assertIn("libgomp1", text) if __name__ == "__main__": unittest.main()