File size: 6,631 Bytes
901bf8c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | """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()
|