tests/test_deploy_image.py

"""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()