stack/dev/scripts/vendor_bcda_app.py

"""Sparse vendoring of BCDA upstream artefacts (CMSgov/bcda-app + bcda.cms.gov).

The ``bcda`` module is tested against what CMS actually ships, not
against our reading of it: the v3 handler and its tests, the request
validation that decides which headers and parameters are legal, the
OperationOutcome writer, the v3 Postman smoke collection, the gzip
download fixtures, and the official v3 sample ndjson + data dictionary.
That is a few hundred kilobytes of a 50 MB repository, so this is a
manifest, not a submodule.

Manifest: ``dev/vendor/MANIFEST.toml``

    [[source]]
    name = "bcda-app"
    kind = "github"
    repo = "CMSgov/bcda-app"
    ref = "main"
    commit = "<sha the files were last synced from>"
    files = ["swagger.json", ...]

    [[source]]
    name = "bcda-site"
    kind = "https"
    base = "https://bcda.cms.gov/assets/downloads/"
    files = ["BCDA_v3_Data_Dictionary.xlsx", ...]

    [sha256]
    "bcda-app/swagger.json" = "..."      # local file digests

Usage::

    uv run python dev/scripts/vendor_bcda_app.py --check   # drift? exit 1
    uv run python dev/scripts/vendor_bcda_app.py --sync    # refresh + rewrite manifest
    uv run python dev/scripts/vendor_bcda_app.py --verify  # local tree matches manifest (offline)

``--check`` compares upstream against the manifest without touching
the tree: for GitHub it resolves ``ref`` to a commit and diffs the blob
shas of the listed paths against the pinned commit's tree; for plain
HTTPS it downloads and compares sha256.  The weekly ``bcda-vendor``
workflow runs ``--check`` and files an issue on drift — a payload
change upstream is something a human should read before the
flattener is re-pointed, so nothing is auto-committed.

Standard library only: this runs on a bare runner before ``uv sync``.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import sys
import tomllib
import urllib.request
from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]
VENDOR = ROOT / "dev" / "vendor"
MANIFEST = VENDOR / "MANIFEST.toml"
UA = {"User-Agent": "stack-vendor-bcda/1.0 (+https://git.fhirworx.io/homelab/stack)"}


# ── I/O ─────────────────────────────────────────────────────────


def _get(url: str, *, accept: str | None = None) -> bytes:
    headers = dict(UA)
    if accept:
        headers["Accept"] = accept
    req = urllib.request.Request(url, headers=headers)
    with urllib.request.urlopen(req, timeout=120) as resp:  # noqa: S310 — https only
        return resp.read()


def _sha256(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def _git_blob_sha(data: bytes) -> str:
    """Git's blob object id for ``data`` (what the trees API reports)."""
    header = f"blob {len(data)}\0".encode()
    return hashlib.sha1(header + data).hexdigest()  # noqa: S324 — git's own id


def load_manifest(path: Path | None = None) -> dict:
    return tomllib.loads((path or MANIFEST).read_text())


def dump_manifest(m: dict, path: Path | None = None) -> None:
    """Write the manifest back.  Hand-rolled TOML: two array-of-table
    sources plus one flat string table — no dependency needed."""
    lines = [
        "# Sparse vendoring manifest — regenerate with",
        "#   uv run python dev/scripts/vendor_bcda_app.py --sync",
        "# Check for upstream drift (no changes written) with --check.",
        "",
    ]
    for src in m["source"]:
        lines.append("[[source]]")
        for key in ("name", "kind", "repo", "ref", "commit", "base", "license"):
            if key in src:
                lines.append(f'{key} = "{src[key]}"')
        lines.append("files = [")
        for f in src["files"]:
            lines.append(f'  "{f}",')
        lines.append("]")
        lines.append("")
    lines.append("[sha256]")
    for key in sorted(m.get("sha256", {})):
        lines.append(f'"{key}" = "{m["sha256"][key]}"')
    lines.append("")
    (path or MANIFEST).write_text("\n".join(lines))


# ── GitHub source ───────────────────────────────────────────────


def _gh_json(url: str) -> dict:
    return json.loads(_get(url, accept="application/vnd.github+json"))


def _gh_head_commit(repo: str, ref: str) -> str:
    return _gh_json(f"https://api.github.com/repos/{repo}/commits/{ref}")["sha"]


