From 4c7078345a7e61b873e58b538e3ef7fc19a63873 Mon Sep 17 00:00:00 2001 From: Chriopter Date: Wed, 7 Oct 2026 20:48:27 +0200 Subject: [PATCH 1/6] Added opt-in GitHub release scanning --- .github/workflows/publish.yml | 9 +- .github/workflows/scan-releases.yml | 82 ++++++ README.md | 37 +++ add-resources-and-info.py | 2 +- data/pspdx.json | 3 +- homebrew_database/homebrew.py | 7 +- scan-releases.py | 342 ++++++++++++++++++++++++ tests/test_scan_releases.py | 396 ++++++++++++++++++++++++++++ 8 files changed, 873 insertions(+), 5 deletions(-) create mode 100644 .github/workflows/scan-releases.yml create mode 100755 scan-releases.py create mode 100644 tests/test_scan_releases.py diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index a0a4ab0..f31a5e1 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -4,6 +4,11 @@ on: push: branches: [main] pull_request: + workflow_dispatch: + +concurrency: + group: website-${{ github.ref }} + cancel-in-progress: false jobs: build: @@ -27,7 +32,7 @@ jobs: publish: runs-on: ubuntu-latest needs: [build] - if: github.event_name == 'push' && github.ref_name == 'main' + if: github.ref_name == 'main' && (github.event_name == 'push' || github.event_name == 'workflow_dispatch') permissions: contents: read pages: write @@ -39,4 +44,4 @@ jobs: steps: - name: Deploy to GitHub Pages id: deployment - uses: actions/deploy-pages@v5 \ No newline at end of file + uses: actions/deploy-pages@v5 diff --git a/.github/workflows/scan-releases.yml b/.github/workflows/scan-releases.yml new file mode 100644 index 0000000..0f51ff4 --- /dev/null +++ b/.github/workflows/scan-releases.yml @@ -0,0 +1,82 @@ +name: Scan releases + +on: + schedule: + - cron: '37 * * * *' + workflow_dispatch: + pull_request: + paths: + - 'scan-releases.py' + - 'tests/**' + - 'homebrew_database/**' + - '.github/workflows/scan-releases.yml' + +permissions: + contents: read + +concurrency: + group: release-scanner-${{ github.ref }} + cancel-in-progress: false + +jobs: + tests: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v6 + - uses: actions/setup-python@v6 + with: + python-version: '3.12' + - run: python -m unittest discover -s tests -v + + scan: + needs: tests + if: github.repository == 'pspdev/homebrew' && github.ref_name == 'main' && github.event_name != 'pull_request' + runs-on: ubuntu-latest + timeout-minutes: 45 + permissions: + contents: write + actions: write + steps: + - uses: actions/checkout@v6 + with: + ref: main + - uses: actions/setup-python@v6 + with: + python-version: '3.12' + - name: Scan approved entries + id: scan + continue-on-error: true + env: + GITHUB_TOKEN: ${{ github.token }} + run: python scan-releases.py + - name: Check for changes + id: changes + run: | + git add -- data resources/icons + if ! git diff --cached --quiet; then + echo 'changed=true' >> "$GITHUB_OUTPUT" + fi + - name: Validate with the existing website build + if: steps.changes.outputs.changed == 'true' + run: | + python -m pip install -r requirements.txt + python build.py + - name: Commit and push validated updates + if: steps.changes.outputs.changed == 'true' + run: | + git config user.name 'github-actions[bot]' + git config user.email '41898282+github-actions[bot]@users.noreply.github.com' + git commit -m 'Added releases from approved repositories' + # A concurrent change to main rejects this push. Never force or + # rebase generated data over a curator's edits; the next run retries. + git push origin HEAD:main + - name: Publish updated database + if: steps.changes.outputs.changed == 'true' + env: + GH_TOKEN: ${{ github.token }} + run: gh workflow run publish.yml --ref main + - name: Report scan failures + if: always() && steps.scan.outcome == 'failure' + run: | + echo 'Some entries could not be scanned; see the scan log. Their existing data was kept.' + exit 1 diff --git a/README.md b/README.md index be6bd8f..adcbb05 100644 --- a/README.md +++ b/README.md @@ -91,6 +91,43 @@ Here is a description for each field in a release: After creating the file, run the `add-resources-and-info.py` script once. This will add the icon and information like the md5 checksum of the `EBOOT.PBP` and the sha256 checksum and sie of the archive for each release. If this wasn't done, the build will fail. +## Release updates + +- **Opt-in:** `"scan_releases": true` in [`data/`](data/); `source` names the GitHub repo. +- **Hourly:** add stable releases, checksums and the latest icon. +- **Metadata:** unchanged; drafts and prereleases are skipped. +- **ZIP:** one ZIP, or exactly one with `psp` in its name. + Missing or ambiguous ZIPs are skipped. + +```text +hourly scan -> changes? -> build check -> commit + push -> website build -> deploy + no -> done +``` + +
+Scanner · checks, workflow, failures + +- **Checks:** download size, SHA-256 when GitHub supplies it, ZIP paths and CRCs, + one EBOOT with a valid PBP header. Records ZIP SHA-256 and EBOOT MD5. +- **History:** keeps existing releases; `v1.0` and `1.0` count as one. +- **Skipped:** missing or ambiguous ZIPs and invalid packages; reason in the log. +- **Errors:** network or download verification fails → entry unchanged; run fails. +- **Build or push fails:** no update to `main`; no force-push. +- **Publish fails:** rerun **CI** on `main`. + +```sh +python scan-releases.py --dry-run # check +python scan-releases.py # update +python -m unittest discover -s tests -v +``` + +[`scan-releases.py`](scan-releases.py) · [hourly workflow](.github/workflows/scan-releases.yml) + +Requires push access to `main`. `GITHUB_TOKEN` raises the API rate limit. +Known ZIPs are not downloaded again; differing GitHub digests are reported. + +
+ ## Using the homebrew database data in applications The PSP Homebrew Database supports a couple of formats for applications like homebrew stores to use. In every single format, the homebrew list is returned by last updated with the newest at the top. Below they will be listed with a description of how to use each. diff --git a/add-resources-and-info.py b/add-resources-and-info.py index 62d772b..1b3b581 100755 --- a/add-resources-and-info.py +++ b/add-resources-and-info.py @@ -101,7 +101,7 @@ def main(): logging.info(f"Writing {homebrew_json_path}") with open(homebrew_json_path, "w") as fd: - fd.write(json.dumps(homebrew.to_dict(), indent=2)) + fd.write(json.dumps(homebrew.to_dict(include_ci=True), indent=2)) except Exception: logging.error("Could not read %s", homebrew_json_path, exc_info=True) logging.info("Done, make sure to add and commit changes to the data and resources directory to git") diff --git a/data/pspdx.json b/data/pspdx.json index d179041..b455cdb 100644 --- a/data/pspdx.json +++ b/data/pspdx.json @@ -27,5 +27,6 @@ }, "website": "https://chriopter.github.io/pspdx-catalog/", "source": "https://github.com/chriopter/pspdx-app", + "scan_releases": true, "license": "GPLv2.0" -} \ No newline at end of file +} diff --git a/homebrew_database/homebrew.py b/homebrew_database/homebrew.py index 76edcd4..de3d444 100644 --- a/homebrew_database/homebrew.py +++ b/homebrew_database/homebrew.py @@ -59,6 +59,8 @@ class Homebrew: tags: list[str] = field(default_factory=list) languages: list[str] = field(default_factory=list) icon: str = None + # CI controls must survive add-resources-and-info.py's JSON round trip. + scan_releases: bool | None = None def __gt__(self, other: 'Homebrew') -> bool: self_last_release = self.get_last_release() @@ -74,7 +76,7 @@ def get_last_release(self) -> Release | None: return None return sorted(self.releases, reverse=True)[0] - def to_dict(self) -> dict: + def to_dict(self, *, include_ci: bool = False) -> dict: return_dict = { "name": self.name, "id": self.id, @@ -102,6 +104,8 @@ def to_dict(self) -> dict: return_dict["media"]["icon"] = self.icon if len(self.languages) > 0: return_dict["languages"] = self.languages + if include_ci and self.scan_releases is not None: + return_dict["scan_releases"] = self.scan_releases self.releases.sort(reverse=True) for release in self.releases: @@ -146,6 +150,7 @@ def get_homebrew_from_json_data(id: str, data: dict) -> Homebrew: tags=data.get("tags", list()), icon=data["media"].get("icon", None), languages=data.get("languages", list()), + scan_releases=data.get("scan_releases", None), ) except (KeyError, ValueError) as e: raise HomebrewProcessingException(f"Could not process json data for {id}") from e diff --git a/scan-releases.py b/scan-releases.py new file mode 100755 index 0000000..d6ca382 --- /dev/null +++ b/scan-releases.py @@ -0,0 +1,342 @@ +#!/usr/bin/env python3 +"""Refresh opted-in database entries from GitHub releases, without .pspdx. + +Only releases and, when a newer package supplies one, the icon are changed. +Each entry is prepared completely before writing; failures leave it intact. +""" + +import argparse +import copy +import datetime +import hashlib +import http.client +import io +import json +import logging +import os +import re +import stat +import tempfile +import urllib.parse +import urllib.request +import zipfile +import zlib +from pathlib import Path + +ROOT = Path(__file__).resolve().parent +LOGGER = logging.getLogger(__name__) +MAX_DOWNLOAD = 256 * 1024 * 1024 +MAX_API_RESPONSE = 8 * 1024 * 1024 +MAX_UNPACKED = 512 * 1024 * 1024 +MAX_EBOOT = 64 * 1024 * 1024 +MAX_ICON = 2 * 1024 * 1024 + + +class ScanError(ValueError): + pass + + +class PackageError(ScanError): + pass + + +class Redirect(urllib.request.HTTPRedirectHandler): + def redirect_request(self, req, fp, code, msg, headers, newurl): + if urllib.parse.urlsplit(newurl).scheme != "https": + raise ScanError("redirect outside HTTPS") + redirected = super().redirect_request(req, fp, code, msg, headers, newurl) + if redirected is not None: + redirected.remove_header("Authorization") + return redirected + + +def fetch(url): + headers = {"User-Agent": "pspdev-homebrew-release-scanner"} + limit = MAX_DOWNLOAD + if urllib.parse.urlsplit(url).hostname == "api.github.com": + headers["Accept"] = "application/vnd.github+json" + headers["X-GitHub-Api-Version"] = "2022-11-28" + token = os.environ.get("GITHUB_TOKEN") + if token: + headers["Authorization"] = f"Bearer {token}" + limit = MAX_API_RESPONSE + request = urllib.request.Request(url, headers=headers) + with urllib.request.build_opener(Redirect()).open(request, timeout=60) as response: + content = response.read(limit + 1) + if len(content) > limit: + raise ScanError("response too large") + return content + + +def repository(source): + if not isinstance(source, str): + raise ScanError("source must name a GitHub repository") + match = re.fullmatch( + r"https://github\.com/([A-Za-z0-9_.-]{1,39})/" + r"([A-Za-z0-9_.-]{1,100})/?", + source, + ) + if not match: + raise ScanError("source must name a GitHub repository") + owner, repo = match.groups() + repo = repo.removesuffix(".git") + if not any(c.isalnum() for c in owner) or not any(c.isalnum() for c in repo): + raise ScanError("invalid repository name") + return f"{owner}/{repo}" + + +def fetch_releases(repo, fetcher): + result = [] + for page in range(1, 101): + batch = json.loads( + fetcher( + f"https://api.github.com/repos/{repo}/releases?per_page=100&page={page}" + ) + ) + if not isinstance(batch, list): + raise ScanError("GitHub did not return a release list") + result.extend(r for r in batch if not r["draft"] and not r["prerelease"]) + if len(batch) < 100: + return sorted(result, key=lambda r: r["published_at"], reverse=True) + raise ScanError("too many release pages; refusing incomplete history") + + +def tag_key(tag): + # Curated entries sometimes spell GitHub's v0.8.5 as 0.8.5. + return tag[1:] if tag.startswith("v") and len(tag) > 1 and tag[1].isdigit() else tag + + +def select_asset(release): + candidates = [a for a in release["assets"] if a["name"].lower().endswith(".zip")] + if len(candidates) > 1: + candidates = [a for a in candidates if "psp" in a["name"].lower()] + return candidates[0] if len(candidates) == 1 else None + + +def inspect_package(archive): + # Never extract files or execute code from the downloaded package. + try: + with zipfile.ZipFile(io.BytesIO(archive)) as zipped: + entries = zipped.infolist() + if len(entries) > 10000 or sum(i.file_size for i in entries) > MAX_UNPACKED: + raise PackageError("ZIP exceeds unpacked limits") + names, eboots = set(), [] + for entry in entries: + name = entry.filename.replace("\\", "/") + parts = name.rstrip("/").split("/") + if ( + name.startswith("/") + or any(p in {"", ".", ".."} or ":" in p for p in parts) + or any(ord(c) < 32 for c in entry.orig_filename) + ): + raise PackageError("unsafe ZIP path") + mode = entry.external_attr >> 16 + if ( + stat.S_IFMT(mode) not in {0, stat.S_IFREG, stat.S_IFDIR} + or entry.flag_bits & 1 + ): + raise PackageError("special file or encrypted ZIP entry") + key = name.rstrip("/").casefold() + if key in names: + raise PackageError("duplicate ZIP path") + names.add(key) + if not entry.is_dir() and parts[-1].lower() == "eboot.pbp": + eboots.append(entry) + # Deliberately stricter than the client: ambiguous layouts need a + # curator rather than an unattended guess. + if len(eboots) != 1: + raise PackageError("ZIP must contain exactly one EBOOT.PBP") + if eboots[0].file_size > MAX_EBOOT: + raise PackageError("EBOOT.PBP too large") + # Check every payload, not just the PBP: a corrupt resource file + # must not become an automatically approved package. + bad = zipped.testzip() + if bad is not None: + raise PackageError(f"ZIP CRC mismatch: {bad}") + eboot = zipped.read(eboots[0]) # zipfile checks this entry's CRC. + except (zipfile.BadZipFile, RuntimeError, EOFError, zlib.error) as error: + raise PackageError(f"invalid ZIP: {error}") from error + if len(eboot) < 40 or eboot[:4] != b"\x00PBP": + raise PackageError("invalid EBOOT.PBP header") + offsets = [int.from_bytes(eboot[at : at + 4], "little") for at in range(8, 40, 4)] + if offsets[0] < 40 or offsets != sorted(offsets) or offsets[-1] > len(eboot): + raise PackageError("invalid EBOOT.PBP offsets") + icon = eboot[offsets[1] : offsets[2]] + if len(icon) > MAX_ICON or (icon and not icon.startswith(b"\x89PNG\r\n\x1a\n")): + raise PackageError("invalid ICON0.PNG") + return hashlib.md5(eboot, usedforsecurity=False).hexdigest(), icon + + +def release_record(repo, release, asset, fetcher): + url = asset["browser_download_url"] + if not url.startswith(f"https://github.com/{repo}/releases/download/"): + raise ScanError("asset does not belong to the approved repository") + if type(asset["size"]) is not int or not 0 < asset["size"] <= MAX_DOWNLOAD: + raise ScanError("invalid asset size") + archive = fetcher(url) + if len(archive) != asset["size"]: + raise ScanError("asset size does not match GitHub") + sha = hashlib.sha256(archive).hexdigest() + if asset.get("digest") and asset["digest"] != "sha256:" + sha: + raise ScanError("asset SHA-256 does not match GitHub") + eboot_md5, icon = inspect_package(archive) + published = datetime.datetime.fromisoformat( + release["published_at"].replace("Z", "+00:00") + ) + if published.tzinfo is None: + raise ScanError("release timestamp has no timezone") + date = published.astimezone(datetime.timezone.utc).date().isoformat() + record = { + "tag": release["tag_name"], + "url": url, + "published_at": date, + "size": len(archive), + "sha256": sha, + "eboot_md5": eboot_md5, + } + if release.get("body"): + record["changelog"] = release["body"] + return record, icon + + +def check_known_asset(repo, release, existing): + # A replacement needs manual review, but must not block newer releases. + for asset in release["assets"]: + if asset["browser_download_url"] != existing["url"]: + continue + digest = asset.get("digest") + if ( + digest + and existing.get("sha256") + and digest != "sha256:" + existing["sha256"] + ): + LOGGER.warning( + "%s %s: known release asset changed; kept existing record", + repo, + release["tag_name"], + ) + + +def prepare(data, fetcher): + repo = repository(data.get("source")) + existing = data["releases"] + if not isinstance(existing, list): + raise ScanError("releases must be a list") + known = {tag_key(r["tag"]): r for r in existing} + # Optional: read the repo's .pspdx here for release selection or metadata updates. + remote = fetch_releases(repo, fetcher) + added, icon = [], None + latest_date = max((r["published_at"] for r in existing), default="") + newest_seen = False + for release in remote: + key = tag_key(release["tag_name"]) + if key in known: + newest_seen = True + check_known_asset(repo, release, known[key]) + continue + asset = select_asset(release) + if asset is None: + LOGGER.warning( + "%s %s: no unique ZIP asset, skipped", repo, release["tag_name"] + ) + continue + try: + record, new_icon = release_record(repo, release, asset, fetcher) + except PackageError as error: + LOGGER.warning("%s %s: %s, skipped", repo, release["tag_name"], error) + continue + if not newest_seen and record["published_at"] >= latest_date: + icon = new_icon or None + newest_seen = True + added.append(record) + known[key] = record + updated = copy.deepcopy(data) + if added: + # Dates in this database have no time of day. Use GitHub's full + # timestamp order to keep same-day backfills behind newer releases. + order = {tag_key(r["tag_name"]): index for index, r in enumerate(remote)} + updated["releases"] = sorted( + added + existing, + key=lambda r: ( + r["published_at"], + -order.get(tag_key(r["tag"]), len(remote)), + ), + reverse=True, + ) + return updated, icon + + +def atomic_write(path, content): + path.parent.mkdir(parents=True, exist_ok=True) + name = None + try: + with tempfile.NamedTemporaryFile(dir=path.parent, delete=False) as temporary: + name = Path(temporary.name) + temporary.write(content) + os.replace(name, path) + finally: + if name is not None: + name.unlink(missing_ok=True) + + +def scan(root, fetcher=fetch, dry_run=False): + changed, errors = 0, [] + for path in sorted((root / "data").rglob("*.json")): + try: + data = json.loads(path.read_text(encoding="utf-8")) + flag = data.get("scan_releases", False) + if type(flag) is not bool: + raise ScanError("scan_releases must be a boolean") + if not flag: + continue + updated, icon = prepare(data, fetcher) + if updated == data: + continue + icon_path = None + if icon: + name = f"icons/{path.stem}.png" + updated["media"]["icon"] = name + icon_path = root / "resources" / name + content = json.dumps(updated, indent=2).encode("utf-8") + if not dry_run: + if icon_path: + atomic_write(icon_path, icon) + atomic_write(path, content) + changed += 1 + LOGGER.info( + "%s: %d releases added", + path.name, + len(updated["releases"]) - len(data["releases"]), + ) + except ( + ValueError, + KeyError, + TypeError, + AttributeError, + OSError, + http.client.HTTPException, + ) as error: + errors.append(f"{path.name}: {error}") + LOGGER.error("%s", errors[-1]) + return changed, errors + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--dry-run", action="store_true", help="download and check without writing" + ) + args = parser.parse_args() + logging.basicConfig(level=logging.INFO, format="%(message)s") + changed, errors = scan(ROOT, dry_run=args.dry_run) + LOGGER.info( + "%d entries %s, %d errors", + changed, + "would change" if args.dry_run else "changed", + len(errors), + ) + return 1 if errors else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_scan_releases.py b/tests/test_scan_releases.py new file mode 100644 index 0000000..434d78b --- /dev/null +++ b/tests/test_scan_releases.py @@ -0,0 +1,396 @@ +import copy +import hashlib +import http.client +import importlib.util +import io +import json +import tempfile +import unittest +import urllib.request +import zipfile +from pathlib import Path +from unittest import mock + +from homebrew_database.homebrew import get_homebrew_from_json_data + +ROOT = Path(__file__).resolve().parents[1] +spec = importlib.util.spec_from_file_location("scanner", ROOT / "scan-releases.py") +scanner = importlib.util.module_from_spec(spec) +spec.loader.exec_module(scanner) + + +def archive(entries=None, icon=b""): + offsets = [40, 40] + [40 + len(icon)] * 6 + eboot = ( + b"\x00PBP" + + b"\x00\x00\x01\x00" + + b"".join(offset.to_bytes(4, "little") for offset in offsets) + + icon + + b"payload" + ) + out = io.BytesIO() + with zipfile.ZipFile(out, "w") as zipped: + for name, data in (entries or {"PSP/GAME/App/EBOOT.PBP": eboot}).items(): + zipped.writestr(name, data) + return out.getvalue() + + +def release(tag="v2.0", date="2026-10-01T12:00:00Z", name="psp.zip", content=None): + content = archive() if content is None else content + return { + "tag_name": tag, + "published_at": date, + "draft": False, + "prerelease": False, + "body": "Release notes", + "assets": [ + { + "name": name, + "size": len(content), + "browser_download_url": f"https://github.com/test/app/releases/download/{tag}/{name}", + "digest": "sha256:" + hashlib.sha256(content).hexdigest(), + } + ], + } + + +def entry(): + return { + "name": "App", + "summary": "Curated summary", + "author": "Author", + "ai_used": False, + "requires_additional_files": False, + "category": "application", + "tags": ["tools"], + "source": "https://github.com/test/app", + "scan_releases": True, + "media": {"screenshots": ["custom.png"], "icon": "icons/app.png"}, + "releases": [ + { + "tag": "1.0", + "published_at": "2026-09-01", + "size": 1, + "url": "https://github.com/test/app/releases/download/v1.0/psp.zip", + "sha256": "a" * 64, + "eboot_md5": "b" * 32, + } + ], + } + + +def network(items, content=None): + content = archive() if content is None else content + + def get(url): + if url.startswith("https://api.github.com/"): + return json.dumps(items).encode() + return content + + return get + + +class ScannerTests(unittest.TestCase): + def test_invalid_historical_package_does_not_block_new_releases(self): + invalid = b"old invalid package" + older = release("v0.9", date="2026-08-01T12:00:00Z", content=invalid) + good = release() + + def fetcher(url): + if "api.github.com" in url: + return json.dumps([good, older]).encode() + return invalid if "/v0.9/" in url else archive() + + with self.assertLogs(scanner.LOGGER, level="WARNING"): + updated, _ = scanner.prepare(entry(), fetcher) + self.assertEqual([r["tag"] for r in updated["releases"]], ["v2.0", "1.0"]) + + def test_newest_icon_uses_existing_filename_and_unicode_ids(self): + png = (ROOT / "resources" / "icons" / "pspdx.png").read_bytes() + content = archive(icon=png) + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "data").mkdir() + path = root / "data" / "app_日本語.json" + data = entry() + data["name"] = "日本語" + path.write_text(json.dumps(data)) + icon = root / "resources" / "icons" / "app_日本語.png" + icon.parent.mkdir(parents=True) + icon.write_bytes(b"previous icon") + newest = release(content=content) + self.assertEqual(scanner.scan(root, network([newest], content)), (1, [])) + updated = json.loads(path.read_text()) + self.assertEqual(updated["media"]["icon"], "icons/app_日本語.png") + self.assertEqual(icon.read_bytes(), png) + self.assertEqual(list(icon.parent.iterdir()), [icon]) + self.assertEqual(path.read_text(), json.dumps(updated, indent=2)) + older = release("v0.9", date="2026-08-01T12:00:00Z", content=content) + newest_bytes = path.read_bytes() + self.assertEqual(scanner.scan(root, network([newest], content)), (0, [])) + self.assertEqual(path.read_bytes(), newest_bytes) + self.assertEqual( + scanner.scan(root, network([newest, older], content)), (1, []) + ) + self.assertEqual(icon.read_bytes(), png) + + def test_backfill_does_not_replace_existing_icon(self): + png = (ROOT / "resources" / "icons" / "pspdx.png").read_bytes() + content = archive(icon=png) + older = release("v0.9", date="2026-08-01T12:00:00Z", content=content) + updated, icon = scanner.prepare(entry(), network([older], content)) + self.assertIsNone(icon) + self.assertEqual(updated["media"], entry()["media"]) + + def test_public_metadata_omits_ci_settings(self): + model = get_homebrew_from_json_data("app", entry()) + self.assertNotIn("scan_releases", model.to_dict()) + self.assertIs(model.to_dict(include_ci=True)["scan_releases"], True) + + def test_ambiguous_release_does_not_block_a_usable_release(self): + ambiguous = release("v3.0") + ambiguous["assets"].append(copy.deepcopy(ambiguous["assets"][0])) + updated, _ = scanner.prepare(entry(), network([ambiguous, release()])) + self.assertEqual([r["tag"] for r in updated["releases"]], ["v2.0", "1.0"]) + + def test_invalid_package_is_skipped_without_changing_the_entry(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "data").mkdir() + path = root / "data" / "app.json" + path.write_text(json.dumps(entry())) + before = path.read_bytes() + corrupt = b"not a zip" + changed, errors = scanner.scan( + root, network([release(content=corrupt)], corrupt) + ) + self.assertEqual((changed, errors), (0, [])) + self.assertEqual(path.read_bytes(), before) + + def test_failed_atomic_replace_keeps_original_and_removes_temporary_file(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + path = root / "entry.json" + path.write_bytes(b"original") + with ( + mock.patch.object( + scanner.os, "replace", side_effect=OSError("read only") + ), + self.assertRaises(OSError), + ): + scanner.atomic_write(path, b"updated") + self.assertEqual(path.read_bytes(), b"original") + self.assertEqual(list(root.iterdir()), [path]) + + def test_invalid_opt_in_is_reported_without_requests(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "data").mkdir() + data = entry() + data["scan_releases"] = "true" + (root / "data" / "app.json").write_text(json.dumps(data)) + changed, errors = scanner.scan(root, lambda url: self.fail(url)) + self.assertEqual((changed, len(errors)), (0, 1)) + + def test_new_releases_preserve_curated_fields_and_history(self): + original = entry() + updated, icon = scanner.prepare(original, network([release()])) + self.assertEqual(original, entry()) + for key in original.keys() - {"releases"}: + self.assertEqual(updated[key], original[key]) + self.assertEqual(updated["releases"][1], original["releases"][0]) + self.assertEqual( + updated["releases"][0]["sha256"], hashlib.sha256(archive()).hexdigest() + ) + self.assertEqual(updated["releases"][0]["changelog"], "Release notes") + self.assertIsNone(icon) + + def test_known_tag_with_leading_v_is_not_downloaded_or_duplicated(self): + known = release("v1.0") + known["assets"][0]["digest"] = "sha256:" + "a" * 64 + + def get(url): + self.assertIn("api.github.com", url) + return json.dumps([known]).encode() + + updated, _ = scanner.prepare(entry(), get) + self.assertEqual(updated, entry()) + + def test_known_asset_replacement_is_reported_without_blocking_updates(self): + with self.assertLogs(scanner.LOGGER, level="WARNING") as logs: + updated, _ = scanner.prepare(entry(), network([release("v1.0"), release()])) + self.assertIn("known release asset changed", logs.output[0]) + self.assertEqual([r["tag"] for r in updated["releases"]], ["v2.0", "1.0"]) + self.assertEqual(updated["releases"][1], entry()["releases"][0]) + + def test_drafts_and_prereleases_are_ignored(self): + draft, preview = release(), release("v3.0") + draft["draft"], preview["prerelease"] = True, True + self.assertEqual( + scanner.prepare(entry(), network([draft, preview]))[0], entry() + ) + + def test_pagination_reads_history_and_sorts_by_publication(self): + first = [release(f"v{i}") for i in range(100)] + last = release("v100", date="2026-10-02T12:00:00Z") + + def get(url): + return json.dumps(first if url.endswith("page=1") else [last]).encode() + + fetched = scanner.fetch_releases("test/app", get) + self.assertEqual(len(fetched), 101) + self.assertEqual(fetched[0]["tag_name"], "v100") + + def test_same_day_backfill_stays_behind_the_known_newest_release(self): + data = entry() + data["releases"][0]["published_at"] = "2026-10-01" + newest = release("v1.0", date="2026-10-01T18:00:00Z") + newest["assets"][0]["digest"] = "sha256:" + "a" * 64 + older = release("v0.9", date="2026-10-01T08:00:00Z") + updated, _ = scanner.prepare(data, network([older, newest])) + self.assertEqual([r["tag"] for r in updated["releases"]], ["1.0", "v0.9"]) + + def test_truncated_download_is_isolated_to_its_entry(self): + + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "data").mkdir() + bad = entry() + bad["source"] = "https://github.com/test/bad" + path = root / "data" / "bad.json" + path.write_text(json.dumps(bad)) + before = path.read_bytes() + (root / "data" / "good.json").write_text(json.dumps(entry())) + + def get(url): + if "/test/bad/" in url: + raise http.client.IncompleteRead(b"partial", 100) + return network([release()])(url) + + changed, errors = scanner.scan(root, get) + self.assertEqual((changed, len(errors)), (1, 1)) + self.assertEqual(path.read_bytes(), before) + + def test_zip_selection_requires_a_unique_match(self): + r = release() + r["assets"] += [copy.deepcopy(r["assets"][0])] + self.assertIsNone(scanner.select_asset(r)) + r["assets"][1]["name"] = "linux.zip" + self.assertEqual(scanner.select_asset(r)["name"], "psp.zip") + + def test_wrong_size_digest_and_download_host_are_rejected(self): + for field, value in [ + ("size", 1), + ("digest", "sha256:" + "0" * 64), + ("browser_download_url", "https://example.org/app.zip"), + ]: + with self.subTest(field=field): + r = release() + r["assets"][0][field] = value + with self.assertRaises(scanner.ScanError): + scanner.prepare(entry(), network([r])) + + def test_unsafe_ambiguous_and_invalid_packages_are_rejected(self): + for entries in [ + {"../EBOOT.PBP": b"x"}, + {"EBOOT.PBP": b"x"}, + {"a/EBOOT.PBP": b"x", "b/EBOOT.PBP": b"x"}, + {"EBOOT.PBP": b"x", "eboot.pbp": b"x"}, + ]: + with self.subTest(entries=entries), self.assertRaises(scanner.ScanError): + scanner.inspect_package(archive(entries)) + + def test_corrupt_non_eboot_resource_is_rejected(self): + eboot = b"\x00PBP" + b"\x00\x00\x01\x00" + (40).to_bytes(4, "little") * 8 + content = archive( + {"EBOOT.PBP": eboot, "resource.dat": b"unique resource bytes"} + ) + damaged = content.replace(b"unique resource bytes", b"broken resource bytes", 1) + with self.assertRaisesRegex(scanner.ScanError, "CRC"): + scanner.inspect_package(damaged) + + def test_release_without_zip_does_not_block_other_releases(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "data").mkdir() + path = root / "data" / "app.json" + path.write_text(json.dumps(entry())) + older = release("v1.5", date="2026-09-15T12:00:00Z") + older["assets"] = [] + changed, errors = scanner.scan(root, network([release(), older])) + self.assertEqual((changed, errors), (1, [])) + self.assertEqual( + [r["tag"] for r in json.loads(path.read_text())["releases"]], + ["v2.0", "1.0"], + ) + + def test_redirects_drop_authentication_and_refuse_http(self): + + req = urllib.request.Request( + "https://api.github.com/start", headers={"Authorization": "Bearer secret"} + ) + redirect = scanner.Redirect() + redirected = redirect.redirect_request( + req, None, 302, "Found", {}, "https://other.example/end" + ) + self.assertFalse(redirected.has_header("Authorization")) + with self.assertRaises(scanner.ScanError): + redirect.redirect_request( + req, None, 302, "Found", {}, "http://other.example/end" + ) + + def test_only_opted_in_entries_are_requested(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "data").mkdir() + for flag in [None, False]: + data = entry() + if flag is None: + del data["scan_releases"] + else: + data["scan_releases"] = flag + (root / "data" / "app.json").write_text(json.dumps(data)) + self.assertEqual( + scanner.scan(root, lambda url: self.fail(url)), (0, []) + ) + + def test_failed_entry_is_unchanged_while_other_entry_updates(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "data").mkdir() + bad, good = root / "data" / "bad.json", root / "data" / "good.json" + broken = entry() + broken["source"] = "https://example.org/" + bad.write_text(json.dumps(broken)) + before = bad.read_bytes() + good.write_text(json.dumps(entry())) + changed, errors = scanner.scan(root, network([release()])) + self.assertEqual(changed, 1) + self.assertEqual(len(errors), 1) + self.assertEqual(bad.read_bytes(), before) + self.assertEqual(json.loads(good.read_text())["releases"][0]["tag"], "v2.0") + + def test_dry_run_and_second_scan_do_not_write(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "data").mkdir() + path = root / "data" / "app.json" + path.write_text(json.dumps(entry())) + before = path.read_bytes() + self.assertEqual(scanner.scan(root, network([release()]), True), (1, [])) + self.assertEqual(path.read_bytes(), before) + self.assertEqual(scanner.scan(root, network([release()])), (1, [])) + after = path.read_bytes() + self.assertEqual(scanner.scan(root, network([release()])), (0, [])) + self.assertEqual(path.read_bytes(), after) + + def test_ci_controls_survive_existing_enrichment_script_round_trip(self): + for flag in [True, False]: + data = entry() + data["scan_releases"] = flag + updated = get_homebrew_from_json_data("app", data).to_dict(include_ci=True) + self.assertIs(updated["scan_releases"], flag) + + +if __name__ == "__main__": + unittest.main() From 184f79d97467f050a6a37ca01a14bc2cc590b26b Mon Sep 17 00:00:00 2001 From: chriopter <82179548+chriopter@users.noreply.github.com> Date: Thu, 8 Oct 2026 10:06:17 +0200 Subject: [PATCH 2/6] Update .github/workflows/scan-releases.yml Co-authored-by: Wouter Wijsman --- .github/workflows/scan-releases.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/scan-releases.yml b/.github/workflows/scan-releases.yml index 0f51ff4..94fbade 100644 --- a/.github/workflows/scan-releases.yml +++ b/.github/workflows/scan-releases.yml @@ -22,7 +22,7 @@ jobs: tests: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v6 + - uses: actions/checkout@v7 - uses: actions/setup-python@v6 with: python-version: '3.12' From ef008f5ab5a1e616c2c5f1af40d36f6eb2bc5399 Mon Sep 17 00:00:00 2001 From: chriopter <82179548+chriopter@users.noreply.github.com> Date: Thu, 8 Oct 2026 10:06:23 +0200 Subject: [PATCH 3/6] Update .github/workflows/scan-releases.yml Co-authored-by: Wouter Wijsman --- .github/workflows/scan-releases.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/scan-releases.yml b/.github/workflows/scan-releases.yml index 94fbade..dd365e7 100644 --- a/.github/workflows/scan-releases.yml +++ b/.github/workflows/scan-releases.yml @@ -37,7 +37,7 @@ jobs: contents: write actions: write steps: - - uses: actions/checkout@v6 + - uses: actions/checkout@v7 with: ref: main - uses: actions/setup-python@v6 From 218127147bd883991402bc6e8b9b3335018e20c3 Mon Sep 17 00:00:00 2001 From: chriopter <82179548+chriopter@users.noreply.github.com> Date: Thu, 8 Oct 2026 10:09:10 +0200 Subject: [PATCH 4/6] Run release scanner daily instead of hourly --- .github/workflows/scan-releases.yml | 2 +- README.md | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/.github/workflows/scan-releases.yml b/.github/workflows/scan-releases.yml index dd365e7..6376f9d 100644 --- a/.github/workflows/scan-releases.yml +++ b/.github/workflows/scan-releases.yml @@ -2,7 +2,7 @@ name: Scan releases on: schedule: - - cron: '37 * * * *' + - cron: '37 3 * * *' workflow_dispatch: pull_request: paths: diff --git a/README.md b/README.md index adcbb05..83fdd5e 100644 --- a/README.md +++ b/README.md @@ -94,13 +94,13 @@ After creating the file, run the `add-resources-and-info.py` script once. This w ## Release updates - **Opt-in:** `"scan_releases": true` in [`data/`](data/); `source` names the GitHub repo. -- **Hourly:** add stable releases, checksums and the latest icon. +- **Daily:** add stable releases, checksums and the latest icon. - **Metadata:** unchanged; drafts and prereleases are skipped. - **ZIP:** one ZIP, or exactly one with `psp` in its name. Missing or ambiguous ZIPs are skipped. ```text -hourly scan -> changes? -> build check -> commit + push -> website build -> deploy +daily scan -> changes? -> build check -> commit + push -> website build -> deploy no -> done ``` @@ -121,7 +121,7 @@ python scan-releases.py # update python -m unittest discover -s tests -v ``` -[`scan-releases.py`](scan-releases.py) · [hourly workflow](.github/workflows/scan-releases.yml) +[`scan-releases.py`](scan-releases.py) · [daily workflow](.github/workflows/scan-releases.yml) Requires push access to `main`. `GITHUB_TOKEN` raises the API rate limit. Known ZIPs are not downloaded again; differing GitHub digests are reported. From 8b03a23ef8a335d7bcc5bccc3066e3de2c15a512 Mon Sep 17 00:00:00 2001 From: chriopter <82179548+chriopter@users.noreply.github.com> Date: Thu, 8 Oct 2026 10:15:22 +0200 Subject: [PATCH 5/6] Update scan-releases.py Co-authored-by: Wouter Wijsman --- scan-releases.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scan-releases.py b/scan-releases.py index d6ca382..bbb82e6 100755 --- a/scan-releases.py +++ b/scan-releases.py @@ -328,7 +328,7 @@ def main(): ) args = parser.parse_args() logging.basicConfig(level=logging.INFO, format="%(message)s") - changed, errors = scan(ROOT, dry_run=args.dry_run) + changed, errors = scan(root=ROOT, dry_run=args.dry_run) LOGGER.info( "%d entries %s, %d errors", changed, From 210ec116443aabc6d662212df58a2c767d8e45bd Mon Sep 17 00:00:00 2001 From: chriopter <82179548+chriopter@users.noreply.github.com> Date: Thu, 8 Oct 2026 10:19:27 +0200 Subject: [PATCH 6/6] Use named arguments for release scanner function calls --- scan-releases.py | 28 +++---- tests/test_scan_releases.py | 141 +++++++++++++++++++++++++----------- 2 files changed, 114 insertions(+), 55 deletions(-) diff --git a/scan-releases.py b/scan-releases.py index bbb82e6..96cd97d 100755 --- a/scan-releases.py +++ b/scan-releases.py @@ -179,7 +179,7 @@ def release_record(repo, release, asset, fetcher): sha = hashlib.sha256(archive).hexdigest() if asset.get("digest") and asset["digest"] != "sha256:" + sha: raise ScanError("asset SHA-256 does not match GitHub") - eboot_md5, icon = inspect_package(archive) + eboot_md5, icon = inspect_package(archive=archive) published = datetime.datetime.fromisoformat( release["published_at"].replace("Z", "+00:00") ) @@ -218,30 +218,32 @@ def check_known_asset(repo, release, existing): def prepare(data, fetcher): - repo = repository(data.get("source")) + repo = repository(source=data.get("source")) existing = data["releases"] if not isinstance(existing, list): raise ScanError("releases must be a list") - known = {tag_key(r["tag"]): r for r in existing} + known = {tag_key(tag=r["tag"]): r for r in existing} # Optional: read the repo's .pspdx here for release selection or metadata updates. - remote = fetch_releases(repo, fetcher) + remote = fetch_releases(repo=repo, fetcher=fetcher) added, icon = [], None latest_date = max((r["published_at"] for r in existing), default="") newest_seen = False for release in remote: - key = tag_key(release["tag_name"]) + key = tag_key(tag=release["tag_name"]) if key in known: newest_seen = True - check_known_asset(repo, release, known[key]) + check_known_asset(repo=repo, release=release, existing=known[key]) continue - asset = select_asset(release) + asset = select_asset(release=release) if asset is None: LOGGER.warning( "%s %s: no unique ZIP asset, skipped", repo, release["tag_name"] ) continue try: - record, new_icon = release_record(repo, release, asset, fetcher) + record, new_icon = release_record( + repo=repo, release=release, asset=asset, fetcher=fetcher + ) except PackageError as error: LOGGER.warning("%s %s: %s, skipped", repo, release["tag_name"], error) continue @@ -254,12 +256,12 @@ def prepare(data, fetcher): if added: # Dates in this database have no time of day. Use GitHub's full # timestamp order to keep same-day backfills behind newer releases. - order = {tag_key(r["tag_name"]): index for index, r in enumerate(remote)} + order = {tag_key(tag=r["tag_name"]): index for index, r in enumerate(remote)} updated["releases"] = sorted( added + existing, key=lambda r: ( r["published_at"], - -order.get(tag_key(r["tag"]), len(remote)), + -order.get(tag_key(tag=r["tag"]), len(remote)), ), reverse=True, ) @@ -289,7 +291,7 @@ def scan(root, fetcher=fetch, dry_run=False): raise ScanError("scan_releases must be a boolean") if not flag: continue - updated, icon = prepare(data, fetcher) + updated, icon = prepare(data=data, fetcher=fetcher) if updated == data: continue icon_path = None @@ -300,8 +302,8 @@ def scan(root, fetcher=fetch, dry_run=False): content = json.dumps(updated, indent=2).encode("utf-8") if not dry_run: if icon_path: - atomic_write(icon_path, icon) - atomic_write(path, content) + atomic_write(path=icon_path, content=icon) + atomic_write(path=path, content=content) changed += 1 LOGGER.info( "%s: %d releases added", diff --git a/tests/test_scan_releases.py b/tests/test_scan_releases.py index 434d78b..ba78a0a 100644 --- a/tests/test_scan_releases.py +++ b/tests/test_scan_releases.py @@ -93,7 +93,7 @@ def get(url): class ScannerTests(unittest.TestCase): def test_invalid_historical_package_does_not_block_new_releases(self): invalid = b"old invalid package" - older = release("v0.9", date="2026-08-01T12:00:00Z", content=invalid) + older = release(tag="v0.9", date="2026-08-01T12:00:00Z", content=invalid) good = release() def fetcher(url): @@ -102,7 +102,7 @@ def fetcher(url): return invalid if "/v0.9/" in url else archive() with self.assertLogs(scanner.LOGGER, level="WARNING"): - updated, _ = scanner.prepare(entry(), fetcher) + updated, _ = scanner.prepare(data=entry(), fetcher=fetcher) self.assertEqual([r["tag"] for r in updated["releases"]], ["v2.0", "1.0"]) def test_newest_icon_uses_existing_filename_and_unicode_ids(self): @@ -119,38 +119,60 @@ def test_newest_icon_uses_existing_filename_and_unicode_ids(self): icon.parent.mkdir(parents=True) icon.write_bytes(b"previous icon") newest = release(content=content) - self.assertEqual(scanner.scan(root, network([newest], content)), (1, [])) + self.assertEqual( + scanner.scan( + root=root, + fetcher=network(items=[newest], content=content), + ), + (1, []), + ) updated = json.loads(path.read_text()) self.assertEqual(updated["media"]["icon"], "icons/app_日本語.png") self.assertEqual(icon.read_bytes(), png) self.assertEqual(list(icon.parent.iterdir()), [icon]) self.assertEqual(path.read_text(), json.dumps(updated, indent=2)) - older = release("v0.9", date="2026-08-01T12:00:00Z", content=content) + older = release(tag="v0.9", date="2026-08-01T12:00:00Z", content=content) newest_bytes = path.read_bytes() - self.assertEqual(scanner.scan(root, network([newest], content)), (0, [])) + self.assertEqual( + scanner.scan( + root=root, + fetcher=network(items=[newest], content=content), + ), + (0, []), + ) self.assertEqual(path.read_bytes(), newest_bytes) self.assertEqual( - scanner.scan(root, network([newest, older], content)), (1, []) + scanner.scan( + root=root, + fetcher=network(items=[newest, older], content=content), + ), + (1, []), ) self.assertEqual(icon.read_bytes(), png) def test_backfill_does_not_replace_existing_icon(self): png = (ROOT / "resources" / "icons" / "pspdx.png").read_bytes() content = archive(icon=png) - older = release("v0.9", date="2026-08-01T12:00:00Z", content=content) - updated, icon = scanner.prepare(entry(), network([older], content)) + older = release(tag="v0.9", date="2026-08-01T12:00:00Z", content=content) + updated, icon = scanner.prepare( + data=entry(), + fetcher=network(items=[older], content=content), + ) self.assertIsNone(icon) self.assertEqual(updated["media"], entry()["media"]) def test_public_metadata_omits_ci_settings(self): - model = get_homebrew_from_json_data("app", entry()) + model = get_homebrew_from_json_data(id="app", data=entry()) self.assertNotIn("scan_releases", model.to_dict()) self.assertIs(model.to_dict(include_ci=True)["scan_releases"], True) def test_ambiguous_release_does_not_block_a_usable_release(self): - ambiguous = release("v3.0") + ambiguous = release(tag="v3.0") ambiguous["assets"].append(copy.deepcopy(ambiguous["assets"][0])) - updated, _ = scanner.prepare(entry(), network([ambiguous, release()])) + updated, _ = scanner.prepare( + data=entry(), + fetcher=network(items=[ambiguous, release()]), + ) self.assertEqual([r["tag"] for r in updated["releases"]], ["v2.0", "1.0"]) def test_invalid_package_is_skipped_without_changing_the_entry(self): @@ -162,7 +184,8 @@ def test_invalid_package_is_skipped_without_changing_the_entry(self): before = path.read_bytes() corrupt = b"not a zip" changed, errors = scanner.scan( - root, network([release(content=corrupt)], corrupt) + root=root, + fetcher=network(items=[release(content=corrupt)], content=corrupt), ) self.assertEqual((changed, errors), (0, [])) self.assertEqual(path.read_bytes(), before) @@ -178,7 +201,7 @@ def test_failed_atomic_replace_keeps_original_and_removes_temporary_file(self): ), self.assertRaises(OSError), ): - scanner.atomic_write(path, b"updated") + scanner.atomic_write(path=path, content=b"updated") self.assertEqual(path.read_bytes(), b"original") self.assertEqual(list(root.iterdir()), [path]) @@ -189,12 +212,18 @@ def test_invalid_opt_in_is_reported_without_requests(self): data = entry() data["scan_releases"] = "true" (root / "data" / "app.json").write_text(json.dumps(data)) - changed, errors = scanner.scan(root, lambda url: self.fail(url)) + changed, errors = scanner.scan( + root=root, + fetcher=lambda url: self.fail(url), + ) self.assertEqual((changed, len(errors)), (0, 1)) def test_new_releases_preserve_curated_fields_and_history(self): original = entry() - updated, icon = scanner.prepare(original, network([release()])) + updated, icon = scanner.prepare( + data=original, + fetcher=network(items=[release()]), + ) self.assertEqual(original, entry()) for key in original.keys() - {"releases"}: self.assertEqual(updated[key], original[key]) @@ -206,48 +235,55 @@ def test_new_releases_preserve_curated_fields_and_history(self): self.assertIsNone(icon) def test_known_tag_with_leading_v_is_not_downloaded_or_duplicated(self): - known = release("v1.0") + known = release(tag="v1.0") known["assets"][0]["digest"] = "sha256:" + "a" * 64 def get(url): self.assertIn("api.github.com", url) return json.dumps([known]).encode() - updated, _ = scanner.prepare(entry(), get) + updated, _ = scanner.prepare(data=entry(), fetcher=get) self.assertEqual(updated, entry()) def test_known_asset_replacement_is_reported_without_blocking_updates(self): with self.assertLogs(scanner.LOGGER, level="WARNING") as logs: - updated, _ = scanner.prepare(entry(), network([release("v1.0"), release()])) + updated, _ = scanner.prepare( + data=entry(), + fetcher=network(items=[release(tag="v1.0"), release()]), + ) self.assertIn("known release asset changed", logs.output[0]) self.assertEqual([r["tag"] for r in updated["releases"]], ["v2.0", "1.0"]) self.assertEqual(updated["releases"][1], entry()["releases"][0]) def test_drafts_and_prereleases_are_ignored(self): - draft, preview = release(), release("v3.0") + draft, preview = release(), release(tag="v3.0") draft["draft"], preview["prerelease"] = True, True self.assertEqual( - scanner.prepare(entry(), network([draft, preview]))[0], entry() + scanner.prepare( + data=entry(), + fetcher=network(items=[draft, preview]), + )[0], + entry(), ) def test_pagination_reads_history_and_sorts_by_publication(self): - first = [release(f"v{i}") for i in range(100)] - last = release("v100", date="2026-10-02T12:00:00Z") + first = [release(tag=f"v{i}") for i in range(100)] + last = release(tag="v100", date="2026-10-02T12:00:00Z") def get(url): return json.dumps(first if url.endswith("page=1") else [last]).encode() - fetched = scanner.fetch_releases("test/app", get) + fetched = scanner.fetch_releases(repo="test/app", fetcher=get) self.assertEqual(len(fetched), 101) self.assertEqual(fetched[0]["tag_name"], "v100") def test_same_day_backfill_stays_behind_the_known_newest_release(self): data = entry() data["releases"][0]["published_at"] = "2026-10-01" - newest = release("v1.0", date="2026-10-01T18:00:00Z") + newest = release(tag="v1.0", date="2026-10-01T18:00:00Z") newest["assets"][0]["digest"] = "sha256:" + "a" * 64 - older = release("v0.9", date="2026-10-01T08:00:00Z") - updated, _ = scanner.prepare(data, network([older, newest])) + older = release(tag="v0.9", date="2026-10-01T08:00:00Z") + updated, _ = scanner.prepare(data=data, fetcher=network(items=[older, newest])) self.assertEqual([r["tag"] for r in updated["releases"]], ["1.0", "v0.9"]) def test_truncated_download_is_isolated_to_its_entry(self): @@ -265,18 +301,18 @@ def test_truncated_download_is_isolated_to_its_entry(self): def get(url): if "/test/bad/" in url: raise http.client.IncompleteRead(b"partial", 100) - return network([release()])(url) + return network(items=[release()])(url) - changed, errors = scanner.scan(root, get) + changed, errors = scanner.scan(root=root, fetcher=get) self.assertEqual((changed, len(errors)), (1, 1)) self.assertEqual(path.read_bytes(), before) def test_zip_selection_requires_a_unique_match(self): r = release() r["assets"] += [copy.deepcopy(r["assets"][0])] - self.assertIsNone(scanner.select_asset(r)) + self.assertIsNone(scanner.select_asset(release=r)) r["assets"][1]["name"] = "linux.zip" - self.assertEqual(scanner.select_asset(r)["name"], "psp.zip") + self.assertEqual(scanner.select_asset(release=r)["name"], "psp.zip") def test_wrong_size_digest_and_download_host_are_rejected(self): for field, value in [ @@ -288,7 +324,7 @@ def test_wrong_size_digest_and_download_host_are_rejected(self): r = release() r["assets"][0][field] = value with self.assertRaises(scanner.ScanError): - scanner.prepare(entry(), network([r])) + scanner.prepare(data=entry(), fetcher=network(items=[r])) def test_unsafe_ambiguous_and_invalid_packages_are_rejected(self): for entries in [ @@ -298,16 +334,16 @@ def test_unsafe_ambiguous_and_invalid_packages_are_rejected(self): {"EBOOT.PBP": b"x", "eboot.pbp": b"x"}, ]: with self.subTest(entries=entries), self.assertRaises(scanner.ScanError): - scanner.inspect_package(archive(entries)) + scanner.inspect_package(archive=archive(entries=entries)) def test_corrupt_non_eboot_resource_is_rejected(self): eboot = b"\x00PBP" + b"\x00\x00\x01\x00" + (40).to_bytes(4, "little") * 8 content = archive( - {"EBOOT.PBP": eboot, "resource.dat": b"unique resource bytes"} + entries={"EBOOT.PBP": eboot, "resource.dat": b"unique resource bytes"} ) damaged = content.replace(b"unique resource bytes", b"broken resource bytes", 1) with self.assertRaisesRegex(scanner.ScanError, "CRC"): - scanner.inspect_package(damaged) + scanner.inspect_package(archive=damaged) def test_release_without_zip_does_not_block_other_releases(self): with tempfile.TemporaryDirectory() as directory: @@ -315,9 +351,12 @@ def test_release_without_zip_does_not_block_other_releases(self): (root / "data").mkdir() path = root / "data" / "app.json" path.write_text(json.dumps(entry())) - older = release("v1.5", date="2026-09-15T12:00:00Z") + older = release(tag="v1.5", date="2026-09-15T12:00:00Z") older["assets"] = [] - changed, errors = scanner.scan(root, network([release(), older])) + changed, errors = scanner.scan( + root=root, + fetcher=network(items=[release(), older]), + ) self.assertEqual((changed, errors), (1, [])) self.assertEqual( [r["tag"] for r in json.loads(path.read_text())["releases"]], @@ -351,7 +390,7 @@ def test_only_opted_in_entries_are_requested(self): data["scan_releases"] = flag (root / "data" / "app.json").write_text(json.dumps(data)) self.assertEqual( - scanner.scan(root, lambda url: self.fail(url)), (0, []) + scanner.scan(root=root, fetcher=lambda url: self.fail(url)), (0, []) ) def test_failed_entry_is_unchanged_while_other_entry_updates(self): @@ -364,7 +403,10 @@ def test_failed_entry_is_unchanged_while_other_entry_updates(self): bad.write_text(json.dumps(broken)) before = bad.read_bytes() good.write_text(json.dumps(entry())) - changed, errors = scanner.scan(root, network([release()])) + changed, errors = scanner.scan( + root=root, + fetcher=network(items=[release()]), + ) self.assertEqual(changed, 1) self.assertEqual(len(errors), 1) self.assertEqual(bad.read_bytes(), before) @@ -377,18 +419,33 @@ def test_dry_run_and_second_scan_do_not_write(self): path = root / "data" / "app.json" path.write_text(json.dumps(entry())) before = path.read_bytes() - self.assertEqual(scanner.scan(root, network([release()]), True), (1, [])) + self.assertEqual( + scanner.scan( + root=root, + fetcher=network(items=[release()]), + dry_run=True, + ), + (1, []), + ) self.assertEqual(path.read_bytes(), before) - self.assertEqual(scanner.scan(root, network([release()])), (1, [])) + self.assertEqual( + scanner.scan(root=root, fetcher=network(items=[release()])), + (1, []), + ) after = path.read_bytes() - self.assertEqual(scanner.scan(root, network([release()])), (0, [])) + self.assertEqual( + scanner.scan(root=root, fetcher=network(items=[release()])), + (0, []), + ) self.assertEqual(path.read_bytes(), after) def test_ci_controls_survive_existing_enrichment_script_round_trip(self): for flag in [True, False]: data = entry() data["scan_releases"] = flag - updated = get_homebrew_from_json_data("app", data).to_dict(include_ci=True) + updated = get_homebrew_from_json_data(id="app", data=data).to_dict( + include_ci=True, + ) self.assertIs(updated["scan_releases"], flag)