File size: 2,178 Bytes
29ed697 | 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 | #!/usr/bin/env python3
"""Hydrate and verify Xet-backed assets in the single Nanobrush Git checkout."""
import argparse
import hashlib
from pathlib import Path
import subprocess
import sys
import storage_policy as policy
ROOT = Path(__file__).resolve().parents[1]
def run(*args, capture=False):
return subprocess.run(args, cwd=ROOT, check=True, stdout=subprocess.PIPE if capture else None).stdout
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("command", choices=("check", "fetch"))
args = parser.parse_args()
policy.check_setup()
policy.check_index()
if args.command == "fetch":
run("git", "lfs", "pull", "hf")
# A fresh clone's cache starts empty; verify hydrated working bytes directly.
count = 0
paths = policy.git("ls-files", "-z").split(b"\0")
for raw in paths:
if not raw:
continue
path = raw.decode("utf-8")
if not policy.is_asset(path):
continue
pointer = policy.git("show", ":" + path)
policy.check_asset(path, pointer)
if not pointer:
expected_hash, size = hashlib.sha256(b"").hexdigest(), 0
else:
lines = pointer.decode().splitlines()
expected_hash, size = lines[1].split(":", 1)[1], int(lines[2].split()[1])
payload = ROOT / path
if payload.is_symlink() or not payload.is_file() or payload.stat().st_size != size:
raise ValueError(f"Missing/unhydrated asset: {path}; run assets-fetch")
digest = hashlib.sha256()
with payload.open("rb") as stream:
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
digest.update(chunk)
if digest.hexdigest() != expected_hash:
raise ValueError(f"Modified/unhydrated asset: {path}; preserve edits before fetching")
count += 1
print(f"Verified {count} hydrated Xet/LFS assets against staged pointer identities.")
if __name__ == "__main__":
try:
main()
except (ValueError, OSError, subprocess.CalledProcessError) as error:
print(f"Assets: {error}", file=sys.stderr)
sys.exit(1)
|