diff --git a/.github/scripts/sync_vpm_yanks.py b/.github/scripts/sync_vpm_yanks.py index e86dc5d..ede186d 100644 --- a/.github/scripts/sync_vpm_yanks.py +++ b/.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) \ No newline at end of file + raise SystemExit(1) diff --git a/.github/scripts/update_vpm.py b/.github/scripts/update_vpm.py index e07d40b..47e0d5d 100644 --- a/.github/scripts/update_vpm.py +++ b/.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 diff --git a/.github/scripts/vpm_archive.py b/.github/scripts/vpm_archive.py index ce40cba..aaca1b5 100644 --- a/.github/scripts/vpm_archive.py +++ b/.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"]) diff --git a/.github/scripts/vpm_common.py b/.github/scripts/vpm_common.py index bc9dc14..9599543 100644 --- a/.github/scripts/vpm_common.py +++ b/.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( diff --git a/.github/scripts/vpm_listing.py b/.github/scripts/vpm_listing.py index 79cb913..5b8e2ec 100644 --- a/.github/scripts/vpm_listing.py +++ b/.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") diff --git a/.github/scripts/vpm_payload.py b/.github/scripts/vpm_payload.py index 635d05f..e301b07 100644 --- a/.github/scripts/vpm_payload.py +++ b/.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}" ) diff --git a/.github/scripts/vpm_policy.py b/.github/scripts/vpm_policy.py index 9f08b1c..33005e1 100644 --- a/.github/scripts/vpm_policy.py +++ b/.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 \ No newline at end of file + return requested_policy diff --git a/.github/tests/test_vpm_receiver.py b/.github/tests/test_vpm_receiver.py index c952e93..921a44e 100644 --- a/.github/tests/test_vpm_receiver.py +++ b/.github/tests/test_vpm_receiver.py @@ -16,20 +16,29 @@ SCRIPTS = Path(__file__).resolve().parents[1] / "scripts" sys.path.insert(0, str(SCRIPTS)) -from vpm_archive import ( # noqa: E402 +from vpm_archive import ( is_unsafe_zip_path, load_manifest, read_limited, verify_archive_sha256, ) -from vpm_common import ( # noqa: E402 +from vpm_common import ( MAX_PACKAGE_JSON_BYTES, UpdateError, strict_json_dumps, strict_json_loads, ) -from vpm_listing import apply_yank_policy, load_listing, update_listing, write_listing # noqa: E402 -from vpm_payload import resolve_tag_commit, validate_payload, verify_release_commit # noqa: E402 +from vpm_listing import ( + apply_yank_policy, + load_listing, + update_listing, + write_listing, +) +from vpm_payload import ( + resolve_tag_commit, + validate_payload, + verify_release_commit, +) COMMIT_SHA = "a" * 40 OTHER_COMMIT_SHA = "b" * 40 @@ -51,7 +60,9 @@ def valid_payload( "https://github.com/Penguin-Repository/Pure-Base/releases/download/" f"{version}/{asset_name}" ) - release_url = f"https://github.com/Penguin-Repository/Pure-Base/releases/tag/{version}" + release_url = ( + f"https://github.com/Penguin-Repository/Pure-Base/releases/tag/{version}" + ) return { "package_name": "jp.penguin.purebase", "source_repository": "Penguin-Repository/Pure-Base", @@ -64,8 +75,7 @@ def valid_payload( "release_url": release_url, "changelog_url": release_url, "licenses_url": ( - "https://github.com/Penguin-Repository/Pure-Base/blob/" - f"{COMMIT_SHA}/LICENSE" + f"https://github.com/Penguin-Repository/Pure-Base/blob/{COMMIT_SHA}/LICENSE" ), } @@ -216,7 +226,9 @@ def test_accepts_prerelease_payload_with_separate_policy_commit(self) -> None: validated = validate_payload(values) self.assertEqual(validated["version"], prerelease) - self.assertEqual(validated["asset_name"], f"jp.penguin.purebase-{prerelease}.zip") + self.assertEqual( + validated["asset_name"], f"jp.penguin.purebase-{prerelease}.zip" + ) self.assertEqual(validated["policy_commit_sha"], OTHER_COMMIT_SHA) def test_error_names_canonical_package_url_field(self) -> None: @@ -296,7 +308,10 @@ def test_load_manifest_enriches_trusted_metadata(self) -> None: def test_load_manifest_rejects_missing_required_vpm_text(self) -> None: for field in ("displayName", "description", "unity"): - with self.subTest(field=field), tempfile.TemporaryDirectory() as temporary_directory: + with ( + self.subTest(field=field), + tempfile.TemporaryDirectory() as temporary_directory, + ): archive_path = Path(temporary_directory) / "package.zip" manifest = valid_manifest() manifest[field] = " " @@ -337,9 +352,7 @@ def test_rejects_overwriting_existing_version(self) -> None: manifest = valid_manifest() listing: dict[str, object] = { "packages": { - "jp.penguin.purebase": { - "versions": {VERSION: manifest.copy()} - } + "jp.penguin.purebase": {"versions": {VERSION: manifest.copy()}} } } changed_manifest = manifest.copy() @@ -373,9 +386,7 @@ def test_rejects_invalid_existing_version_value_without_mutation(self) -> None: with self.subTest(invalid_value=invalid_value): listing: dict[str, object] = { "packages": { - "jp.penguin.purebase": { - "versions": {VERSION: invalid_value} - } + "jp.penguin.purebase": {"versions": {VERSION: invalid_value}} } } before = copy.deepcopy(listing) @@ -394,7 +405,9 @@ def test_rejects_invalid_existing_package_value_without_mutation(self) -> None: } before = copy.deepcopy(listing) - with self.assertRaisesRegex(UpdateError, "Package entry .* is not an object"): + with self.assertRaisesRegex( + UpdateError, "Package entry .* is not an object" + ): update_listing(listing, manifest) self.assertEqual(listing, before) @@ -444,12 +457,16 @@ def test_policy_failure_prevents_archive_or_listing_access(self) -> None: listing_path.write_text(original, encoding="utf-8") with ( patch.object(update_vpm, "VPM_PATH", listing_path), - patch.object(update_vpm, "validate_payload", return_value=valid_payload()), + patch.object( + update_vpm, "validate_payload", return_value=valid_payload() + ), patch.object(update_vpm, "verify_release_commit"), patch.object( update_vpm, "fetch_yank_policy_snapshot", - side_effect=UpdateError("Policy snapshot no longer matches current master policy"), + side_effect=UpdateError( + "Policy snapshot no longer matches current master policy" + ), ), patch.object( update_vpm, @@ -461,9 +478,9 @@ def test_policy_failure_prevents_archive_or_listing_access(self) -> None: "load_listing", side_effect=AssertionError("Listing must not be loaded."), ), + self.assertRaises(UpdateError), ): - with self.assertRaises(UpdateError): - update_vpm.process_update() + update_vpm.process_update() self.assertEqual(listing_path.read_text(encoding="utf-8"), original) diff --git a/.github/tests/test_vpm_yanks.py b/.github/tests/test_vpm_yanks.py index f9b683d..e7e5d2e 100644 --- a/.github/tests/test_vpm_yanks.py +++ b/.github/tests/test_vpm_yanks.py @@ -17,7 +17,7 @@ WORKFLOWS = Path(__file__).resolve().parents[1] / "workflows" sys.path.insert(0, str(SCRIPTS)) -from vpm_common import UpdateError # noqa: E402 +from vpm_common import UpdateError PACKAGE_NAME = "jp.penguin.purebase" SOURCE_REPOSITORY = "Penguin-Repository/Pure-Base" @@ -72,9 +72,7 @@ def listing_with_versions(*versions: str) -> dict[str, object]: return { "packages": { PACKAGE_NAME: { - "versions": { - version: version_metadata(version) for version in versions - } + "versions": {version: version_metadata(version) for version in versions} } } } @@ -87,7 +85,10 @@ def test_accepts_exactly_64_kib_policy(self) -> None: raw = policy_document_of_size(64 * 1024) self.assertEqual(MAX_YANK_POLICY_BYTES, 64 * 1024) self.assertEqual(len(raw), MAX_YANK_POLICY_BYTES) - self.assertEqual(load_yank_policy(raw)["versions"], {VERSION: "x" * (len(raw) - len(policy_document({VERSION: ""})))}) + self.assertEqual( + load_yank_policy(raw)["versions"], + {VERSION: "x" * (len(raw) - len(policy_document({VERSION: ""})))}, + ) def test_rejects_policy_larger_than_64_kib(self) -> None: from vpm_policy import MAX_YANK_POLICY_BYTES, load_yank_policy @@ -118,12 +119,27 @@ def test_rejects_non_strict_json_policy_cases(self) -> None: from vpm_policy import load_yank_policy invalid_documents = ( - ("utf8_bom", b'\xef\xbb\xbf{"schemaVersion":1,"package":"jp.penguin.purebase","versions":{}}'), - ("invalid_utf8", b'\xff'), - ("trailing_content", b'{"schemaVersion":1,"package":"jp.penguin.purebase","versions":{}} trailing'), - ("trailing_whitespace", b'{"schemaVersion":1,"package":"jp.penguin.purebase","versions":{}}\n'), - ("nan_constant", b'{"schemaVersion":NaN,"package":"jp.penguin.purebase","versions":{}}'), - ("infinity_constant", b'{"schemaVersion":Infinity,"package":"jp.penguin.purebase","versions":{}}'), + ( + "utf8_bom", + b'\xef\xbb\xbf{"schemaVersion":1,"package":"jp.penguin.purebase","versions":{}}', + ), + ("invalid_utf8", b"\xff"), + ( + "trailing_content", + b'{"schemaVersion":1,"package":"jp.penguin.purebase","versions":{}} trailing', + ), + ( + "trailing_whitespace", + b'{"schemaVersion":1,"package":"jp.penguin.purebase","versions":{}}\n', + ), + ( + "nan_constant", + b'{"schemaVersion":NaN,"package":"jp.penguin.purebase","versions":{}}', + ), + ( + "infinity_constant", + b'{"schemaVersion":Infinity,"package":"jp.penguin.purebase","versions":{}}', + ), ) for contract_id, raw in invalid_documents: with self.subTest(contract_id=contract_id), self.assertRaises(UpdateError): @@ -151,15 +167,42 @@ def test_rejects_invalid_schema_policy_cases(self) -> None: from vpm_policy import load_yank_policy invalid_documents = ( - ("boolean_schema_version", b'{"schemaVersion":true,"package":"jp.penguin.purebase","versions":{}}'), - ("string_schema_version", b'{"schemaVersion":"1","package":"jp.penguin.purebase","versions":{}}'), - ("wrong_package", b'{"schemaVersion":1,"package":"other.package","versions":{}}'), - ("unexpected_top_level_key", b'{"schemaVersion":1,"package":"jp.penguin.purebase","versions":{},"extra":true}'), - ("non_object_versions", b'{"schemaVersion":1,"package":"jp.penguin.purebase","versions":[]}'), - ("invalid_semver_version_key", b'{"schemaVersion":1,"package":"jp.penguin.purebase","versions":{"v0.1.0":"reason"}}'), - ("empty_reason", b'{"schemaVersion":1,"package":"jp.penguin.purebase","versions":{"0.1.0":""}}'), - ("whitespace_reason", b'{"schemaVersion":1,"package":"jp.penguin.purebase","versions":{"0.1.0":" "}}'), - ("non_string_reason", b'{"schemaVersion":1,"package":"jp.penguin.purebase","versions":{"0.1.0":true}}'), + ( + "boolean_schema_version", + b'{"schemaVersion":true,"package":"jp.penguin.purebase","versions":{}}', + ), + ( + "string_schema_version", + b'{"schemaVersion":"1","package":"jp.penguin.purebase","versions":{}}', + ), + ( + "wrong_package", + b'{"schemaVersion":1,"package":"other.package","versions":{}}', + ), + ( + "unexpected_top_level_key", + b'{"schemaVersion":1,"package":"jp.penguin.purebase","versions":{},"extra":true}', + ), + ( + "non_object_versions", + b'{"schemaVersion":1,"package":"jp.penguin.purebase","versions":[]}', + ), + ( + "invalid_semver_version_key", + b'{"schemaVersion":1,"package":"jp.penguin.purebase","versions":{"v0.1.0":"reason"}}', + ), + ( + "empty_reason", + b'{"schemaVersion":1,"package":"jp.penguin.purebase","versions":{"0.1.0":""}}', + ), + ( + "whitespace_reason", + b'{"schemaVersion":1,"package":"jp.penguin.purebase","versions":{"0.1.0":" "}}', + ), + ( + "non_string_reason", + b'{"schemaVersion":1,"package":"jp.penguin.purebase","versions":{"0.1.0":true}}', + ), ) for contract_id, raw in invalid_documents: with self.subTest(contract_id=contract_id), self.assertRaises(UpdateError): @@ -167,7 +210,9 @@ def test_rejects_invalid_schema_policy_cases(self) -> None: class YankPolicyFetchTests(unittest.TestCase): - def assert_sync_rejection_leaves_listing_unchanged(self, error: UpdateError) -> None: + def assert_sync_rejection_leaves_listing_unchanged( + self, error: UpdateError + ) -> None: """Assert a rejected policy sync never persists a candidate listing.""" import sync_vpm_yanks @@ -188,9 +233,9 @@ def assert_sync_rejection_leaves_listing_unchanged(self, error: UpdateError) -> "fetch_yank_policy_snapshot", side_effect=error, ), + self.assertRaises(UpdateError), ): - with self.assertRaises(UpdateError): - sync_vpm_yanks.process_yank_sync() + sync_vpm_yanks.process_yank_sync() self.assertEqual(listing_path.read_text(encoding="utf-8"), original) @@ -234,7 +279,9 @@ def test_rejects_base64_content_with_non_api_whitespace_characters(self) -> None for content in invalid_contents: with self.subTest(content=content), self.assertRaises(UpdateError): - decode_policy_content({"content": content, "encoding": "base64"}, COMMIT_SHA) + decode_policy_content( + {"content": content, "encoding": "base64"}, COMMIT_SHA + ) def test_rejects_invalid_policy_commit_before_github_api_access(self) -> None: from vpm_policy import fetch_yank_policy_snapshot @@ -266,15 +313,23 @@ def api_get(url: str) -> dict[str, object]: if "/compare/" in url: return {"status": "ahead"} if f"ref={COMMIT_SHA}" in url: - return {"content": base64.b64encode(requested).decode("ascii"), "encoding": "base64"} + return { + "content": base64.b64encode(requested).decode("ascii"), + "encoding": "base64", + } if "ref=master" in url: - return {"content": base64.b64encode(current).decode("ascii"), "encoding": "base64"} + return { + "content": base64.b64encode(current).decode("ascii"), + "encoding": "base64", + } return {"sha": CURRENT_COMMIT_SHA} with self.assertRaises(UpdateError): fetch_yank_policy_snapshot(COMMIT_SHA, api_get=api_get) - def test_accepts_old_reachable_snapshot_when_current_policy_content_is_identical(self) -> None: + def test_accepts_old_reachable_snapshot_when_current_policy_content_is_identical( + self, + ) -> None: from vpm_policy import fetch_yank_policy_snapshot raw = policy_document({VERSION: "same desired state"}) @@ -283,13 +338,18 @@ def api_get(url: str) -> dict[str, object]: if "/compare/" in url: return {"status": "ahead"} if f"ref={COMMIT_SHA}" in url or "ref=master" in url: - return {"content": base64.b64encode(raw).decode("ascii"), "encoding": "base64"} + return { + "content": base64.b64encode(raw).decode("ascii"), + "encoding": "base64", + } return {"sha": CURRENT_COMMIT_SHA} policy = fetch_yank_policy_snapshot(COMMIT_SHA, api_get=api_get) self.assertEqual(policy["versions"], {VERSION: "same desired state"}) - def test_stale_policy_snapshot_rejection_leaves_listing_byte_identical(self) -> None: + def test_stale_policy_snapshot_rejection_leaves_listing_byte_identical( + self, + ) -> None: self.assert_sync_rejection_leaves_listing_unchanged( UpdateError("Policy snapshot no longer matches current master policy") ) @@ -373,22 +433,32 @@ def test_yank_mutation_is_limited_to_allowed_vrc_get_yanked_path(self) -> None: listing = listing_with_versions(VERSION) before = copy.deepcopy(listing) - policy = {"schemaVersion": 1, "package": PACKAGE_NAME, "versions": {VERSION: "new reason"}} + policy = { + "schemaVersion": 1, + "package": PACKAGE_NAME, + "versions": {VERSION: "new reason"}, + } self.assertTrue(apply_yank_policy(listing, policy)) expected = copy.deepcopy(before) - expected["packages"][PACKAGE_NAME]["versions"][VERSION]["vrc-get"]["yanked"] = "new reason" + expected["packages"][PACKAGE_NAME]["versions"][VERSION]["vrc-get"]["yanked"] = ( + "new reason" + ) self.assertEqual(listing, expected) def test_unyank_removes_only_yanked_and_keeps_other_vrc_get_fields(self) -> None: from vpm_listing import apply_yank_policy listing = listing_with_versions(VERSION) - listing["packages"][PACKAGE_NAME]["versions"][VERSION]["vrc-get"][ - "yanked" - ] = "old reason" - self.assertTrue(apply_yank_policy(listing, {"schemaVersion": 1, "package": PACKAGE_NAME, "versions": {}})) + listing["packages"][PACKAGE_NAME]["versions"][VERSION]["vrc-get"]["yanked"] = ( + "old reason" + ) + self.assertTrue( + apply_yank_policy( + listing, {"schemaVersion": 1, "package": PACKAGE_NAME, "versions": {}} + ) + ) self.assertEqual( listing["packages"][PACKAGE_NAME]["versions"][VERSION]["vrc-get"], {"note": "retain me"}, @@ -398,23 +468,37 @@ def test_full_clear_removes_empty_vrc_get_object(self) -> None: from vpm_listing import apply_yank_policy listing = listing_with_versions(VERSION) - listing["packages"][PACKAGE_NAME]["versions"][VERSION]["vrc-get"] = {"yanked": "old reason"} + listing["packages"][PACKAGE_NAME]["versions"][VERSION]["vrc-get"] = { + "yanked": "old reason" + } - self.assertTrue(apply_yank_policy(listing, {"schemaVersion": 1, "package": PACKAGE_NAME, "versions": {}})) + self.assertTrue( + apply_yank_policy( + listing, {"schemaVersion": 1, "package": PACKAGE_NAME, "versions": {}} + ) + ) - self.assertNotIn("vrc-get", listing["packages"][PACKAGE_NAME]["versions"][VERSION]) + self.assertNotIn( + "vrc-get", listing["packages"][PACKAGE_NAME]["versions"][VERSION] + ) def test_rejects_non_object_vrc_get_without_mutating_listing(self) -> None: from vpm_listing import apply_yank_policy listing = listing_with_versions(VERSION) - listing["packages"][PACKAGE_NAME]["versions"][VERSION]["vrc-get"] = "not an object" + listing["packages"][PACKAGE_NAME]["versions"][VERSION]["vrc-get"] = ( + "not an object" + ) before = copy.deepcopy(listing) with self.assertRaises(UpdateError): apply_yank_policy( listing, - {"schemaVersion": 1, "package": PACKAGE_NAME, "versions": {VERSION: "reason"}}, + { + "schemaVersion": 1, + "package": PACKAGE_NAME, + "versions": {VERSION: "reason"}, + }, ) self.assertEqual(listing, before) @@ -439,7 +523,11 @@ def test_reapplying_identical_policy_is_a_no_op(self) -> None: from vpm_listing import apply_yank_policy listing = listing_with_versions(VERSION, "0.1.0") - policy = {"schemaVersion": 1, "package": PACKAGE_NAME, "versions": {VERSION: "reason"}} + policy = { + "schemaVersion": 1, + "package": PACKAGE_NAME, + "versions": {VERSION: "reason"}, + } self.assertTrue(apply_yank_policy(listing, policy)) before = copy.deepcopy(listing) @@ -448,7 +536,9 @@ def test_reapplying_identical_policy_is_a_no_op(self) -> None: class YankWorkflowContractTests(unittest.TestCase): - def test_receiver_workflow_runs_yank_contract_suite_on_relevant_changes(self) -> None: + def test_receiver_workflow_runs_yank_contract_suite_on_relevant_changes( + self, + ) -> None: workflow = (WORKFLOWS / "receiver-tests.yml").read_text(encoding="utf-8") self.assertIn(".github/scripts/**", workflow) @@ -459,7 +549,9 @@ def test_receiver_workflow_runs_yank_contract_suite_on_relevant_changes(self) -> def assert_workflow_uses_shared_update_serialization(self, filename: str) -> None: """Assert one named VPM writer serializes updates with the shared concurrency contract.""" workflow = (WORKFLOWS / filename).read_text(encoding="utf-8") - concurrency = "concurrency:\n group: vpm-repository-update\n cancel-in-progress: false" + concurrency = ( + "concurrency:\n group: vpm-repository-update\n cancel-in-progress: false" + ) self.assertIn(concurrency, workflow) @@ -474,16 +566,20 @@ def assert_mutating_workflow_is_fixed_to_master(self, filename: str) -> None: workflow = (WORKFLOWS / filename).read_text(encoding="utf-8") self.assertIn("ref: master", workflow) self.assertIn('git push origin "HEAD:master"', workflow) - self.assertIn('github.event.repository.default_branch != \'master\'', workflow) + self.assertIn("github.event.repository.default_branch != 'master'", workflow) self.assertIn("contents: write", workflow) - def test_update_workflow_is_fixed_to_master_and_rejects_other_defaults(self) -> None: + def test_update_workflow_is_fixed_to_master_and_rejects_other_defaults( + self, + ) -> None: self.assert_mutating_workflow_is_fixed_to_master("update-vpm.yml") def test_yank_workflow_is_fixed_to_master_and_rejects_other_defaults(self) -> None: self.assert_mutating_workflow_is_fixed_to_master("sync-vpm-yanks.yml") - def test_update_workflow_accepts_only_release_event_and_manual_policy_commit(self) -> None: + def test_update_workflow_accepts_only_release_event_and_manual_policy_commit( + self, + ) -> None: workflow = (WORKFLOWS / "update-vpm.yml").read_text(encoding="utf-8") self.assertIn("types: [update-vpm]", workflow) @@ -492,7 +588,9 @@ def test_update_workflow_accepts_only_release_event_and_manual_policy_commit(sel self.assertNotIn("policy_path", workflow) self.assertNotIn("policyPath", workflow) - def test_yank_workflow_accepts_only_yank_event_and_manual_fixed_source_inputs(self) -> None: + def test_yank_workflow_accepts_only_yank_event_and_manual_fixed_source_inputs( + self, + ) -> None: workflow = (WORKFLOWS / "sync-vpm-yanks.yml").read_text(encoding="utf-8") self.assertIn("types: [sync-vpm-yanks]", workflow) @@ -505,4 +603,4 @@ def test_yank_workflow_accepts_only_yank_event_and_manual_fixed_source_inputs(se if __name__ == "__main__": - unittest.main() \ No newline at end of file + unittest.main()