diff --git a/scripts/artifacts.py b/scripts/artifacts.py index 8bf26f18..c280e9cd 100644 --- a/scripts/artifacts.py +++ b/scripts/artifacts.py @@ -6,6 +6,7 @@ it is only meaningful if the comparison ignores the hour.""" from __future__ import annotations +from collections.abc import Callable, Iterator import contextlib import os import tempfile @@ -22,7 +23,9 @@ _TIMESTAMP_PATTERNS = [ re.compile(r'
[^<]*
'), ] -def write_if_changed(path: str, content: str, normalize=None) -> bool: +def write_if_changed( + path: str, content: str, normalize: Callable[[str], str] | None = None +) -> bool: """Write content to path only if the non-timestamp content differs. Compares new and existing content after stripping timestamp lines. @@ -86,7 +89,7 @@ def write_bytes_atomic(path: str, content: bytes) -> None: @contextlib.contextmanager -def file_lock(lock_path: str | os.PathLike, shared: bool = False): +def file_lock(lock_path: str | os.PathLike, shared: bool = False) -> Iterator[None]: """Hold a lock on lock_path, waiting for it if taken. Several sessions refresh the same caches: without it, two swaps of one @@ -140,7 +143,7 @@ HELD_LOCK_ENV = "RETROBIOS_HELD_ARTIFACT_DIR" @contextlib.contextmanager -def artifact_lock(directory: str, exclusive: bool = True): +def artifact_lock(directory: str, exclusive: bool = True) -> Iterator[None]: """Serialize access to a shared artifact directory across processes. Two pipeline runs building the same dist/ leave readers looking at @@ -175,7 +178,7 @@ def artifact_lock(directory: str, exclusive: bool = True): @contextlib.contextmanager -def hold_artifact_lock(directory: str): +def hold_artifact_lock(directory: str) -> Iterator[None]: """Hold a directory exclusively across several child steps. A lock taken step by step frees the directory between them: another diff --git a/scripts/common.py b/scripts/common.py index 02d306f6..4677c690 100644 --- a/scripts/common.py +++ b/scripts/common.py @@ -6,6 +6,7 @@ and file resolution - eliminates DRY violations across scripts. from __future__ import annotations +from collections.abc import Callable from collections import Counter import hashlib @@ -619,7 +620,7 @@ def declared_hash_verdict( md5_list: list, crc_candidates: set, zipped_file: str | None, - declared_size, + declared_size: int | list[int] | None, file_entry: dict | None = None, db: dict | None = None, ) -> str: @@ -669,7 +670,11 @@ def declared_hash_verdict( return "data_dir_hash_exact" -def _resolve_in_data_dirs(names_to_try, data_dir_registry, verdict): +def _resolve_in_data_dirs( + names_to_try: list[str], + data_dir_registry: dict, + verdict: Callable[[str], str], +) -> tuple[tuple[str, str] | None, str | None]: """Look for one of these names in the cached data directories. Returns the first candidate whose content satisfies the entry, and @@ -743,7 +748,9 @@ def forget_data_dir_index() -> None: _data_dir_index.clear() -def _resolve_agnostic(file_entry: dict, files_db: dict, has_strong_hash: bool): +def _resolve_agnostic( + file_entry: dict, files_db: dict, has_strong_hash: bool +) -> tuple[str, str] | None: """Any file of the right shape under the declared prefix. Some cores accept whatever filename sits in their BIOS directory, so the @@ -2380,7 +2387,7 @@ from dumpcatalog import ( # noqa: E402,F401 ) -def run_cli(main) -> None: +def run_cli(main: Callable[[], int | None]) -> None: """Run a command-line entry point to its exit status. A reader that stops early (`verify.py --all | head`) closed the pipe diff --git a/scripts/generate_pack.py b/scripts/generate_pack.py index 0990af11..b3d14433 100644 --- a/scripts/generate_pack.py +++ b/scripts/generate_pack.py @@ -12,6 +12,8 @@ and 3-tier storage (embedded/external/user_provided). from __future__ import annotations +from collections.abc import Callable, Iterator + from artifacts import _build_timestamp import argparse import contextlib @@ -331,7 +333,7 @@ def _data_directory_members( seen_destinations: set, seen_lower: set, seen_parents: set, -): +) -> Iterator[tuple[str, str]]: """Yield (source, destination) for the cached data directories a platform's systems declare. @@ -386,7 +388,7 @@ def _data_directory_members( def _pack_data_directories( - zf, + zf: zipfile.ZipFile, pack_systems: dict, data_registry: dict | None, platform_name: str, @@ -458,10 +460,10 @@ def _select_variants( emulators_dir: str, db: dict, base_dest: str, - emu_profiles, - target_cores, + emu_profiles: dict | None, + target_cores: set[str] | None, source: str, - regions, + regions: list[str] | None, one_per_slot: bool, required_only: bool = False, ) -> tuple[set, list, list]: @@ -1129,7 +1131,7 @@ def generate_pack( def _extract_zip_to_archive( source_zip: str, dest_prefix: str, target_zf: zipfile.ZipFile -): +) -> None: """Extract contents of a source ZIP into target ZIP under dest_prefix.""" with zipfile.ZipFile(source_zip, "r") as src: for info in src.infolist(): @@ -1153,7 +1155,7 @@ def _add_zip_to_pack( dest_path: str, target_zf: zipfile.ZipFile, file_entry: dict | None = None, -): +) -> None: """Write a source ZIP into the pack. A declared outer hash (md5/sha1/crc32) means the platform or profile @@ -1179,7 +1181,7 @@ def _add_zip_to_pack( def _normalize_zip_for_pack( source_zip: str, dest_path: str, target_zf: zipfile.ZipFile -): +) -> None: """Add a MAME BIOS ZIP to the pack as a deterministic rebuild. Instead of copying the original ZIP (with non-deterministic metadata), @@ -2151,7 +2153,7 @@ def generate_target_manifests(targets_dir: str, output_dir: str) -> None: print(f"Generated {count} target manifest(s) in {output_dir}") -def _validate_args(args, parser): +def _validate_args(args: argparse.Namespace, parser: argparse.ArgumentParser) -> None: """Validate argument combinations and mutual exclusion rules.""" has_platform = bool(args.platform) has_all = args.all @@ -2251,8 +2253,13 @@ def _write_manifest_if_changed(path: str, manifest: dict) -> None: def _run_manifest_mode( - args, groups, db, zip_contents, emu_profiles, target_cores_cache -): + args: argparse.Namespace, + groups: list[tuple[list[str], str]], + db: dict, + zip_contents: dict, + emu_profiles: dict, + target_cores_cache: dict[str, set[str] | None], +) -> None: """Generate JSON manifests instead of ZIP packs.""" registry_path = os.path.join(args.platforms_dir, "_registry.yml") os.makedirs(args.output_dir, exist_ok=True) @@ -2350,7 +2357,7 @@ def _run_manifest_mode( @contextlib.contextmanager -def _pack_output_lock(output_dir: str, exclusive: bool = True): +def _pack_output_lock(output_dir: str, exclusive: bool = True) -> Iterator[None]: """Hold the output directory for the duration of a pack run.""" try: with artifact_lock(output_dir, exclusive=exclusive): @@ -2360,7 +2367,7 @@ def _pack_output_lock(output_dir: str, exclusive: bool = True): sys.exit(1) -def _run_verify_packs(args): +def _run_verify_packs(args: argparse.Namespace) -> None: """Verify each pack in the output directory against its platform. Delegates to verify_pack_against_platform so the command answers the same @@ -2445,7 +2452,7 @@ def _pack_label(source: str, required_only: bool) -> str: def _truth_layer_is_empty( - platform: str, platforms_dir: str, emu_profiles: dict, target_cores + platform: str, platforms_dir: str, emu_profiles: dict, target_cores: set[str] | None ) -> bool: """Whether a platform runs no core a profile describes. @@ -2456,8 +2463,14 @@ def _truth_layer_is_empty( return not _platform_cores(config, emu_profiles or {}, target_cores=target_cores) -def _skip_empty_truth(args, source: str, representative: str, emu_profiles, tc, - failed: list[str]) -> bool: +def _skip_empty_truth( + args: argparse.Namespace, + source: str, + representative: str, + emu_profiles: dict, + tc: set[str] | None, + failed: list[str], +) -> bool: """Leave out a truth pack that would be empty; refuse it when asked for.""" if source != "truth" or not _truth_layer_is_empty( representative, args.platforms_dir, emu_profiles, tc @@ -2475,21 +2488,21 @@ def _skip_empty_truth(args, source: str, representative: str, emu_profiles, tc, return True -def _produced(split: bool, zip_path, zip_paths) -> bool: +def _produced(split: bool, zip_path: str | None, zip_paths: list[str] | None) -> bool: """Whether the build wrote what the run asked for.""" return bool(zip_paths if split else zip_path) def _run_platform_packs( - args, - groups, - db, - zip_contents, - data_registry, - emu_profiles, - target_cores_cache, - system_filter, -): + args: argparse.Namespace, + groups: list[tuple[list[str], str]], + db: dict, + zip_contents: dict, + data_registry: dict | None, + emu_profiles: dict, + target_cores_cache: dict[str, set[str] | None], + system_filter: list[str] | None, +) -> None: """Generate ZIP packs for platform groups and verify.""" if args.all_variants: variants = [ @@ -2632,7 +2645,7 @@ def _run_platform_packs( sys.exit(1) -def _refuse_for_all_variants(args, parser) -> None: +def _refuse_for_all_variants(args: argparse.Namespace, parser: argparse.ArgumentParser) -> None: """Flags --all-variants would override, or modes that never read it.""" if not args.all_variants: return @@ -2662,7 +2675,7 @@ def _refuse_for_all_variants(args, parser) -> None: parser.error(f"{flag} is incompatible with --verify-packs") -def _refuse_refresh_data(args, parser) -> None: +def _refuse_refresh_data(args: argparse.Namespace, parser: argparse.ArgumentParser) -> None: """Only the platform pack path refreshes the data directories. Elsewhere the cache on disk was packed as it stood, and --offline wins @@ -2684,14 +2697,16 @@ def _refuse_refresh_data(args, parser) -> None: parser.error(f"{name} is incompatible with {flag}") -def _refuse_include_archived(args, parser) -> None: +def _refuse_include_archived( + args: argparse.Namespace, parser: argparse.ArgumentParser +) -> None: """Only --all chooses among registered platforms; every other mode names its own platform, emulator, system or hash.""" if args.include_archived and not args.all: parser.error("--include-archived requires --all") -def _refuse_unapplied_flags(args, parser) -> None: +def _refuse_unapplied_flags(args: argparse.Namespace, parser: argparse.ArgumentParser) -> None: """Refuse every flag the requested mode would not apply. A mode applies a narrowing flag or refuses it: a swallowed flag answers @@ -2749,7 +2764,7 @@ QUICK_MODE_FLAGS: dict[str, set[str]] = { } -def _given_flags(args) -> set[str]: +def _given_flags(args: argparse.Namespace) -> set[str]: """Flags that select or narrow what a run produces, as typed.""" flags = { "--platform": args.platform, @@ -2787,7 +2802,7 @@ def _given_flags(args) -> set[str]: _MISSING_DATA_DIRS: set[str] = set() -def main(): +def main() -> None: with contextlib.ExitStack() as held: _main(held) if _MISSING_DATA_DIRS: @@ -3281,8 +3296,8 @@ def _attach_claims( claims: list[dict], db: dict, bios_dir: str, - zip_contents, - offline, + zip_contents: dict | None, + offline: bool | None, required_only: bool, validation_index: dict | None, owners: dict | None, @@ -3353,7 +3368,7 @@ def _record_slot_choices( entry: dict, choices: list[tuple[str, str | None]], declarations: list[tuple[list[str], dict, str]], - settle, + settle: Callable[[dict, str, str, str | None], tuple[str | None, str]], dest: str, db: dict, repo_root: str, @@ -3416,8 +3431,8 @@ def _manifest_core_entries( bios_dir: str, base_dest: str, repo_root: str, - zip_contents, - offline, + zip_contents: dict | None, + offline: bool | None, region_drops: set, case_insensitive: bool, seen_destinations: set, @@ -3425,7 +3440,7 @@ def _manifest_core_entries( seen_parents: set, manifest_files: list, omitted_by_destination: dict, - record_omission, + record_omission: Callable[..., None], pack_only_sizes: list[int], selection: SelectionIndex, required_only: bool = False, @@ -3612,7 +3627,7 @@ def _record_absent( required_only: bool, seen_destinations: set, omitted_by_destination: dict, - record_omission, + record_omission: Callable[..., None], ) -> None: """A file a core needs that the collection does not hold is an omission. @@ -3641,7 +3656,7 @@ def _record_absent( ) -def _record_unplaceable(unplaceable: list[dict], record_omission) -> None: +def _record_unplaceable(unplaceable: list[dict], record_omission: Callable[..., None]) -> None: """Held by the collection, nowhere to put it on this platform. Said in the manifest, where RomM's DOSBox Pure ROMs were simply absent. @@ -3831,7 +3846,9 @@ def generate_manifest( del remaining[sid] return order - def settle(file_entry: dict, sys_id: str, dest: str, override: str | None): + def settle( + file_entry: dict, sys_id: str, dest: str, override: str | None + ) -> tuple[str | None, str]: """The file served once the slot is settled, or None and why not.""" local_path, status = slots.apply_override(override, *resolve_file( file_entry, @@ -4056,7 +4073,7 @@ def _manifest_core_index( config: dict, profiles: dict[str, dict], files: list[dict], - omitted, + omitted: list[dict], ) -> tuple[dict[str, list[str]], dict[str, list[str]]]: """What the installer needs to narrow by a list of core names. diff --git a/scripts/validation.py b/scripts/validation.py index 259fdcfd..e80eeba3 100644 --- a/scripts/validation.py +++ b/scripts/validation.py @@ -9,7 +9,7 @@ from __future__ import annotations import os import zipfile -from collections.abc import Mapping +from collections.abc import Iterator, Mapping from common import ( adler32_of, @@ -574,7 +574,7 @@ def find_validated_variant( current_real = os.path.realpath(current_path) seen: set[str] = set() - def candidates(): + def candidates() -> Iterator[str]: # Sorted: the index holds sets, whose order follows the per-process # string hash, and the first passing candidate is the one shipped. expected = validation_index[fname] @@ -789,10 +789,10 @@ def validated_choice( # The cores that read this destination judge it, each by its own # entries. judges = owning & set(rules["emulators"]) - def check(path: str, strict: bool): + def check(path: str, strict: bool) -> tuple[str, list[str]] | None: return check_file_validation(path, name, validation_index, bios_dir, judges, strict) - def variant(strict: bool): + def variant(strict: bool) -> str | None: return find_validated_variant( file_entry, db, local_path, validation_index, bios_dir, platform_digest=platform_digest, emulators=judges, strict=strict, diff --git a/tests/test_audit_regressions.py b/tests/test_audit_regressions.py index 7d61ac49..4cba5f43 100644 --- a/tests/test_audit_regressions.py +++ b/tests/test_audit_regressions.py @@ -3747,5 +3747,38 @@ class TargetListsReadThePinnedRelease(unittest.TestCase): self.assertNotIn("fetch_github_latest_version(", source) + +class SignaturesCarryTheirTypes(unittest.TestCase): + """Type hints on every signature is a written rule; 177 had none, the + shared API among them (artifact locks, the data-dir resolver, the region + priority list of _select_variants).""" + + # Lowered as modules are annotated; never raised. + CEILING = 139 + COMPLETE = ("artifacts.py", "common.py", "validation.py", "generate_pack.py") + + @staticmethod + def _missing(path: Path) -> list[str]: + found = [] + for node in ast.walk(ast.parse(path.read_text(encoding="utf-8"))): + if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + continue + args = node.args + params = [a for a in (*args.posonlyargs, *args.args, *args.kwonlyargs) + if a.arg not in ("self", "cls")] + params += [a for a in (args.vararg, args.kwarg) if a] + if node.returns is None or any(a.annotation is None for a in params): + found.append(f"{path.name}:{node.lineno} {node.name}") + return found + + def test_the_count_only_goes_down(self): + paths = [ROOT / "install.py", *sorted((ROOT / "scripts").rglob("*.py"))] + missing = [m for path in paths for m in self._missing(path)] + self.assertLessEqual(len(missing), self.CEILING, missing[:20]) + for name in self.COMPLETE: + with self.subTest(module=name): + self.assertEqual(self._missing(ROOT / "scripts" / name), []) + + if __name__ == "__main__": unittest.main()