def _gh_tree(repo: str, commit: str) -> dict[str, str]:
    """``{path: blob_sha}`` for every blob at ``commit``."""
    data = _gh_json(
        f"https://api.github.com/repos/{repo}/git/trees/{commit}?recursive=1"
    )
    if data.get("truncated"):
        raise RuntimeError(f"{repo}@{commit}: tree listing truncated")
    return {e["path"]: e["sha"] for e in data["tree"] if e["type"] == "blob"}


def _gh_raw(repo: str, commit: str, path: str) -> bytes:
    return _get(f"https://raw.githubusercontent.com/{repo}/{commit}/{path}")


# ── Operations ──────────────────────────────────────────────────


def verify(m: dict) -> list[str]:
    """Offline: every manifest file exists locally with the recorded sha256."""
    problems = []
    for src in m["source"]:
        for f in src["files"]:
            key = f"{src['name']}/{f}"
            local = VENDOR / key
            if not local.exists():
                problems.append(f"missing {key}")
                continue
            want = m.get("sha256", {}).get(key)
            got = _sha256(local.read_bytes())
            if want != got:
                problems.append(f"sha256 mismatch {key}: manifest {want} local {got}")
    return problems


def check(m: dict) -> list[str]:
    """Online: what upstream changed relative to the manifest."""
    drift: list[str] = []
    for src in m["source"]:
        if src["kind"] == "github":
            head = _gh_head_commit(src["repo"], src["ref"])
            if head == src.get("commit"):
                continue
            pinned = _gh_tree(src["repo"], src["commit"])
            current = _gh_tree(src["repo"], head)
            for f in src["files"]:
                if f not in current:
                    drift.append(f"{src['name']}: {f} removed upstream ({head[:12]})")
                elif pinned.get(f) != current[f]:
                    drift.append(
                        f"{src['name']}: {f} changed ({src['commit'][:12]} → {head[:12]})"
                    )
            # v3 handler directory gaining a file we don't track is worth
            # a look too — the point of vendoring the tests is to see them.
            watched = {Path(f).parent.as_posix() for f in src["files"] if "/" in f}
            for path in current:
                parent = Path(path).parent.as_posix()
                if (
                    parent in watched
                    and path not in pinned
                    and path not in src["files"]
                ):
                    drift.append(f"{src['name']}: new upstream file {path}")
        elif src["kind"] == "https":
            for f in src["files"]:
                key = f"{src['name']}/{f}"
                got = _sha256(_get(src["base"] + f))
                if got != m.get("sha256", {}).get(key):
                    drift.append(f"{src['name']}: {f} changed (sha256 {got[:12]})")
        else:  # pragma: no cover — manifest is ours
            raise ValueError(f"unknown source kind {src['kind']!r}")
    return drift


def sync(m: dict) -> list[str]:
    """Download every manifest file from upstream HEAD; rewrite the manifest."""
    changed: list[str] = []
    shas = dict(m.get("sha256", {}))
    for src in m["source"]:
        if src["kind"] == "github":
            commit = _gh_head_commit(src["repo"], src["ref"])
            src["commit"] = commit
            fetch = lambda f, _s=src, _c=commit: _gh_raw(_s["repo"], _c, f)  # noqa: E731
        else:
            fetch = lambda f, _s=src: _get(_s["base"] + f)  # noqa: E731
        for f in src["files"]:
            key = f"{src['name']}/{f}"
            data = fetch(f)
            digest = _sha256(data)
            local = VENDOR / key
            if shas.get(key) != digest or not local.exists():
                local.parent.mkdir(parents=True, exist_ok=True)
                local.write_bytes(data)
                changed.append(key)
            shas[key] = digest
    m["sha256"] = shas
    dump_manifest(m)
    return changed


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    mode = ap.add_mutually_exclusive_group(required=True)
    mode.add_argument(
        "--check", action="store_true", help="report upstream drift (exit 1)"
    )
    mode.add_argument("--sync", action="store_true", help="refresh files + manifest")
    mode.add_argument("--verify", action="store_true", help="offline tree == manifest")
    args = ap.parse_args(argv)
    m = load_manifest()

    if args.verify:
        problems = verify(m)
        for p in problems:
            print(p)
        print("vendor tree OK" if not problems else f"{len(problems)} problem(s)")
        return 1 if problems else 0
    if args.check:
        drift = check(m)
        for d in drift:
            print(d)
        print("no upstream drift" if not drift else f"{len(drift)} upstream change(s)")
        return 1 if drift else 0
    changed = sync(m)
    for c in changed:
        print(f"updated {c}")
    print(f"synced; {len(changed)} file(s) changed")
    return 0


if __name__ == "__main__":  # pragma: no cover
    sys.exit(main())