net-kingdom/tools/contract-tests/fetch_owners.py

#!/usr/bin/env python3
"""Fetch owner source for the bounded-execution contract tests (NK-WP-0058-T02). Read-only, no credentials.

  python3 tools/contract-tests/fetch_owners.py --dest DIR [--ref pinned|main]   # CI: owners next to net-kingdom
  python3 tools/contract-tests/fetch_owners.py --drift                          # local: pins vs sibling checkouts
  python3 tools/contract-tests/fetch_owners.py --bump                           # local: pin the siblings' HEADs

`contracts/owner-pins.json` names each owner's commit. CI on push tests against the pins, so an owner change
reaches net-kingdom when a pin moves; the nightly run tests against each owner's main, so a breaking owner change
fails within a day. Owners are anonymous partial clones of public Forgejo repositories (no credentials, no
prompt). --bump refuses a sibling with uncommitted or unpushed work.
"""
import argparse
import json
import subprocess
import sys
from pathlib import Path

REPO = Path(__file__).resolve().parents[2]
PINS = REPO / 'contracts/owner-pins.json'


def load():
    """The pins as {repo: commit}. The file lists objects (repo, commit) rather than repo-named keys: a key like
    'key-cape' or 'access-engine' next to a hex value reads as a credential to the commit-time secret scanner."""
    doc = json.loads(PINS.read_text())
    return {**doc, 'owners': {o['repo']: o['commit'] for o in doc['owners']}}


def fetch(dest, ref='pinned', run=subprocess.run):
    """A partial clone of each owner (no blobs until needed), checked out at the pin or at `ref`. History stays
    available: the readiness check reads informed-decision at the revision Argo pins."""
    pins = load()
    dest.mkdir(parents=True, exist_ok=True)
    fetched = {}
    env = {'GIT_TERMINAL_PROMPT': '0', 'PATH': '/usr/local/bin:/usr/bin:/bin'}
    for owner, commit in pins['owners'].items():
        target = dest / owner
        rev = commit if ref == 'pinned' else 'origin/' + ref
        for argv in (['git', 'clone', '--quiet', '--filter=blob:none', '--no-checkout',
                      f"{pins['base_url']}/{owner}.git", str(target)],
                     ['git', '-C', str(target), 'checkout', '--quiet', '--detach', rev]):
            if run(argv, capture_output=True, env=env).returncode:
                raise SystemExit(f'{owner}: {argv[1] if argv[1] != "-C" else "checkout"} failed')
        fetched[owner] = run(['git', '-C', str(target), 'rev-parse', 'HEAD'], capture_output=True,
                             text=True).stdout.strip()
    return fetched


def git(path, *args):
    return subprocess.run(['git', '-C', str(path), *args], capture_output=True, text=True)


def drift(siblings=REPO.parent):
    out = {}
    for owner, commit in load()['owners'].items():
        head = git(siblings / owner, 'rev-parse', 'HEAD').stdout.strip()
        out[owner] = 'pinned' if head == commit else ('absent' if not head else f'sibling_at_{head[:12]}')
    return out


def bump(siblings=REPO.parent):
    pins = load()
    for owner in pins['owners']:
        path = siblings / owner
        if git(path, 'status', '--porcelain').stdout.strip():
            raise SystemExit(f'{owner}: uncommitted work; not pinned')
        if git(path, 'rev-list', '--count', '@{u}..HEAD').stdout.strip() not in ('0', ''):
            raise SystemExit(f'{owner}: unpushed commits; not pinned')
        pins['owners'][owner] = git(path, 'rev-parse', 'HEAD').stdout.strip()
    doc = {**pins, 'owners': [{'repo': o, 'commit': c} for o, c in pins['owners'].items()]}
    PINS.write_text(json.dumps(doc, indent=2) + '\n')
    return {owner: commit[:12] for owner, commit in pins['owners'].items()}


def main(argv=None):
    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument('--dest', type=Path)
    p.add_argument('--ref', default='pinned', help="'pinned' (default) or a branch name, e.g. main")
    p.add_argument('--drift', action='store_true')
    p.add_argument('--bump', action='store_true')
    a = p.parse_args(argv)
    if a.drift:
        print(json.dumps(drift()))
    elif a.bump:
        print(json.dumps(bump()))
    elif a.dest:
        print(json.dumps(fetch(a.dest, a.ref)))
    else:
        p.error('give --dest, --drift or --bump')
    return 0


if __name__ == '__main__':
    sys.exit(main())