rob thijssen
32c8a3c2b5
The suite set `HF_HUB_DISABLE_XET=1` in the client environment. That is not a configuration anyone runs: `hf_xet` installs as a `huggingface_hub` dependency, so a real user's client is Xet-capable whether they asked for it or not. Every conformance run so far has therefore been proving something about a client with Xet switched off by the harness, and nothing at all about the one rustingface has to work with. That matters more than it sounds. Downstream Xet behaviour is half of spec §8 and it is the half exposed to the Hub's migration: rustingface implements no Xet, so the whole "point HF_ENDPOINT at it and everything works" claim rests on a Xet-capable client *choosing* HTTP. That choice was untested. Stop disabling it, and add a check that asserts the choice from inside the client rather than by inspecting our own headers -- the question is not which headers we sent but what the client concluded from them. It refuses to pass vacuously: if `hf_xet` is absent it fails loudly rather than quietly proving nothing, which is how the harness's own opt-out was found. The full suite passes with Xet enabled, 12 checks. So the posture in spec §8 holds against a real client, and now says so. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PqNtYNhov3fukx46KS9R7L
633 lines
26 KiB
Python
Executable File
633 lines
26 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}"
|
|
|
|
[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"),
|
|
# Xet is deliberately *not* disabled here. `hf_xet` installs as a
|
|
# huggingface_hub dependency, so a real user's client is Xet-capable
|
|
# whether or not they asked for it, and the whole claim rustingface
|
|
# makes is that pointing HF_ENDPOINT at it works with the client
|
|
# people actually have. Setting HF_HUB_DISABLE_XET would test a
|
|
# configuration nobody runs, and would hide the one failure mode
|
|
# that matters here: a client choosing Xet against a service that
|
|
# implements none of it (spec §8).
|
|
"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}}))
|
|
"""
|
|
|
|
# The Xet negotiation, from the client's own side. `hf_xet` ships as a
|
|
# huggingface_hub dependency, so a real user's client is Xet-capable whether or
|
|
# not they asked for it. What keeps rustingface compatible is that the *server*
|
|
# never advertises Xet: `parse_xet_file_data_from_response` returns None, and
|
|
# `hf_hub_download` dispatches to plain HTTP (file_download.py, "if
|
|
# xet_file_data is not None and is_xet_available()").
|
|
#
|
|
# Asserted from inside the client rather than by inspecting headers ourselves,
|
|
# because the question is not "which headers did we send" but "what did the
|
|
# client conclude from them".
|
|
XET_FALLBACK = """
|
|
import json
|
|
from huggingface_hub import get_session, hf_hub_url
|
|
from huggingface_hub.utils._runtime import is_xet_available
|
|
from huggingface_hub.utils._xet import parse_xet_file_data_from_response
|
|
|
|
url = hf_hub_url({repo!r}, "config.json", revision={revision!r})
|
|
response = get_session().get(url, headers={{"Accept-Encoding": "identity"}}, allow_redirects=False)
|
|
response.raise_for_status()
|
|
print(json.dumps({{
|
|
"xet_capable": bool(is_xet_available()),
|
|
"xet_offered": parse_xet_file_data_from_response(response) is not None,
|
|
"xet_headers": sorted(h for h in response.headers if h.lower().startswith("x-xet-")),
|
|
}}))
|
|
"""
|
|
|
|
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)
|
|
|
|
def xet_capable_client_falls_back_to_http() -> None:
|
|
"""The central compatibility claim: point HF_ENDPOINT here and it works.
|
|
|
|
Xet is the thing most likely to break that quietly. The Hub is moving to
|
|
it, `hf_xet` is installed alongside `huggingface_hub` whether or not the
|
|
user asked for it, and rustingface implements none of it (spec §8) --
|
|
deliberately, because a Xet-backed fetch would leave the bucket
|
|
unreadable without a Xet client. That only holds while the client
|
|
*chooses* HTTP, so assert the choice rather than assume it.
|
|
"""
|
|
with Instance(args.binary, workdir, storage, open_upstream) as rf:
|
|
seen = json.loads(
|
|
run_client(XET_FALLBACK.format(repo=REPO, revision=REVISION), rf.endpoint, hf_home)
|
|
)
|
|
if not seen["xet_capable"]:
|
|
raise Failure(
|
|
"hf_xet is not installed, so this check proves nothing: it would pass against a "
|
|
"client that could not use Xet even if rustingface offered it. Install the pinned "
|
|
"requirements."
|
|
)
|
|
if seen["xet_offered"] or seen["xet_headers"]:
|
|
raise Failure(
|
|
"rustingface advertised Xet to the client "
|
|
f"(headers {seen['xet_headers']}). It implements none of it, and a client that "
|
|
"took that path would write a bucket nobody can read back without a Xet client, "
|
|
"which forfeits state portability (spec §8)."
|
|
)
|
|
print(" client is Xet-capable, rustingface offered no Xet, download stays on HTTP")
|
|
|
|
results.check("a Xet-capable client falls back to plain HTTP", xet_capable_client_falls_back_to_http)
|
|
|
|
# ------------------------------------------------------------------
|
|
# 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())
|