cesnimda
57 lines
2.7 KiB
Python
57 lines
2.7 KiB
Python
"""Verify the dedicated development model cache before local inference."""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
MODEL = "qwen2.5:7b"
|
|
MANIFEST = "845dbda0ea48ed749caafd9e6037047aa19acfcfd82e704d7ca97d631a0b697e"
|
|
IMAGE = "ollama/ollama:0.40.11@sha256:459985308d740b99071719baec2740a60f5b2b2d9b835c171991ae25690203da"
|
|
CONTAINER = "jobjakt-dev-ollama-ollama-1"
|
|
|
|
|
|
def run(*arguments):
|
|
return subprocess.check_output(["rtk", "proxy", *arguments], text=True).strip()
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--manifest", required=True, type=Path)
|
|
parser.add_argument("--output", required=True, type=Path)
|
|
args = parser.parse_args()
|
|
source = args.manifest.read_bytes()
|
|
assert hashlib.sha256(source).hexdigest() == MANIFEST, "Registry manifest changed"
|
|
manifest = json.loads(source)
|
|
container = json.loads(run("docker", "inspect", CONTAINER))[0]
|
|
assert container["Config"]["Labels"]["com.docker.compose.project"] == "jobjakt-dev-ollama"
|
|
assert container["Image"] == json.loads(run("docker", "image", "inspect", IMAGE))[0]["Id"]
|
|
assert "OLLAMA_NO_CLOUD=1" in container["Config"]["Env"]
|
|
assert container["HostConfig"]["ReadonlyRootfs"]
|
|
assert container["HostConfig"]["Memory"] == 8 * 1024**3
|
|
assert all(mount.get("Name") == "jobjakt-dev-ollama-models" for mount in container["Mounts"] if mount["Type"] == "volume")
|
|
descriptors = [manifest["config"], *manifest["layers"], {"digest": "sha256:" + MANIFEST}]
|
|
checked = []
|
|
for descriptor in descriptors:
|
|
digest = descriptor["digest"]
|
|
assert re.fullmatch(r"sha256:[a-f0-9]{64}", digest)
|
|
path = "/root/.ollama/models/blobs/" + digest.replace(":", "-")
|
|
actual = run("docker", "exec", CONTAINER, "sha256sum", path).split()[0]
|
|
assert actual == digest[7:], "Downloaded model blob changed"
|
|
checked.append(digest)
|
|
with urllib.request.urlopen("http://127.0.0.1:11434/api/tags", timeout=5) as response:
|
|
models = json.load(response)["models"]
|
|
actual = next(model for model in models if model["name"] == MODEL)
|
|
assert actual["digest"] == MANIFEST
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
record = {"status": "PASS", "model": MODEL, "manifestSha256": MANIFEST, "checkedBlobs": checked,
|
|
"image": IMAGE, "details": actual["details"], "limits": "Local artifact verification only; no inference, production digest parity or extraction accuracy claim."}
|
|
args.output.write_text(json.dumps(record, indent=2) + "\n", encoding="utf-8")
|
|
print(json.dumps(record))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|