grenadeandClaude Opus 5
059c781422
The suite's first run on a runner reported "10 passed, 0 failed" and
exited 0. Locally it is 11. The difference is the `hf` CLI check, which
did not run: the workflow invokes .venv/bin/python directly without
putting the venv's bin on PATH, so shutil.which("hf") found nothing and
the check skipped.
A skipped check reads exactly like a passing one, and this is the suite
that runs unattended on a weekly timer specifically to notice when a
client changes underneath us -- so a silent skip defeats its purpose.
Exports the venv's bin in the workflow and asserts `hf version` before
the suite, so a missing CLI fails the job outright rather than being
quietly stepped over. Also counts skips and lists them in the summary,
where they cannot be mistaken for passes.
Verified both ways: with the venv off PATH the suite now reports
"10 passed, 0 failed, 1 skipped" and names it; with it on PATH, 11
passed and 0 skipped.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XZG2i4AmfSqE97EJGBVb64
572 lines
22 KiB
Python
Executable File
572 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)
|
|
skipped: list[tuple[str, str]] = field(default_factory=list)
|
|
|
|
def skip(self, name: str, why: str) -> None:
|
|
print(f"\n=== {name}\n skipped: {why}")
|
|
self.skipped.append((name, why))
|
|
|
|
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, "
|
|
f"{len(self.skipped)} skipped"
|
|
)
|
|
for name, err in self.failed:
|
|
print(f" FAILED {name}: {err}")
|
|
# Surfaced in the summary, not just where it happened: a skipped check
|
|
# reads exactly like a passing one in a scrollback, and this suite runs
|
|
# unattended on a timer.
|
|
for name, why in self.skipped:
|
|
print(f" SKIPPED {name}: {why}")
|
|
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:
|
|
results.skip(
|
|
"the hf CLI downloads through rustingface",
|
|
"hf not on PATH (is the venv's bin directory exported?)",
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# 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())
|