chore: format VPM scripts and tests

.github/scripts/sync_vpm_yanks.py

@@ -7,7 +7,13 @@
 import sys
 from typing import Any
 
-from vpm_common import PACKAGE_NAME, SOURCE_REPOSITORY, UpdateError, VPM_PATH, required_env
+from vpm_common import (
+    PACKAGE_NAME,
+    SOURCE_REPOSITORY,
+    VPM_PATH,
+    UpdateError,
+    required_env,
+)
 from vpm_listing import apply_yank_policy, load_listing, write_listing
 from vpm_policy import fetch_yank_policy_snapshot
 
@@ -74,4 +80,4 @@ def main() -> int:
         raise SystemExit(main())
     except UpdateError as error:
         print(f"error: {error}", file=sys.stderr)
-        raise SystemExit(1)
+        raise SystemExit(1)

.github/scripts/update_vpm.py

@@ -9,7 +9,7 @@
 from pathlib import Path
 
 from vpm_archive import download_archive, load_manifest, verify_archive_sha256
-from vpm_common import UpdateError, VPM_PATH
+from vpm_common import VPM_PATH, UpdateError
 from vpm_listing import apply_yank_policy, load_listing, update_listing, write_listing
 from vpm_payload import validate_payload, verify_release_commit
 from vpm_policy import fetch_yank_policy_snapshot

.github/scripts/vpm_archive.py

@@ -8,7 +8,7 @@
 import zipfile
 from collections.abc import Mapping
 from pathlib import Path, PurePosixPath
-from typing import Any, BinaryIO
+from typing import Any, IO
 
 from vpm_common import (
     EXPECTED_LICENSE,
@@ -32,10 +32,15 @@ def download_archive(url: str, destination: Path) -> str:
     )
     digest = hashlib.sha256()
     total = 0
-    with urllib.request.urlopen(request, timeout=60) as response, destination.open("wb") as output:
+    with (
+        urllib.request.urlopen(request, timeout=60) as response,
+        destination.open("wb") as output,
+    ):
         final_url = response.geturl()
         if not final_url.startswith("https://"):
-            raise UpdateError(f"Release asset redirected to a non-HTTPS URL: {final_url!r}.")
+            raise UpdateError(
+                f"Release asset redirected to a non-HTTPS URL: {final_url!r}."
+            )
         while chunk := response.read(1024 * 1024):
             total += len(chunk)
             if total > MAX_ARCHIVE_BYTES:
@@ -69,7 +74,7 @@ def is_unsafe_zip_path(name: str) -> bool:
     )
 
 
-def read_limited(stream: BinaryIO, limit: int) -> bytes:
+def read_limited(stream: IO[bytes], limit: int) -> bytes:
     """Read at most limit bytes and fail before retaining oversized content."""
     chunks: list[bytes] = []
     remaining = limit + 1
@@ -81,7 +86,9 @@ def read_limited(stream: BinaryIO, limit: int) -> bytes:
         remaining -= len(chunk)
     data = b"".join(chunks)
     if len(data) > limit:
-        raise UpdateError("package.json exceeds the 1 MiB safety limit after decompression.")
+        raise UpdateError(
+            "package.json exceeds the 1 MiB safety limit after decompression."
+        )
     return data
 
 
@@ -90,7 +97,9 @@ def validate_required_text_fields(manifest: Mapping[str, Any]) -> None:
     for field in REQUIRED_TEXT_FIELDS:
         value = manifest.get(field)
         if not isinstance(value, str) or not value.strip():
-            raise UpdateError(f"package.json field {field!r} must be a non-empty string.")
+            raise UpdateError(
+                f"package.json field {field!r} must be a non-empty string."
+            )
 
 
 def validate_optional_url(manifest: dict[str, Any], field: str, expected: str) -> None:
