grenadeandClaude Opus 5
cb1ca8b6af
Scaffolds the workspace per architecture/generic.md §1 and implements
phases 0-3 of doc/spec.md §12.
Crates:
entities manifest/ref/repo schemas, bucket key layout, config, the
X-Error-Code taxonomy. No I/O.
core resolver, freeze pinning, single-flight, the streaming tee,
policy, gc/verify/refresh. Defines the Store and Upstream
ports.
data object_store (S3 + local) and reqwest Hub adapters.
api the axum surface: resolve, model/dataset info, tree, refs,
whoami, metrics, bearer auth, range handling.
bin one binary: serve plus fetch/pin/refresh/list/show/rm/gc/
verify/doctor.
Deployment targets bob.hanzalova.internal:20482 (port derived per
architecture/port-allocations.md §3), storing to the MinIO on
caveman.kosherinata.internal, fronted by hanzalova at rf.internal.
Ships the sysusers drop-in, hardened unit, firewalld service, nginx
vhost, config template, infra-setup.sh and the Gitea Actions
ci/deploy/conformance workflows.
Testing: 112 unit and integration tests, including the sovereignty
suite (cold fetch, sealed replay, single-flight, client disconnect,
range resume, freeze stability, gc-after-rm, digest mismatch), plus a
conformance suite driving a pinned huggingface_hub against a real Hub.
Deviations from the spec, all deliberate:
- one binary with subcommands (spec §9) rather than generic.md's
separate -api and -cli binaries; the library split is unchanged.
- a dedicated sysusers account and hardened unit (generic.md §8)
rather than the spec's illustrative DynamicUser unit.
- manifests carry an optional repo_tree recorded verbatim, resolving
spec §13's "record whole, filter on read" question for the tree
endpoint as well as model-info.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XZG2i4AmfSqE97EJGBVb64
556 lines
22 KiB
Python
Executable File
556 lines
22 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Conformance suite: test against real clients, not against the document.
|
|
|
|
The spec is only as good as the client compatibility, so this drives an
|
|
unmodified ``huggingface_hub`` (and the ``hf`` CLI) through rustingface and
|
|
checks the behaviour a stack pinned to a model revision actually depends on.
|
|
|
|
The test that matters is the offline one. Its steps here mirror spec §11:
|
|
|
|
1. fresh bucket, ``HF_ENDPOINT`` set, pull a model, confirm bytes landed;
|
|
2. cut rustingface off from upstream;
|
|
3. wipe every client-side cache;
|
|
4. pull again -- it must succeed identically, with no upstream traffic;
|
|
5. kill rustingface, start a *new* process against the same bucket, repeat.
|
|
|
|
Step 2 differs from the spec in one honest respect. The spec null-routes
|
|
``huggingface.co`` at the site router and verifies there, which CI cannot do.
|
|
This suite instead does something strictly stronger for the sealed run --
|
|
``upstream.enabled = false`` means no upstream client is constructed at all, so
|
|
there is no code path that *could* reach the network -- and, for the
|
|
"upstream enabled but unreachable" run, points the endpoint at a closed port
|
|
and asserts that a miss fails loudly rather than hanging. Run the real
|
|
null-route test by hand before trusting a deployment.
|
|
|
|
Usage::
|
|
|
|
python3 test/conformance/conformance.py --binary target/release/rustingface
|
|
|
|
Set ``--s3-endpoint``/``--s3-bucket`` (plus ``AWS_ACCESS_KEY_ID`` and
|
|
``AWS_SECRET_ACCESS_KEY``) to run against a real bucket instead of a directory.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import shutil
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
# A small, ungated, stable model. Small enough that CI can pull it repeatedly,
|
|
# real enough that it exercises the LFS bridge and a safetensors file.
|
|
REPO = os.environ.get("CONFORMANCE_REPO", "hf-internal-testing/tiny-random-gpt2")
|
|
REVISION = os.environ.get("CONFORMANCE_REVISION", "main")
|
|
|
|
# The X-Error-Code values rustingface emits. These are client-internal
|
|
# constants upstream and are re-checked against the installed client below.
|
|
EXPECTED_ERROR_CODES = {
|
|
"RepoNotFound",
|
|
"RevisionNotFound",
|
|
"EntryNotFound",
|
|
"GatedRepo",
|
|
}
|
|
|
|
# A disabled repo is the odd one out: the client dispatches DisabledRepoError
|
|
# on this exact X-Error-Message, not on an X-Error-Code, so rustingface emits
|
|
# the message verbatim. If this string moves, the mapping in
|
|
# rustingface-api/src/error.rs moves with it.
|
|
DISABLED_REPO_MESSAGE = "Access to this resource is disabled."
|
|
|
|
|
|
class Failure(Exception):
|
|
"""A conformance check that did not hold."""
|
|
|
|
|
|
@dataclass
|
|
class Results:
|
|
passed: list[str] = field(default_factory=list)
|
|
failed: list[tuple[str, str]] = field(default_factory=list)
|
|
|
|
def check(self, name: str, fn) -> None:
|
|
print(f"\n=== {name}")
|
|
try:
|
|
fn()
|
|
except Exception as err: # noqa: BLE001 - a suite reports, it does not raise
|
|
print(f" FAILED: {err}")
|
|
self.failed.append((name, str(err)))
|
|
else:
|
|
print(" ok")
|
|
self.passed.append(name)
|
|
|
|
def report(self) -> int:
|
|
print(f"\n{len(self.passed)} passed, {len(self.failed)} failed")
|
|
for name, err in self.failed:
|
|
print(f" FAILED {name}: {err}")
|
|
return 1 if self.failed else 0
|
|
|
|
|
|
def free_port() -> int:
|
|
with socket.socket() as sock:
|
|
sock.bind(("127.0.0.1", 0))
|
|
return sock.getsockname()[1]
|
|
|
|
|
|
def closed_port() -> int:
|
|
"""A port nothing is listening on, standing in for an unreachable Hub."""
|
|
return free_port()
|
|
|
|
|
|
class Instance:
|
|
"""A running rustingface, configured for one phase of the suite."""
|
|
|
|
def __init__(self, binary: Path, workdir: Path, storage: dict, upstream: dict):
|
|
self.binary = binary
|
|
self.workdir = workdir
|
|
self.port = free_port()
|
|
self.config_path = workdir / f"config-{self.port}.toml"
|
|
self.process: subprocess.Popen | None = None
|
|
|
|
storage_lines = "\n".join(f"{k} = {v}" for k, v in storage.items())
|
|
upstream_lines = "\n".join(f"{k} = {v}" for k, v in upstream.items())
|
|
self.config_path.write_text(
|
|
f"""
|
|
[server]
|
|
listen = "127.0.0.1:{self.port}"
|
|
request_timeout = "600s"
|
|
|
|
[storage]
|
|
{storage_lines}
|
|
|
|
[upstream]
|
|
{upstream_lines}
|
|
|
|
[policy]
|
|
ref_resolution = "freeze"
|
|
|
|
[observability]
|
|
metrics = true
|
|
log_format = "text"
|
|
"""
|
|
)
|
|
|
|
@property
|
|
def endpoint(self) -> str:
|
|
return f"http://127.0.0.1:{self.port}"
|
|
|
|
def __enter__(self) -> "Instance":
|
|
self.process = subprocess.Popen(
|
|
[str(self.binary), "--config", str(self.config_path), "serve"],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
text=True,
|
|
)
|
|
for _ in range(100):
|
|
if self.process.poll() is not None:
|
|
raise Failure(f"rustingface exited early:\n{self.process.stdout.read()}")
|
|
try:
|
|
with urllib.request.urlopen(f"{self.endpoint}/healthz", timeout=1):
|
|
return self
|
|
except (urllib.error.URLError, ConnectionError, TimeoutError):
|
|
time.sleep(0.1)
|
|
raise Failure("rustingface did not become healthy")
|
|
|
|
def __exit__(self, *_exc) -> None:
|
|
if self.process is not None:
|
|
self.process.terminate()
|
|
try:
|
|
self.process.wait(timeout=30)
|
|
except subprocess.TimeoutExpired:
|
|
self.process.kill()
|
|
|
|
def metrics(self) -> str:
|
|
with urllib.request.urlopen(f"{self.endpoint}/metrics", timeout=5) as response:
|
|
return response.read().decode()
|
|
|
|
def cli(self, *args: str) -> str:
|
|
result = subprocess.run(
|
|
[str(self.binary), "--config", str(self.config_path), *args],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
if result.returncode != 0:
|
|
raise Failure(f"`rustingface {' '.join(args)}` failed:\n{result.stdout}\n{result.stderr}")
|
|
return result.stdout
|
|
|
|
|
|
def wipe_client_caches(home: Path) -> None:
|
|
"""Remove every client-side cache, so a pull cannot be served locally."""
|
|
for sub in ("hub", "assets", "xet"):
|
|
shutil.rmtree(home / sub, ignore_errors=True)
|
|
shutil.rmtree(home.parent / ".cache" / "huggingface", ignore_errors=True)
|
|
|
|
|
|
def client_env(endpoint: str, home: Path) -> dict:
|
|
env = dict(os.environ)
|
|
env.update(
|
|
{
|
|
"HF_ENDPOINT": endpoint,
|
|
"HF_HOME": str(home),
|
|
"HF_HUB_CACHE": str(home / "hub"),
|
|
# rustingface implements no Xet client and never advertises the
|
|
# capability; setting this makes the client's side explicit too.
|
|
"HF_HUB_DISABLE_XET": "1",
|
|
"HF_HUB_DISABLE_TELEMETRY": "1",
|
|
"HF_HUB_DISABLE_PROGRESS_BARS": "1",
|
|
}
|
|
)
|
|
env.pop("HF_TOKEN", None)
|
|
return env
|
|
|
|
|
|
def run_client(script: str, endpoint: str, home: Path) -> str:
|
|
"""Run a snippet in a fresh interpreter, so no client state leaks between phases."""
|
|
result = subprocess.run(
|
|
[sys.executable, "-c", script],
|
|
capture_output=True,
|
|
text=True,
|
|
env=client_env(endpoint, home),
|
|
)
|
|
if result.returncode != 0:
|
|
raise Failure(f"client failed:\n{result.stdout}\n{result.stderr}")
|
|
return result.stdout
|
|
|
|
|
|
DOWNLOAD_ONE = """
|
|
import hashlib, json, sys
|
|
from huggingface_hub import hf_hub_download
|
|
path = hf_hub_download({repo!r}, "config.json", revision={revision!r})
|
|
data = open(path, "rb").read()
|
|
print(json.dumps({{"sha256": hashlib.sha256(data).hexdigest(), "len": len(data)}}))
|
|
"""
|
|
|
|
SNAPSHOT = """
|
|
import hashlib, json, os
|
|
from huggingface_hub import snapshot_download
|
|
root = snapshot_download({repo!r}, revision={revision!r})
|
|
files = {{}}
|
|
for base, _dirs, names in os.walk(root):
|
|
for name in names:
|
|
full = os.path.join(base, name)
|
|
if os.path.islink(full) and not os.path.exists(full):
|
|
continue
|
|
rel = os.path.relpath(full, root)
|
|
files[rel] = hashlib.sha256(open(full, "rb").read()).hexdigest()
|
|
print(json.dumps(files, sort_keys=True))
|
|
"""
|
|
|
|
MODEL_INFO = """
|
|
import json
|
|
from huggingface_hub import HfApi
|
|
info = HfApi().model_info({repo!r}, revision={revision!r})
|
|
print(json.dumps({{"sha": info.sha, "siblings": sorted(s.rfilename for s in info.siblings)}}))
|
|
"""
|
|
|
|
FILE_METADATA = """
|
|
import json
|
|
from huggingface_hub import get_hf_file_metadata, hf_hub_url
|
|
url = hf_hub_url({repo!r}, "config.json", revision={revision!r})
|
|
meta = get_hf_file_metadata(url)
|
|
print(json.dumps({{"commit": meta.commit_hash, "etag": meta.etag, "size": meta.size}}))
|
|
"""
|
|
|
|
MISSING_FILE = """
|
|
import json
|
|
from huggingface_hub import hf_hub_download
|
|
from huggingface_hub.errors import EntryNotFoundError, RepositoryNotFoundError, RevisionNotFoundError
|
|
try:
|
|
hf_hub_download({repo!r}, "no-such-file.bin", revision={revision!r})
|
|
except EntryNotFoundError:
|
|
print(json.dumps({{"raised": "EntryNotFoundError"}}))
|
|
except (RepositoryNotFoundError, RevisionNotFoundError) as err:
|
|
print(json.dumps({{"raised": type(err).__name__}}))
|
|
else:
|
|
print(json.dumps({{"raised": None}}))
|
|
"""
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--binary", type=Path, default=Path("target/release/rustingface"))
|
|
parser.add_argument("--s3-endpoint")
|
|
parser.add_argument("--s3-bucket")
|
|
parser.add_argument("--keep", action="store_true", help="keep the work directory")
|
|
args = parser.parse_args()
|
|
|
|
if not args.binary.exists():
|
|
print(f"{args.binary} does not exist; build it with `cargo build --release`", file=sys.stderr)
|
|
return 2
|
|
|
|
results = Results()
|
|
workdir = Path(tempfile.mkdtemp(prefix="rustingface-conformance-"))
|
|
print(f"work directory: {workdir}")
|
|
|
|
bucket_dir = workdir / "bucket"
|
|
bucket_dir.mkdir()
|
|
hf_home = workdir / "hf"
|
|
|
|
if args.s3_endpoint:
|
|
storage = {
|
|
"endpoint": json.dumps(args.s3_endpoint),
|
|
"bucket": json.dumps(args.s3_bucket or "rustingface-conformance"),
|
|
"region": '"us-east-1"',
|
|
"path_style": "true",
|
|
"multipart_part_size": '"5MiB"',
|
|
}
|
|
for var, key in (("AWS_ACCESS_KEY_ID", "access_key_id_file"),
|
|
("AWS_SECRET_ACCESS_KEY", "secret_access_key_file")):
|
|
value = os.environ.get(var)
|
|
if not value:
|
|
print(f"{var} must be set to run against S3", file=sys.stderr)
|
|
return 2
|
|
secret = workdir / key
|
|
secret.write_text(value)
|
|
storage[key] = json.dumps(str(secret))
|
|
else:
|
|
storage = {"local_path": json.dumps(str(bucket_dir)), "multipart_part_size": '"5MiB"'}
|
|
|
|
open_upstream = {"enabled": "true", "endpoint": '"https://huggingface.co"'}
|
|
unreachable_upstream = {"enabled": "true", "endpoint": f'"http://127.0.0.1:{closed_port()}"',
|
|
"connect_timeout": '"2s"'}
|
|
sealed_upstream = {"enabled": "false"}
|
|
|
|
# ------------------------------------------------------------------
|
|
# The client's own constants, re-checked rather than assumed.
|
|
# ------------------------------------------------------------------
|
|
def check_error_codes() -> None:
|
|
import huggingface_hub
|
|
|
|
print(f" huggingface_hub {huggingface_hub.__version__}")
|
|
source = Path(huggingface_hub.__file__).parent
|
|
found = set()
|
|
for path in source.rglob("*.py"):
|
|
text = path.read_text(errors="ignore")
|
|
for code in EXPECTED_ERROR_CODES:
|
|
if f'"{code}"' in text or f"'{code}'" in text:
|
|
found.add(code)
|
|
missing = EXPECTED_ERROR_CODES - found
|
|
if missing:
|
|
raise Failure(
|
|
f"this client does not mention {sorted(missing)}; the X-Error-Code strings "
|
|
"rustingface emits are version-coupled and need review"
|
|
)
|
|
http = (source / "utils" / "_http.py").read_text()
|
|
if DISABLED_REPO_MESSAGE not in http:
|
|
raise Failure(
|
|
f"this client no longer dispatches DisabledRepoError on "
|
|
f"{DISABLED_REPO_MESSAGE!r}; rustingface-api/src/error.rs needs review"
|
|
)
|
|
print(f" {len(found)} codes plus the disabled-repo message all match")
|
|
|
|
results.check("X-Error-Code strings match the pinned client", check_error_codes)
|
|
|
|
baseline: dict = {}
|
|
snapshot_baseline: dict = {}
|
|
|
|
# ------------------------------------------------------------------
|
|
# 1. Cold pull through rustingface.
|
|
# ------------------------------------------------------------------
|
|
def cold_pull() -> None:
|
|
nonlocal baseline, snapshot_baseline
|
|
wipe_client_caches(hf_home)
|
|
with Instance(args.binary, workdir, storage, open_upstream) as rf:
|
|
baseline = json.loads(
|
|
run_client(DOWNLOAD_ONE.format(repo=REPO, revision=REVISION), rf.endpoint, hf_home)
|
|
)
|
|
snapshot_baseline = json.loads(
|
|
run_client(SNAPSHOT.format(repo=REPO, revision=REVISION), rf.endpoint, hf_home)
|
|
)
|
|
print(f" {len(snapshot_baseline)} files in the snapshot")
|
|
listing = rf.cli("list")
|
|
if REPO not in listing:
|
|
raise Failure(f"`rustingface list` does not mention {REPO}:\n{listing}")
|
|
if "rustingface_bytes_served_total" not in rf.metrics():
|
|
raise Failure("/metrics did not expose the served-bytes counter")
|
|
|
|
results.check("a cold pull succeeds and lands in the bucket", cold_pull)
|
|
|
|
# ------------------------------------------------------------------
|
|
# 2-4. Cut off upstream, wipe the client, pull again.
|
|
# ------------------------------------------------------------------
|
|
def sealed_pull() -> None:
|
|
wipe_client_caches(hf_home)
|
|
with Instance(args.binary, workdir, storage, sealed_upstream) as rf:
|
|
again = json.loads(
|
|
run_client(DOWNLOAD_ONE.format(repo=REPO, revision=REVISION), rf.endpoint, hf_home)
|
|
)
|
|
if again != baseline:
|
|
raise Failure(f"sealed pull differs from the first: {again} != {baseline}")
|
|
snapshot_again = json.loads(
|
|
run_client(SNAPSHOT.format(repo=REPO, revision=REVISION), rf.endpoint, hf_home)
|
|
)
|
|
if snapshot_again != snapshot_baseline:
|
|
missing = set(snapshot_baseline) - set(snapshot_again)
|
|
raise Failure(f"sealed snapshot differs; missing {sorted(missing)}")
|
|
|
|
results.check("a sealed instance serves the identical snapshot", sealed_pull)
|
|
|
|
# ------------------------------------------------------------------
|
|
# 5. A brand-new process against the same bucket.
|
|
# ------------------------------------------------------------------
|
|
def portable_state() -> None:
|
|
wipe_client_caches(hf_home)
|
|
# Nothing carries over: a fresh process, a fresh client cache, only the
|
|
# bucket in common. This is the step that tests sovereignty rather
|
|
# than a cache.
|
|
with Instance(args.binary, workdir, storage, sealed_upstream) as rf:
|
|
again = json.loads(
|
|
run_client(DOWNLOAD_ONE.format(repo=REPO, revision=REVISION), rf.endpoint, hf_home)
|
|
)
|
|
if again != baseline:
|
|
raise Failure(f"reinstalled instance differs: {again} != {baseline}")
|
|
|
|
results.check("a reinstalled instance serves from the bucket alone", portable_state)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Metadata replay.
|
|
# ------------------------------------------------------------------
|
|
def metadata_replay() -> None:
|
|
with Instance(args.binary, workdir, storage, open_upstream) as rf:
|
|
live = json.loads(
|
|
run_client(FILE_METADATA.format(repo=REPO, revision=REVISION), rf.endpoint, hf_home)
|
|
)
|
|
with Instance(args.binary, workdir, storage, sealed_upstream) as rf:
|
|
replayed = json.loads(
|
|
run_client(FILE_METADATA.format(repo=REPO, revision=REVISION), rf.endpoint, hf_home)
|
|
)
|
|
if live != replayed:
|
|
raise Failure(f"file metadata changed once sealed: {live} != {replayed}")
|
|
print(f" commit {live['commit']} etag {live['etag']} size {live['size']}")
|
|
|
|
results.check("file metadata replays byte-identically when sealed", metadata_replay)
|
|
|
|
def model_info() -> None:
|
|
with Instance(args.binary, workdir, storage, open_upstream) as rf:
|
|
info = json.loads(
|
|
run_client(MODEL_INFO.format(repo=REPO, revision=REVISION), rf.endpoint, hf_home)
|
|
)
|
|
if not info["sha"]:
|
|
raise Failure("model_info returned no commit sha")
|
|
print(f" {info['sha']} with {len(info['siblings'])} siblings")
|
|
|
|
results.check("model_info answers through rustingface", model_info)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Failures are loud.
|
|
# ------------------------------------------------------------------
|
|
def missing_file_raises() -> None:
|
|
with Instance(args.binary, workdir, storage, sealed_upstream) as rf:
|
|
result = json.loads(
|
|
run_client(MISSING_FILE.format(repo=REPO, revision=REVISION), rf.endpoint, hf_home)
|
|
)
|
|
if result["raised"] is None:
|
|
raise Failure("a missing file returned successfully instead of raising")
|
|
print(f" client raised {result['raised']}")
|
|
|
|
results.check("a missing file raises a typed client exception", missing_file_raises)
|
|
|
|
def unreachable_upstream_fails_loudly() -> None:
|
|
wipe_client_caches(hf_home)
|
|
with Instance(args.binary, workdir, storage, unreachable_upstream) as rf:
|
|
# Something already stored still serves.
|
|
served = json.loads(
|
|
run_client(DOWNLOAD_ONE.format(repo=REPO, revision=REVISION), rf.endpoint, hf_home)
|
|
)
|
|
if served != baseline:
|
|
raise Failure("a stored file stopped serving when upstream became unreachable")
|
|
# Something not stored fails, rather than returning an empty file.
|
|
result = json.loads(
|
|
run_client(MISSING_FILE.format(repo=REPO, revision=REVISION), rf.endpoint, hf_home)
|
|
)
|
|
if result["raised"] is None:
|
|
raise Failure("an unfetchable file returned successfully")
|
|
|
|
results.check(
|
|
"an unreachable upstream serves what is stored and refuses what is not",
|
|
unreachable_upstream_fails_loudly,
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# The hf CLI.
|
|
# ------------------------------------------------------------------
|
|
def hf_cli() -> None:
|
|
with Instance(args.binary, workdir, storage, sealed_upstream) as rf:
|
|
env = client_env(rf.endpoint, hf_home)
|
|
whoami = subprocess.run(
|
|
["hf", "auth", "whoami"], capture_output=True, text=True, env=env
|
|
)
|
|
# `hf auth whoami` without a token is allowed to report "not logged
|
|
# in"; what must not happen is a crash against /api/whoami-v2.
|
|
if "Traceback" in whoami.stderr:
|
|
raise Failure(f"hf auth whoami crashed:\n{whoami.stderr}")
|
|
download = subprocess.run(
|
|
["hf", "download", REPO, "config.json", "--revision", REVISION],
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
)
|
|
if download.returncode != 0:
|
|
raise Failure(f"hf download failed:\n{download.stdout}\n{download.stderr}")
|
|
|
|
if shutil.which("hf"):
|
|
results.check("the hf CLI downloads through rustingface", hf_cli)
|
|
else:
|
|
print("\n=== the hf CLI downloads through rustingface\n skipped: hf not on PATH")
|
|
|
|
# ------------------------------------------------------------------
|
|
# safetensors, the path an inference stack actually takes.
|
|
# ------------------------------------------------------------------
|
|
def safetensors_load() -> None:
|
|
script = """
|
|
import glob, json, os
|
|
from huggingface_hub import snapshot_download
|
|
from safetensors import safe_open
|
|
root = snapshot_download({repo!r}, revision={revision!r})
|
|
files = glob.glob(os.path.join(root, "**", "*.safetensors"), recursive=True)
|
|
if not files:
|
|
print(json.dumps({{"tensors": 0, "note": "repo has no safetensors"}}))
|
|
else:
|
|
total = 0
|
|
for path in files:
|
|
with safe_open(path, framework="np") as handle:
|
|
total += len(handle.keys())
|
|
print(json.dumps({{"tensors": total}}))
|
|
""".format(repo=REPO, revision=REVISION)
|
|
with Instance(args.binary, workdir, storage, sealed_upstream) as rf:
|
|
out = json.loads(run_client(script, rf.endpoint, hf_home))
|
|
print(f" {out}")
|
|
|
|
results.check("safetensors files load from a sealed snapshot", safetensors_load)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Admin operations against the same bucket.
|
|
# ------------------------------------------------------------------
|
|
def admin_operations() -> None:
|
|
with Instance(args.binary, workdir, storage, sealed_upstream) as rf:
|
|
show = rf.cli("show", REPO)
|
|
manifest = json.loads(show)
|
|
if not manifest["entries"]:
|
|
raise Failure("the manifest records no entries")
|
|
verify = rf.cli("verify", REPO)
|
|
if "0 corrupt" not in verify:
|
|
raise Failure(f"verify reported problems:\n{verify}")
|
|
dry = rf.cli("gc", "--dry-run")
|
|
if "0 reclaimable" not in dry:
|
|
raise Failure(f"gc found reclaimable blobs while everything is referenced:\n{dry}")
|
|
|
|
results.check("show, verify and gc agree with what was stored", admin_operations)
|
|
|
|
if not args.keep:
|
|
shutil.rmtree(workdir, ignore_errors=True)
|
|
else:
|
|
print(f"\nwork directory kept at {workdir}")
|
|
return results.report()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|