File size: 5,166 Bytes
c37a0e2 | 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 141 142 143 144 145 146 147 148 | """Release-location tests; model construction is replaced to avoid GPU work."""
import json
import re
from pathlib import Path
import pytest
from huggingface_hub import constants
from huggingface_hub.errors import LocalEntryNotFoundError
import inference
FILES = (
"config.json",
"decision_head.safetensors",
"adapter/adapter_config.json",
"adapter/adapter_model.safetensors",
)
class ProbeMiniJev(inference.MiniJev):
def __init__(self, release_dir, device="cuda"):
self.release_dir = release_dir
self.device = device
self.config = json.loads((Path(release_dir) / "config.json").read_text())
@pytest.fixture
def release(tmp_path):
for name in FILES:
path = tmp_path / name
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text('{"marker": "resolved"}' if name.endswith('.json') else 'weights')
return tmp_path
@pytest.fixture
def hub(monkeypatch, release):
calls = []
def download(**kwargs):
calls.append(kwargs)
return str(release)
monkeypatch.setattr(inference, "snapshot_download", download, raising=False)
return calls
def test_local_string_takes_precedence_over_repo_id(monkeypatch, tmp_path, release, hub):
local = tmp_path / "example" / "model"
local.mkdir(parents=True)
(local / "config.json").write_text('{"marker": "local"}')
monkeypatch.chdir(tmp_path)
model = ProbeMiniJev.load("example/model", "cuda:1")
assert model.config["marker"] == "local"
assert model.release_dir == "example/model"
assert model.device == "cuda:1"
assert hub == []
def test_path_remains_local(release, hub):
model = ProbeMiniJev.load(release)
assert model.release_dir is release
assert model.config["marker"] == "resolved"
assert hub == []
@pytest.mark.parametrize("argument", [None, "samatv256/mini-Jev", "example/model"])
def test_hub_default_and_explicit_repo_use_exact_allowlist(argument, release, hub):
model = ProbeMiniJev.load() if argument is None else ProbeMiniJev.load(argument)
assert model.config["marker"] == "resolved"
assert Path(model.release_dir) == release
assert len(hub) == 1
assert hub[0] == {
"repo_id": argument or "samatv256/mini-Jev",
"revision": None,
"allow_patterns": list(FILES),
}
@pytest.mark.parametrize("revision", ["step-010626", "a" * 40])
def test_revision_is_resolved_in_one_snapshot(revision, release, hub):
model = ProbeMiniJev.load("example/model", revision=revision)
assert model.config["marker"] == "resolved"
assert len(hub) == 1
assert hub[0]["revision"] == revision
@pytest.mark.parametrize("missing", FILES)
def test_incomplete_hub_snapshot_fails_before_constructor(missing, release, hub):
(release / missing).unlink()
with pytest.raises(FileNotFoundError, match="missing required release files:.*" + re.escape(missing)):
ProbeMiniJev.load("example/model")
@pytest.mark.parametrize("argument", ["./missing/model", "../missing/model", "/missing/model", "missing", Path("missing/model")])
def test_explicit_missing_local_paths_do_not_download(argument, hub):
with pytest.raises(FileNotFoundError):
ProbeMiniJev.load(argument)
assert hub == []
def test_local_revision_does_not_change_loading(release, hub):
model = ProbeMiniJev.load(str(release), revision="step-010626")
assert model.config["marker"] == "resolved"
assert hub == []
def test_offline_cache_miss_is_propagated(monkeypatch):
def download(**kwargs):
raise LocalEntryNotFoundError("no cached release")
monkeypatch.setattr(inference, "snapshot_download", download, raising=False)
with pytest.raises(LocalEntryNotFoundError, match="no cached release"):
ProbeMiniJev.load("example/model")
@pytest.mark.parametrize("revision", [None, "a" * 40])
def test_real_hub_offline_cache_hit(monkeypatch, tmp_path, release, revision):
cache = tmp_path / "cache"
storage = cache / "models--example--model"
snapshot = storage / "snapshots" / ("a" * 40)
entries = {}
for name in FILES:
data = (release / name).read_bytes()
path = snapshot / name
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(data)
entries[name] = {"size": len(data), "blob_id": "b" * 40}
(storage / "refs").mkdir()
(storage / "refs" / "main").write_text("a" * 40)
(storage / "trees").mkdir()
(storage / "trees" / (("a" * 40) + ".json")).write_text(
json.dumps({"format_version": 1, "files": entries})
)
monkeypatch.setattr(constants, "HF_HUB_CACHE", str(cache))
monkeypatch.setattr(constants, "HF_HUB_OFFLINE", True)
model = ProbeMiniJev.load("example/model", revision=revision)
assert Path(model.release_dir) == snapshot
assert model.config["marker"] == "resolved"
def test_real_hub_offline_cache_miss(monkeypatch, tmp_path):
monkeypatch.setattr(constants, "HF_HUB_CACHE", str(tmp_path / "empty-cache"))
monkeypatch.setattr(constants, "HF_HUB_OFFLINE", True)
with pytest.raises(LocalEntryNotFoundError):
ProbeMiniJev.load("example/uncached")
|