@@ -142,7 +151,9 @@ def load_manifest(
     try:
         manifest = strict_json_loads(raw_manifest.decode("utf-8-sig"))
     except (UnicodeDecodeError, ValueError) as error:
-        raise UpdateError(f"package.json is not valid strict UTF-8 JSON: {error}") from error
+        raise UpdateError(
+            f"package.json is not valid strict UTF-8 JSON: {error}"
+        ) from error
 
     if not isinstance(manifest, dict):
         raise UpdateError("package.json must contain a JSON object.")
@@ -163,7 +174,9 @@ def load_manifest(
     validate_required_text_fields(manifest)
     embedded_url = manifest.get("url")
     if embedded_url not in (None, "", payload["package_url"]):
-        raise UpdateError(f"package.json contains an unexpected download URL: {embedded_url!r}.")
+        raise UpdateError(
+            f"package.json contains an unexpected download URL: {embedded_url!r}."
+        )
 
     validate_optional_url(manifest, "changelogUrl", payload["changelog_url"])
     validate_optional_url(manifest, "licensesUrl", payload["licenses_url"])

.github/scripts/vpm_common.py

@@ -5,9 +5,10 @@
 import json
 import os
 import re
+from collections.abc import Mapping
 from functools import total_ordering
 from pathlib import Path
-from typing import Any, Mapping
+from typing import Any
 
 PACKAGE_NAME = "jp.penguin.purebase"
 SOURCE_REPOSITORY = "Penguin-Repository/Pure-Base"
@@ -66,7 +67,9 @@ def __lt__(self, other: object) -> bool:
 def parse_semver(version: str) -> SemanticVersion:
     """Parse a strict ASCII SemVer 2 core and prerelease value."""
     if not isinstance(version, str) or not version.isascii():
-        raise UpdateError(f"Repository contains an unsupported version key: {version!r}.")
+        raise UpdateError(
+            f"Repository contains an unsupported version key: {version!r}."
+        )
 
     core_text, separator, prerelease_text = version.partition("-")
     core_identifiers = core_text.split(".")
@@ -75,15 +78,23 @@ def parse_semver(version: str) -> SemanticVersion:
         or (len(identifier) > 1 and identifier.startswith("0"))
         for identifier in core_identifiers
     ):
-        raise UpdateError(f"Repository contains an unsupported version key: {version!r}.")
+        raise UpdateError(
+            f"Repository contains an unsupported version key: {version!r}."
+        )
 
     prerelease: list[tuple[int, int | str]] = []
     if separator:
         if not prerelease_text:
-            raise UpdateError(f"Repository contains an unsupported version key: {version!r}.")
+            raise UpdateError(
+                f"Repository contains an unsupported version key: {version!r}."
+            )
         for identifier in prerelease_text.split("."):
-            if not identifier or not all(character.isalnum() or character == "-" for character in identifier):
-                raise UpdateError(f"Repository contains an unsupported version key: {version!r}.")
+            if not identifier or not all(
+                character.isalnum() or character == "-" for character in identifier
+            ):
+                raise UpdateError(
+                    f"Repository contains an unsupported version key: {version!r}."
+                )
             if identifier.isdecimal():
                 if len(identifier) > 1 and identifier.startswith("0"):
                     raise UpdateError(

.github/scripts/vpm_listing.py

@@ -146,7 +146,9 @@ def write_listing(path: Path, listing: dict[str, Any]) -> None:
     try:
         serialized = strict_json_dumps(listing)
     except (TypeError, ValueError) as error:
-        raise UpdateError(f"Refusing to write non-standard JSON to {path}: {error}") from error
+        raise UpdateError(
+            f"Refusing to write non-standard JSON to {path}: {error}"
+        ) from error
 
     temporary_path = path.with_suffix(path.suffix + ".tmp")
     temporary_path.write_text(serialized, encoding="utf-8", newline="\n")

.github/scripts/vpm_payload.py

@@ -48,9 +48,13 @@ def validate_payload(values: Mapping[str, str] | None = None) -> dict[str, str]:
     if tag != version:
         raise UpdateError(f"tag {tag!r} does not match version {version!r}.")
     if not COMMIT_RE.fullmatch(commit_sha):
-        raise UpdateError("commitSha must be a 40-character hexadecimal Git commit SHA.")
+        raise UpdateError(
+            "commitSha must be a 40-character hexadecimal Git commit SHA."
+        )
     if not COMMIT_RE.fullmatch(policy_commit_sha):
-        raise UpdateError("policyCommitSha must be a 40-character hexadecimal Git commit SHA.")
+        raise UpdateError(
+            "policyCommitSha must be a 40-character hexadecimal Git commit SHA."
+        )
     if not SHA256_RE.fullmatch(expected_sha256):
         raise UpdateError("sha256 must be a 64-character hexadecimal SHA-256 value.")
 
@@ -103,7 +107,9 @@ def github_api_get(url: str) -> dict[str, Any]:
     try:
         result = strict_json_loads(body)
     except (UnicodeDecodeError, ValueError) as error:
-        raise UpdateError(f"GitHub API returned invalid JSON for {url}: {error}") from error
+        raise UpdateError(
+            f"GitHub API returned invalid JSON for {url}: {error}"
+        ) from error
     if not isinstance(result, dict):
         raise UpdateError(f"GitHub API returned a non-object response for {url}.")
     return result
@@ -116,7 +122,9 @@ def resolve_tag_commit(
 ) -> str:
     """Resolve a lightweight or annotated Git tag to its commit SHA."""
     encoded_tag = urllib.parse.quote(tag, safe="")
-    reference = api_get(f"{GITHUB_API_ROOT}/repos/{repository}/git/ref/tags/{encoded_tag}")
+    reference = api_get(
+        f"{GITHUB_API_ROOT}/repos/{repository}/git/ref/tags/{encoded_tag}"
+    )
     target = reference.get("object")
 
     for _ in range(MAX_TAG_DEREFERENCE_DEPTH):
@@ -129,7 +137,9 @@ def resolve_tag_commit(
         if object_type == "commit":
             return object_sha.lower()
         if object_type != "tag":
-            raise UpdateError(f"Tag {tag!r} resolves to unsupported object type {object_type!r}.")
+            raise UpdateError(
+                f"Tag {tag!r} resolves to unsupported object type {object_type!r}."
+            )
         annotated_tag = api_get(
             f"{GITHUB_API_ROOT}/repos/{repository}/git/tags/{object_sha}"
         )

.github/scripts/vpm_policy.py

@@ -31,19 +31,27 @@ def load_yank_policy(raw: bytes) -> dict[str, Any]:
     try:
         text = raw.decode("utf-8")
     except (UnicodeDecodeError, ValueError) as error:
-        raise UpdateError(f"Yank policy is not valid strict UTF-8 JSON: {error}") from error
+        raise UpdateError(
+            f"Yank policy is not valid strict UTF-8 JSON: {error}"
+        ) from error
     if text != text.rstrip(" \t\r\n"):
-        raise UpdateError("Yank policy must not contain trailing bytes after its JSON document.")
+        raise UpdateError(
+            "Yank policy must not contain trailing bytes after its JSON document."
+        )
     try:
         policy = strict_json_loads(text)
     except ValueError as error:
-        raise UpdateError(f"Yank policy is not valid strict UTF-8 JSON: {error}") from error
+        raise UpdateError(
+            f"Yank policy is not valid strict UTF-8 JSON: {error}"
+        ) from error
     if not isinstance(policy, dict) or set(policy) != {
         "schemaVersion",
         "package",
         "versions",
     }:
-        raise UpdateError("Yank policy must contain exactly schemaVersion, package, and versions.")
+        raise UpdateError(
+            "Yank policy must contain exactly schemaVersion, package, and versions."
+        )
     if policy["schemaVersion"] != 1 or isinstance(policy["schemaVersion"], bool):
         raise UpdateError("Yank policy schemaVersion must be integer 1.")
     if policy["package"] != PACKAGE_NAME:
@@ -54,7 +62,9 @@ def load_yank_policy(raw: bytes) -> dict[str, Any]:
     for version, reason in versions.items():
         parse_semver(version)
         if not isinstance(reason, str) or not reason.strip():
-            raise UpdateError(f"Yank reason for {version!r} must be a non-empty string.")
+            raise UpdateError(
+                f"Yank reason for {version!r} must be a non-empty string."
+            )
     return policy
 
 
@@ -67,7 +77,9 @@ def decode_policy_content(response: Mapping[str, Any], ref: str) -> bytes:
         encoded_content = content.replace("\r", "").replace("\n", "")
         return base64.b64decode(encoded_content.encode("ascii"), validate=True)
     except (UnicodeEncodeError, ValueError) as error:
-        raise UpdateError(f"GitHub API returned invalid base64 yank policy content for {ref}.") from error
+        raise UpdateError(
+            f"GitHub API returned invalid base64 yank policy content for {ref}."
+        ) from error
 
 
 def fetch_yank_policy_snapshot(
@@ -76,7 +88,9 @@ def fetch_yank_policy_snapshot(
 ) -> dict[str, Any]:
     """Return a reachable immutable policy only when it matches master exactly."""
     if not COMMIT_RE.fullmatch(policy_commit_sha):
-        raise UpdateError("policyCommitSha must be a 40-character hexadecimal Git commit SHA.")
+        raise UpdateError(
+            "policyCommitSha must be a 40-character hexadecimal Git commit SHA."
+        )
     policy_commit_sha = policy_commit_sha.lower()
     repository_url = f"{GITHUB_API_ROOT}/repos/{SOURCE_REPOSITORY}"
     comparison = api_get(
@@ -98,4 +112,4 @@ def fetch_yank_policy_snapshot(
     load_yank_policy(current_raw)
     if requested_raw != current_raw:
         raise UpdateError("Policy snapshot no longer matches current master policy.")
-    return requested_policy
+    return requested_policy