mirror of
https://github.com/baketnk/frame-yap.git
synced 2026-10-06 00:00:05 +02:00
170 lines
9.3 KiB
Python
170 lines
9.3 KiB
Python
"""Offline hash/status and CLI tests; fixtures are bytes, not models."""
|
|
import contextlib
|
|
from dataclasses import replace
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import io
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "python"))
|
|
from frameyap.model_files import ModelFile, check_file, check_model, load_backends
|
|
|
|
CLI = ROOT / "scripts" / "model-status.py"
|
|
|
|
|
|
class ModelFileTests(unittest.TestCase):
|
|
def fixture(self, root):
|
|
manifests = root / "manifests"
|
|
manifests.mkdir()
|
|
backend = json.loads((ROOT / "assets/backends/redux.json").read_text())
|
|
backend["model"]["files"] = [{"path": "nested/weights.bin", "size": 7,
|
|
"sha256": hashlib.sha256(b"fixture").hexdigest()}]
|
|
(manifests / "redux.json").write_text(json.dumps(backend))
|
|
return manifests, load_backends(manifests)["redux"]
|
|
|
|
def cli(self, manifests, *arguments):
|
|
return subprocess.run([sys.executable, str(CLI), "--manifest-dir", str(manifests),
|
|
"--json", *map(str, arguments)], capture_output=True, text=True, timeout=5)
|
|
|
|
def test_status_and_cli_from_same_verifier(self):
|
|
with tempfile.TemporaryDirectory() as path:
|
|
root = Path(path)
|
|
manifests, backend = self.fixture(root)
|
|
store = root / "models"
|
|
model = store / "redux"
|
|
result = self.cli(manifests, "--list-models", "--model-dir", store)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
listed = json.loads(result.stdout)
|
|
self.assertEqual(list(listed), ["models", "schema"])
|
|
self.assertEqual(listed["models"][0]["state"], "not_installed")
|
|
self.assertEqual(listed["models"][0]["reason"], "directory_missing")
|
|
self.assertEqual(listed["models"][0]["total_bytes"], 7)
|
|
self.assertEqual(self.cli(manifests, "--check-model", "redux", "--model-dir", model).returncode, 1)
|
|
(model / "nested").mkdir(parents=True)
|
|
weight = model / "nested/weights.bin"
|
|
weight.write_bytes(b"fixture")
|
|
checked = self.cli(manifests, "--check-model", "redux", "--model-dir", model)
|
|
self.assertEqual(checked.returncode, 0, checked.stderr)
|
|
self.assertEqual(json.loads(checked.stdout), {"schema": 1, "model": check_model(backend, model)})
|
|
self.assertEqual(check_file(model, backend.files[0]), (None, None))
|
|
weight.write_bytes(b"altered")
|
|
self.assertEqual(check_model(backend, model)["reason"], "hash_mismatch")
|
|
self.assertEqual(json.loads(self.cli(manifests, "--check-model", "redux", "--model-dir", model).stdout)["model"]["state"], "invalid")
|
|
weight.write_bytes(b"short")
|
|
self.assertEqual(check_model(backend, model)["reason"], "size_mismatch")
|
|
weight.unlink()
|
|
self.assertEqual(check_model(backend, model)["state"], "not_installed")
|
|
outside = root / "outside"
|
|
outside.write_bytes(b"fixture")
|
|
weight.symlink_to(outside)
|
|
self.assertEqual(check_model(backend, model)["reason"], "unsafe_file")
|
|
weight.unlink()
|
|
weight.symlink_to(root / "absent")
|
|
self.assertEqual(check_model(backend, model)["reason"], "unsafe_file")
|
|
weight.unlink()
|
|
(model / "nested").rmdir()
|
|
(model / "nested").symlink_to(root)
|
|
self.assertEqual(check_model(backend, model)["reason"], "unsafe_file")
|
|
|
|
def test_compatible_card_still_requires_exact_hash_and_safe_file(self):
|
|
with tempfile.TemporaryDirectory() as path:
|
|
root = Path(path)
|
|
_, backend = self.fixture(root)
|
|
current, previous = b"current card", b"old card"
|
|
item = ModelFile("README.md", len(current), hashlib.sha256(current).hexdigest(),
|
|
((len(previous), hashlib.sha256(previous).hexdigest()),))
|
|
backend = replace(backend, files=(item,))
|
|
card = root / "README.md"
|
|
for content in (current, previous):
|
|
card.write_bytes(content)
|
|
self.assertEqual(check_model(backend, root)["state"], "installed_verified")
|
|
# A download temp pin must never accept an older artifact.
|
|
self.assertEqual(check_file(root, ModelFile(item.path, item.size, item.sha256))[0],
|
|
"size_mismatch")
|
|
card.write_bytes(b"bad card")
|
|
self.assertEqual(check_model(backend, root)["reason"], "hash_mismatch")
|
|
card.unlink()
|
|
outside = root / "outside"
|
|
outside.write_bytes(previous)
|
|
card.symlink_to(outside)
|
|
self.assertEqual(check_model(backend, root)["reason"], "unsafe_file")
|
|
|
|
def test_second_backend_status_without_runtime_changes(self):
|
|
with tempfile.TemporaryDirectory() as path:
|
|
root = Path(path)
|
|
manifests, _ = self.fixture(root)
|
|
independent = json.loads((manifests / "redux.json").read_text())
|
|
independent["id"] = "independent"
|
|
independent["display_name"] = "Independent backend"
|
|
independent["launcher"] = {"type": "executable", "path": "bin/independent-worker",
|
|
"arguments": ["--model", "{model_dir}", "--clip-dir", "{clip_dir}"],
|
|
"protocol": "frameyap-worker-v1"}
|
|
independent["model"]["files"] = [{"path": "alternate.bin", "size": 3,
|
|
"sha256": hashlib.sha256(b"abc").hexdigest()}]
|
|
(manifests / "independent.json").write_text(json.dumps(independent))
|
|
model_dir = root / "store/independent"
|
|
model_dir.mkdir(parents=True)
|
|
(model_dir / "alternate.bin").write_bytes(b"abc")
|
|
listed = self.cli(manifests, "--list-models", "--model-dir", root / "store")
|
|
self.assertEqual(listed.returncode, 0, listed.stderr)
|
|
models = json.loads(listed.stdout)["models"]
|
|
self.assertEqual([(model["id"], model["state"]) for model in models],
|
|
[("independent", "installed_verified"), ("redux", "not_installed")])
|
|
self.assertEqual(self.cli(manifests, "--check-model", "independent", "--model-dir", model_dir).returncode, 0)
|
|
|
|
def test_explicit_fetch_tool_reuses_shared_verifier_offline(self):
|
|
spec = importlib.util.spec_from_file_location("fetch_model", ROOT / "scripts/fetch-model.py")
|
|
fetch = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(fetch)
|
|
with tempfile.TemporaryDirectory() as path:
|
|
root = Path(path)
|
|
_, backend = self.fixture(root)
|
|
backend = replace(backend, files=(ModelFile("weights.bin", 7, hashlib.sha256(b"fixture").hexdigest()),))
|
|
model = root / "local-model"
|
|
# The Redux-only tool uses this small fixture manifest here; all network
|
|
# calls are mocked, never real downloads.
|
|
with patch.object(fetch, "load_backends", return_value={"redux": backend}), \
|
|
patch.object(fetch.urllib.request, "urlopen", return_value=io.BytesIO(b"fixture")) as urlopen, \
|
|
patch.object(sys, "argv", ["fetch-model.py", "--destination", str(model)]):
|
|
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
|
|
fetch.main()
|
|
urlopen.assert_called_once()
|
|
self.assertEqual(check_model(backend, model)["state"], "installed_verified")
|
|
fetch.main()
|
|
urlopen.assert_called_once()
|
|
(model / "weights.bin").write_bytes(b"changed")
|
|
with self.assertRaises(SystemExit):
|
|
fetch.main()
|
|
urlopen.assert_called_once()
|
|
|
|
def test_missing_manifest_unknown_id_and_no_runtime_imports(self):
|
|
with tempfile.TemporaryDirectory() as path:
|
|
manifests, _ = self.fixture(Path(path))
|
|
result = self.cli(manifests, "--check-model", "unknown", "--model-dir", path)
|
|
self.assertEqual(result.returncode, 2)
|
|
self.assertEqual(result.stdout, "")
|
|
result = self.cli(Path(path) / "absent", "--list-models")
|
|
self.assertEqual(result.returncode, 2)
|
|
result = self.cli(manifests, "--list-models")
|
|
self.assertEqual(result.returncode, 0)
|
|
self.assertEqual(json.loads(result.stdout)["models"][0]["state"], "unknown")
|
|
self.assertEqual(json.loads(result.stdout)["models"][0]["reason"], "model_dir_unspecified")
|
|
# This subprocess never imports inference packages even if not installed.
|
|
result = subprocess.run([sys.executable, "-c", "import sys; import frameyap.model_files; "
|
|
"assert not {'moondream', 'torch', 'numpy'} & set(sys.modules)"],
|
|
env={**os.environ, "PYTHONPATH": str(ROOT / "python")},
|
|
capture_output=True, timeout=5)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|