diff --git a/.github/scripts/uk_replay_drain.sh b/.github/scripts/uk_replay_drain.sh index 18ec51a..46380d5 100755 --- a/.github/scripts/uk_replay_drain.sh +++ b/.github/scripts/uk_replay_drain.sh @@ -21,18 +21,20 @@ push_state() { return 0 fi git -C "$STATE" commit -qm "$1" - for attempt in 1 2 3 4 5; do + for _ in 1 2 3 4 5; do if git -C "$STATE" push -q origin "HEAD:$STATE_BRANCH"; then return 0 fi - git -C "$STATE" pull -q --rebase origin "$STATE_BRANCH" + git -C "$STATE" pull -q --rebase origin "$STATE_BRANCH" || true done echo "could not push the state branch after five attempts" >&2 return 1 } while :; do - git -C "$STATE" pull -q --rebase origin "$STATE_BRANCH" + if git ls-remote --exit-code --heads origin "$STATE_BRANCH" >/dev/null; then + git -C "$STATE" pull -q --rebase origin "$STATE_BRANCH" + fi request_id=$("${CONTROL[@]}" -m pipeline.uk_replay_queue pending --state "$STATE" --now "$(now)" | head -n 1) if [ -z "$request_id" ]; then echo "queue empty" diff --git a/.github/scripts/uk_replay_enqueue.sh b/.github/scripts/uk_replay_enqueue.sh new file mode 100755 index 0000000..bae5ebe --- /dev/null +++ b/.github/scripts/uk_replay_enqueue.sh @@ -0,0 +1,24 @@ +#!/usr/bin/env bash +# Enqueue each request file given as an argument on the replay-state branch +# (idempotent), then commit and push, rebasing over concurrent writers. +set -euo pipefail +CONTROL=(uv run --with rfc8785==0.1.4 --with jsonschema python) +now=$(date -u +%Y-%m-%dT%H:%M:%SZ) +for request in "$@"; do + "${CONTROL[@]}" -m pipeline.uk_replay_queue enqueue --state state \ + --request "$request" --run-url "$RUN_URL" --now "$now" +done +git -C state add -A +if git -C state diff --cached --quiet; then + echo "nothing new to queue" + exit 0 +fi +git -C state commit -qm "Queue $# request(s)" +for _ in 1 2 3 4 5; do + if git -C state push -q origin "HEAD:$STATE_BRANCH"; then + exit 0 + fi + git -C state pull -q --rebase origin "$STATE_BRANCH" +done +echo "could not push the state branch after five attempts" >&2 +exit 1 diff --git a/.github/scripts/uk_replay_state.sh b/.github/scripts/uk_replay_state.sh new file mode 100755 index 0000000..090c71f --- /dev/null +++ b/.github/scripts/uk_replay_state.sh @@ -0,0 +1,13 @@ +#!/usr/bin/env bash +# Check out the replay-state branch at ./state (create it, orphaned, if it +# does not exist yet) for the UK replay workflow's jobs. +set -euo pipefail +if git ls-remote --exit-code --heads origin "$STATE_BRANCH" >/dev/null; then + git fetch -q origin "$STATE_BRANCH" + git worktree add -q state "origin/$STATE_BRANCH" + git -C state switch -q -C "$STATE_BRANCH" +else + git worktree add -q --orphan -b "$STATE_BRANCH" state +fi +git -C state config user.name "github-actions[bot]" +git -C state config user.email "41898282+github-actions[bot]@users.noreply.github.com" diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2c35025..eb10a19 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -41,7 +41,7 @@ jobs: # pypdf + pyyaml are NOT optional extras: the HMT lane's strongest # honesty guarantee (every registry title and figure anchor verbatim # in the real PDF) runs as a test, and a bare env silently skipped it. - - run: uv run --with pytest --with hypothesis --with pypdf --with pyyaml --with numpy --with psutil --with rfc8785==0.1.4 --with jsonschema pytest tests/ -q + - run: uv run --with pytest --with hypothesis --with pypdf --with pyyaml --with numpy --with psutil --with rfc8785==0.1.4 --with jsonschema --with packaging pytest tests/ -q env: PYTHONPATH: . # REPLAY NAMESPACES: every committed event replay (the legacy diff --git a/.github/workflows/uk-replay.yml b/.github/workflows/uk-replay.yml index 10b77d9..da81726 100644 --- a/.github/workflows/uk-replay.yml +++ b/.github/workflows/uk-replay.yml @@ -12,8 +12,11 @@ name: UK replay qualification # older pending ones, so a request is first written to the persistent queue # on the `replay-state` branch (enqueue job, no concurrency group). The drain # job then runs queued requests one at a time. A scheduled run drains anything -# a cancelled run left behind. A request whose run died mid-way is retried -# after the stale bound in pipeline/uk_replay_queue.py. +# a cancelled run left behind, and first queues a request for any +# policyengine.py release newer than the newest certified variant whose +# dispatch never arrived (pipeline/uk_replay_discover.py). A request whose +# run died mid-way is retried after the stale bound in +# pipeline/uk_replay_queue.py. # # Simulations: one drain job at a time, at most two workers inside it, so # never more than two simulations at once from this workflow. @@ -44,84 +47,72 @@ jobs: name: record the request if: github.event_name == 'repository_dispatch' || (github.event_name == 'workflow_dispatch' && github.event.inputs.request != '') runs-on: ubuntu-latest + env: + PYTHONPATH: . + RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }} steps: - uses: actions/checkout@v4 - uses: astral-sh/setup-uv@v5 - - name: Check out the state branch - run: | - set -euo pipefail - if git ls-remote --exit-code --heads origin "$STATE_BRANCH" >/dev/null; then - git fetch -q origin "$STATE_BRANCH" - git worktree add -q state "origin/$STATE_BRANCH" - git -C state switch -q -C "$STATE_BRANCH" - else - git worktree add -q --orphan -b "$STATE_BRANCH" state - fi - git -C state config user.name "github-actions[bot]" - git -C state config user.email "41898282+github-actions[bot]@users.noreply.github.com" + - run: bash .github/scripts/uk_replay_state.sh - name: Write the request env: DISPATCH_REQUEST: ${{ toJSON(github.event.client_payload.request) }} MANUAL_REQUEST: ${{ github.event.inputs.request }} + EVENT_NAME: ${{ github.event_name }} run: | set -euo pipefail - if [ "${{ github.event_name }}" = "repository_dispatch" ]; then + if [ "$EVENT_NAME" = "repository_dispatch" ]; then printf '%s' "$DISPATCH_REQUEST" > "$RUNNER_TEMP/request.json" else printf '%s' "$MANUAL_REQUEST" > "$RUNNER_TEMP/request.json" fi - - name: Enqueue and push + - run: bash .github/scripts/uk_replay_enqueue.sh "$RUNNER_TEMP/request.json" + + discover: + name: catch up on undispatched releases + if: github.event_name == 'schedule' || (github.event_name == 'workflow_dispatch' && github.event.inputs.request == '') + runs-on: ubuntu-latest + env: + PYTHONPATH: . + RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }} + steps: + - uses: actions/checkout@v4 + - uses: astral-sh/setup-uv@v5 + - run: bash .github/scripts/uk_replay_state.sh + - name: Build requests for releases no request names yet env: - PYTHONPATH: . - RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }} + HUGGING_FACE_TOKEN: ${{ secrets.HUGGING_FACE_TOKEN }} run: | set -euo pipefail - now=$(date -u +%Y-%m-%dT%H:%M:%SZ) - uv run --with rfc8785==0.1.4 --with jsonschema python -m pipeline.uk_replay_queue \ - enqueue --state state --request "$RUNNER_TEMP/request.json" --run-url "$RUN_URL" --now "$now" \ - | tee "$RUNNER_TEMP/enqueued.txt" - git -C state add -A - if git -C state diff --cached --quiet; then - echo "already recorded: $(cat "$RUNNER_TEMP/enqueued.txt")" - exit 0 + uv run --with rfc8785==0.1.4 --with jsonschema --with huggingface_hub --with packaging \ + python -m pipeline.uk_replay_discover --state state --work "$RUNNER_TEMP/discover" \ + --output-dir "$RUNNER_TEMP/discovered" --commit "$GITHUB_SHA" --run-id "$GITHUB_RUN_ID" + shopt -s nullglob + requests=("$RUNNER_TEMP"/discovered/*.json) + if [ "${#requests[@]}" -gt 0 ]; then + bash .github/scripts/uk_replay_enqueue.sh "${requests[@]}" + else + echo "no undispatched releases" fi - git -C state commit -qm "Queue $(cut -d' ' -f1 "$RUNNER_TEMP/enqueued.txt")" - for attempt in 1 2 3 4 5; do - git -C state push -q origin "HEAD:$STATE_BRANCH" && exit 0 - git -C state pull -q --rebase origin "$STATE_BRANCH" - done - exit 1 drain: name: run queued requests - needs: [enqueue] - if: always() && (needs.enqueue.result == 'success' || needs.enqueue.result == 'skipped') + needs: [enqueue, discover] + if: always() && contains(fromJSON('["success", "skipped"]'), needs.enqueue.result) && contains(fromJSON('["success", "skipped"]'), needs.discover.result) runs-on: ubuntu-latest timeout-minutes: 340 concurrency: group: uk-replay-drain cancel-in-progress: false + env: + PYTHONPATH: . + RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }} steps: - uses: actions/checkout@v4 - uses: astral-sh/setup-uv@v5 - - name: Check out the state branch - run: | - set -euo pipefail - if ! git ls-remote --exit-code --heads origin "$STATE_BRANCH" >/dev/null; then - echo "no state branch yet; nothing queued" - echo "EMPTY=1" >> "$GITHUB_ENV" - exit 0 - fi - git fetch -q origin "$STATE_BRANCH" - git worktree add -q state "origin/$STATE_BRANCH" - git -C state switch -q -C "$STATE_BRANCH" - git -C state config user.name "github-actions[bot]" - git -C state config user.email "41898282+github-actions[bot]@users.noreply.github.com" + - run: bash .github/scripts/uk_replay_state.sh - name: Drain the queue - if: env.EMPTY != '1' env: - PYTHONPATH: . - RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }} HUGGING_FACE_TOKEN: ${{ secrets.HUGGING_FACE_TOKEN }} REPLAY_RUNNER: github-actions ${{ runner.os }} ${{ runner.arch }} run: bash .github/scripts/uk_replay_drain.sh diff --git a/docs/uk_replay/RELEASE_REPLAY.md b/docs/uk_replay/RELEASE_REPLAY.md index 7695de4..2671e37 100644 --- a/docs/uk_replay/RELEASE_REPLAY.md +++ b/docs/uk_replay/RELEASE_REPLAY.md @@ -309,7 +309,15 @@ before the candidate's identity is known yields a `failed` receipt with no execution block; an unexpected exception also yields a `failed` receipt, so a request never loops. A request still `running` seven hours after its last update (a cancelled or crashed job) is retried. A scheduled run every three -hours drains whatever is left. +hours drains whatever is left, after first queueing a `released_bundle` +request for every policyengine.py release newer than the newest certified +variant that no request names yet (`pipeline/uk_replay_discover.py`, at most +three per run, oldest first). That catches releases whose dispatch never +arrived. Their requests are built from primary artifacts: the PyPI wheel +digests (the model's must equal the one in the wrapper's packaged manifest), +the packaged manifest, and the data release manifest's SHA-256 and dataset +size; the expected release bundle is read from a scratch install, so for a +discovered request the receiver's bundle check repeats that reading. At most one drain job runs, with at most two simulation workers, so this workflow never runs more than two simulations at once. These runs use GitHub diff --git a/pipeline/uk_replay_discover.py b/pipeline/uk_replay_discover.py new file mode 100644 index 0000000..d48ac58 --- /dev/null +++ b/pipeline/uk_replay_discover.py @@ -0,0 +1,294 @@ +"""Catch up on policyengine.py releases whose dispatch never arrived. + +Release qualification (W4) notifies this repository after each release, but +a missing GitHub App, an outage or a cancelled run can lose that dispatch. +The scheduled receiver therefore also looks at PyPI itself: for every +policyengine release newer than the newest certified replay variant that no +request already names, it builds a ``released_bundle`` qualification request +and leaves it for the queue. + +The request is built from primary artifacts: the wheel's SHA-256 from PyPI +(checked against the downloaded bytes), the packaged bundle manifest inside +that wheel (its SHA-256, the UK model and certified dataset), the model and +core wheel digests from PyPI (the model digest must equal the one the +manifest records), and the data release manifest from Hugging Face (its +SHA-256 and the dataset's size). The expected release bundle is read from +the wrapper installed in a scratch environment with tokens removed, so for +a discovered request the receiver's bundle check repeats that reading; the +artifact digests above are the independent part. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import re +import subprocess +import sys +import urllib.request +import zipfile +from pathlib import Path + +from packaging.version import Version + +ROOT = Path(__file__).resolve().parent.parent +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from pipeline import uk_replay_qualify as qualify # noqa: E402 +from pipeline import uk_replay_variants as variants # noqa: E402 + +PYPI = "https://pypi.org/pypi/{package}/json" +MANIFEST_PATH = "policyengine/data/bundle/manifest.json" +MAX_PER_RUN = 3 +EXTRA_PIN = re.compile( + r"^(policyengine-uk|policyengine-core)\s*==\s*([^;\s]+)\s*;\s*extra\s*==\s*\"uk\"$" +) +BUNDLE_PROBE = """ +import json, warnings +warnings.simplefilter("ignore") +from pipeline import uk_replay_variants as variants +variants.isolate_certification_environment() +import policyengine as pe +print(json.dumps(dict(pe.uk.uk_latest.release_bundle), sort_keys=True)) +""" + + +class Skipped(Exception): + """A release that cannot become a valid request; the reason is logged.""" + + +def pypi(package: str, version: str | None = None) -> dict: + url = PYPI.format(package=package if version is None else f"{package}/{version}") + with urllib.request.urlopen(url, timeout=60) as response: # noqa: S310 - fixed https host + return json.loads(response.read()) + + +def wheel_entry(document: dict) -> dict: + wheels = [u for u in document["urls"] if u["packagetype"] == "bdist_wheel"] + if len(wheels) != 1: + raise Skipped(f"expected one wheel, found {len(wheels)}") + return { + "filename": wheels[0]["filename"], + "url": wheels[0]["url"], + "sha256": wheels[0]["digests"]["sha256"], + } + + +def requested_versions(state: Path) -> set[str]: + """Wrapper versions that already have a released_bundle request.""" + out = set() + for path in (state / "requests").glob("*/REQUEST.json"): + request = json.loads(path.read_text(encoding="utf-8")) + if request.get("kind") == "released_bundle": + out.add(request["candidate"]["policyengine"]["version"]) + return out + + +def candidates(all_versions: list[str], document: dict, state: Path) -> list[str]: + """Releases newer than the newest certified variant's wrapper, not yet + requested, oldest first (so the queue replays them in release order).""" + certified = [ + Version(r["identity"]["packages"]["policyengine"]["version"]) + for r in document["variants"] + if r["status"] == "certified" + ] + floor = max(certified) + seen = requested_versions(state) | {str(v) for v in certified} + out = [] + for value in set(all_versions): + try: + version = Version(value) + except ValueError: + continue + if version > floor and not version.is_prerelease and value not in seen: + out.append(value) + return sorted(out, key=Version) + + +def uk_pins(metadata: str) -> dict[str, str]: + pins = {} + for line in metadata.splitlines(): + if line.startswith("Requires-Dist:"): + match = EXTRA_PIN.match(line.removeprefix("Requires-Dist:").strip()) + if match: + pins[match.group(1)] = match.group(2) + if set(pins) != {"policyengine-uk", "policyengine-core"}: + raise Skipped(f"the wheel's uk extra pins {pins}, not exactly model and core") + return pins + + +def read_wheel(path: Path) -> tuple[bytes, str]: + with zipfile.ZipFile(path) as wheel: + manifest = wheel.read(MANIFEST_PATH) + metadata = next( + wheel.read(name).decode() + for name in wheel.namelist() + if name.endswith(".dist-info/METADATA") + ) + return manifest, metadata + + +def fetch(url: str, destination: Path, sha256: str) -> Path: + destination.parent.mkdir(parents=True, exist_ok=True) + with urllib.request.urlopen(url, timeout=300) as response: # noqa: S310 - PyPI file URL + payload = response.read() + if hashlib.sha256(payload).hexdigest() != sha256: + raise Skipped(f"{destination.name} bytes differ from PyPI's digest") + destination.write_bytes(payload) + return destination + + +def release_manifest(data_package: dict, *, token: str | None) -> bytes: + from huggingface_hub import hf_hub_download + + try: + path = hf_hub_download( + repo_id=data_package["repo_id"], + repo_type=data_package.get("repo_type") or "model", + filename=data_package["release_manifest_path"], + revision=data_package.get("release_manifest_revision") + or data_package["version"], + token=token or None, + ) + except Exception as exc: # noqa: BLE001 - recorded as a skip + raise Skipped( + f"data release manifest unavailable: {type(exc).__name__}" + ) from exc + return Path(path).read_bytes() + + +def provisional_request( + version: str, work: Path, *, token: str | None, requested_by: dict +) -> dict: + """Everything but the expected release bundle, from primary artifacts.""" + wrapper = wheel_entry(pypi("policyengine", version)) + wheel = fetch(wrapper["url"], work / wrapper["filename"], wrapper["sha256"]) + manifest_bytes, metadata = read_wheel(wheel) + uk = json.loads(manifest_bytes)["data_releases"]["uk"] + pins = uk_pins(metadata) + model = uk["model_package"] + if model["version"] != pins["policyengine-uk"]: + raise Skipped( + f"manifest model {model['version']} differs from the uk extra {pins['policyengine-uk']}" + ) + uk_wheel = wheel_entry(pypi("policyengine-uk", model["version"])) + if model.get("sha256") and model["sha256"] != uk_wheel["sha256"]: + raise Skipped("the manifest's model wheel digest differs from PyPI's") + core_wheel = wheel_entry(pypi("policyengine-core", pins["policyengine-core"])) + artifact = uk["certified_data_artifact"] + dataset = uk["datasets"][artifact["dataset"]] + manifest_payload = release_manifest(uk["data_package"], token=token) + released = json.loads(manifest_payload)["artifacts"].get(artifact["dataset"]) + if released is None or released.get("sha256") != artifact["sha256"]: + raise Skipped("the data release manifest does not certify the bundle's dataset") + return { + "schema": "policyengine.uk-replay-qualification-request/v1", + "kind": "released_bundle", + "attempt": 1, + "candidate": { + "policyengine": { + "version": version, + "source": {"pypi": {"wheel_sha256": wrapper["sha256"]}}, + "packaged_bundle_manifest_sha256": hashlib.sha256( + manifest_bytes + ).hexdigest(), + }, + "policyengine_uk": { + "version": model["version"], + "wheel_sha256": uk_wheel["sha256"], + }, + "policyengine_core": { + "version": pins["policyengine-core"], + "wheel_sha256": core_wheel["sha256"], + }, + "data": { + "build_id": artifact["build_id"], + "dataset": artifact["dataset"], + "repo_id": dataset["repo_id"], + "repo_type": uk["data_package"].get("repo_type") or "model", + "revision": dataset["revision"], + "path": dataset["path"], + "sha256": artifact["sha256"], + "size_bytes": released["size_bytes"], + "release_manifest_sha256": hashlib.sha256(manifest_payload).hexdigest(), + }, + "expected_release_bundle": {}, + }, + "compare_to": "latest_certified", + "requested_by": requested_by, + } + + +def expected_bundle(request: dict, work: Path) -> dict: + """Install the release exactly as the receiver will, then read its bundle.""" + venv = work / "venv" + qualify.prepare(request, venv, work / "wheels") + result = subprocess.run( + [str(venv / "bin" / "python"), "-c", BUNDLE_PROBE], + cwd=ROOT, + env={**os.environ, "PYTHONPATH": str(ROOT)}, + check=True, + capture_output=True, + text=True, + ) + return json.loads(result.stdout.strip().splitlines()[-1]) + + +def discover( + state: Path, work: Path, *, token: str | None, requested_by: dict +) -> list[dict]: + document = variants.load_document() + versions = candidates(list(pypi("policyengine")["releases"]), document, state) + if len(versions) > MAX_PER_RUN: + print( + f"{len(versions)} unreplayed releases; building the oldest {MAX_PER_RUN} now, " + f"the rest on later runs: {versions[MAX_PER_RUN:]}" + ) + requests = [] + for version in versions[:MAX_PER_RUN]: + try: + request = provisional_request( + version, work / version, token=token, requested_by=requested_by + ) + request["candidate"]["expected_release_bundle"] = expected_bundle( + request, work / version + ) + qualify.request_id(request) + except (Skipped, qualify.Rejected, subprocess.CalledProcessError) as exc: + print(f"policyengine {version}: not requested: {exc}") + continue + requests.append(request) + print(f"policyengine {version}: request {qualify.request_id(request)}") + return requests + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + parser.add_argument("--state", type=Path, required=True) + parser.add_argument("--work", type=Path, required=True) + parser.add_argument("--output-dir", type=Path, required=True) + parser.add_argument("--commit", required=True) + parser.add_argument("--run-id", type=int) + args = parser.parse_args(argv) + requested_by = { + "repository": "PolicyEngine/policyengine-scorecard", + "commit": args.commit, + "trigger": "scheduled catch-up of an undispatched policyengine.py release", + } + if args.run_id: + requested_by["workflow_run_id"] = args.run_id + token = os.environ.get("HUGGING_FACE_TOKEN") or os.environ.get("HF_TOKEN") + args.output_dir.mkdir(parents=True, exist_ok=True) + for request in discover( + args.state, args.work, token=token, requested_by=requested_by + ): + path = args.output_dir / f"{qualify.request_id(request)}.json" + path.write_bytes(qualify.canonical_bytes(request)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_uk_replay_discover.py b/tests/test_uk_replay_discover.py new file mode 100644 index 0000000..461ac2d --- /dev/null +++ b/tests/test_uk_replay_discover.py @@ -0,0 +1,280 @@ +"""Catch-up discovery of undispatched policyengine.py releases. + +PyPI, the wheel, the data release manifest and the scratch install are +stubbed; the request is built from them exactly as in a scheduled run. +""" + +import hashlib +import io +import json +import zipfile +from pathlib import Path + +import pytest +from hypothesis import given, settings +from hypothesis import strategies as st + +from pipeline import uk_replay_discover as discover +from pipeline import uk_replay_qualify as qualify +from pipeline import uk_replay_variants as rv + +FIXTURES = Path(__file__).parent / "fixtures" / "uk_replay" +REQUESTED_BY = {"repository": "PolicyEngine/policyengine-scorecard", "commit": "0" * 40} + + +def golden(): + return json.loads((FIXTURES / "qualification_request_golden.json").read_text()) + + +def wheel_bytes(manifest: dict, requires: list[str]) -> bytes: + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, "w") as wheel: + wheel.writestr(discover.MANIFEST_PATH, json.dumps(manifest)) + wheel.writestr( + "policyengine-6.3.0.dist-info/METADATA", + "Metadata-Version: 2.1\nName: policyengine\nVersion: 6.3.0\n" + + "".join(f"Requires-Dist: {line}\n" for line in requires), + ) + return buffer.getvalue() + + +def bundle_manifest(model_sha="b" * 64, dataset_sha="d" * 64): + return { + "data_releases": { + "uk": { + "model_package": { + "name": "policyengine-uk", + "version": "2.126.8", + "sha256": model_sha, + }, + "certified_data_artifact": { + "build_id": "policyengine-uk-data-1.59.0", + "dataset": "enhanced_frs_2024_25", + "sha256": dataset_sha, + }, + "datasets": { + "enhanced_frs_2024_25": { + "repo_id": "policyengine/policyengine-uk-data-private", + "revision": "1.59.0", + "path": "enhanced_frs_2024_25.h5", + "sha256": dataset_sha, + } + }, + "data_package": { + "repo_id": "policyengine/policyengine-uk-data-private", + "repo_type": "model", + "release_manifest_path": "releases/1.59.0/release_manifest.json", + "release_manifest_revision": "a" * 40, + "version": "1.59.0", + }, + } + } + } + + +REQUIRES = [ + 'policyengine-uk==2.126.8; extra == "uk"', + 'policyengine-core==3.34.0; extra == "uk"', + 'policyengine-us==2.3.0; extra == "us"', +] + + +@pytest.fixture +def stubs(tmp_path, monkeypatch): + state = {"manifest": bundle_manifest(), "requires": list(REQUIRES), "release": None} + + def pypi(package, version=None): + if version is None: + return { + "releases": { + "6.2.5": [], + "6.2.6": [], + "6.3.0": [], + "6.3.1rc1": [], + "junk": [], + } + } + if package == "policyengine": + payload = wheel_bytes(state["manifest"], state["requires"]) + state["wheel"] = payload + digest = hashlib.sha256(payload).hexdigest() + else: + digest = {"policyengine-uk": "b" * 64, "policyengine-core": "c" * 64}[ + package + ] + return { + "urls": [ + { + "packagetype": "bdist_wheel", + "filename": f"{package}-{version}-py3-none-any.whl", + "url": f"https://files.example/{package}-{version}.whl", + "digests": {"sha256": digest}, + }, + { + "packagetype": "sdist", + "filename": "x.tar.gz", + "url": "u", + "digests": {"sha256": "s" * 64}, + }, + ] + } + + def fetch(url, destination, sha256): + payload = state["wheel"] + assert hashlib.sha256(payload).hexdigest() == sha256 + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_bytes(payload) + return destination + + def release_manifest(data_package, *, token): + release = state["release"] or { + "artifacts": { + "enhanced_frs_2024_25": {"sha256": "d" * 64, "size_bytes": 130_000_000} + } + } + return json.dumps(release).encode() + + monkeypatch.setattr(discover, "pypi", pypi) + monkeypatch.setattr(discover, "fetch", fetch) + monkeypatch.setattr(discover, "release_manifest", release_manifest) + return state + + +def test_a_release_becomes_a_valid_released_bundle_request(stubs, tmp_path): + request = discover.provisional_request( + "6.3.0", tmp_path, token=None, requested_by=REQUESTED_BY + ) + request["candidate"]["expected_release_bundle"] = { + **golden()["candidate"]["expected_release_bundle"], + "policyengine_version": "6.3.0", + "model_version": "2.126.8", + } + rid = qualify.request_id(request) + assert len(rid) == 64 + candidate = request["candidate"] + assert candidate["policyengine"]["source"] == { + "pypi": {"wheel_sha256": hashlib.sha256(stubs["wheel"]).hexdigest()} + } + manifest_bytes = zipfile.ZipFile(io.BytesIO(stubs["wheel"])).read( + discover.MANIFEST_PATH + ) + assert ( + candidate["policyengine"]["packaged_bundle_manifest_sha256"] + == hashlib.sha256(manifest_bytes).hexdigest() + ) + assert candidate["policyengine_core"] == { + "version": "3.34.0", + "wheel_sha256": "c" * 64, + } + assert candidate["data"]["size_bytes"] == 130_000_000 + assert request["kind"] == "released_bundle" + + +@pytest.mark.parametrize( + "change", + [ + lambda s: s.update(manifest=bundle_manifest(model_sha="x" * 64)), + lambda s: s.update( + release={ + "artifacts": { + "enhanced_frs_2024_25": {"sha256": "e" * 64, "size_bytes": 1} + } + } + ), + lambda s: s.update(release={"artifacts": {}}), + lambda s: s.update(requires=REQUIRES[1:]), + lambda s: s.update( + requires=['policyengine-uk==2.126.9; extra == "uk"', REQUIRES[1]] + ), + ], + ids=[ + "model-digest", + "dataset-digest", + "dataset-absent", + "missing-pin", + "pin-differs", + ], +) +def test_inconsistent_releases_are_skipped_with_a_reason(stubs, tmp_path, change): + change(stubs) + with pytest.raises(discover.Skipped): + discover.provisional_request( + "6.3.0", tmp_path, token=None, requested_by=REQUESTED_BY + ) + + +def test_candidates_are_newer_unrequested_final_releases(tmp_path): + document = rv.load_document() + request = golden() + request["kind"] = "released_bundle" + request["candidate"]["policyengine"]["version"] = "6.3.0" + directory = tmp_path / "requests" / ("a" * 64) + directory.mkdir(parents=True) + (directory / "REQUEST.json").write_text(json.dumps(request)) + versions = [ + "6.2.5", + "6.2.6", + "6.3.0", + "6.3.1", + "6.10.0", + "6.4.0rc1", + "junk", + "7.0.0", + ] + assert discover.candidates(versions, document, tmp_path) == [ + "6.3.1", + "6.10.0", + "7.0.0", + ] + + +def test_discovery_caps_each_run_and_continues_past_a_skip( + stubs, tmp_path, monkeypatch, capsys +): + monkeypatch.setattr(discover, "MAX_PER_RUN", 1) + monkeypatch.setattr( + discover, + "expected_bundle", + lambda request, work: { + **golden()["candidate"]["expected_release_bundle"], + "policyengine_version": request["candidate"]["policyengine"]["version"], + "model_version": "2.126.8", + }, + ) + found = discover.discover( + tmp_path / "state", tmp_path / "work", token=None, requested_by=REQUESTED_BY + ) + assert [r["candidate"]["policyengine"]["version"] for r in found] == ["6.3.0"] + assert ( + "building the oldest 1 now" not in capsys.readouterr().out + ) # one candidate only + stubs["manifest"] = bundle_manifest(model_sha="x" * 64) + assert ( + discover.discover( + tmp_path / "state", + tmp_path / "work2", + token=None, + requested_by=REQUESTED_BY, + ) + == [] + ) + assert "not requested" in capsys.readouterr().out + + +@settings(max_examples=100, deadline=None) +@given( + st.lists( + st.tuples(st.integers(0, 9), st.integers(0, 12), st.integers(0, 12)), + max_size=12, + ) +) +def test_candidates_are_sorted_strictly_above_the_floor(tmp_path_factory, triples): + versions = [f"{a}.{b}.{c}" for a, b, c in triples] + out = discover.candidates( + versions, rv.load_document(), tmp_path_factory.mktemp("s") + ) + from packaging.version import Version + + assert out == sorted(set(out), key=Version) + assert all(Version(v) > Version("6.2.6") for v in out) + assert set(out) == {v for v in versions if Version(v) > Version("6.2.6")}