diff --git a/pixi.toml b/pixi.toml index b612f50..786ce49 100644 --- a/pixi.toml +++ b/pixi.toml @@ -20,4 +20,4 @@ vinca.path = "." [package.build] backend.name = "pixi-build-python" -config.ignore-pypi-mapping = false \ No newline at end of file +config.ignore-pypi-mapping = false diff --git a/pyproject.toml b/pyproject.toml index 49c523b..e464fb8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -67,5 +67,6 @@ include = [ packages = ["vinca"] [tool.pyrefly] -preset = "basic" +preset = "default" project-includes = ["vinca"] +project-excludes = ["vinca/test_*.py"] diff --git a/vinca/config.py b/vinca/config.py index 6210d53..d998bc9 100644 --- a/vinca/config.py +++ b/vinca/config.py @@ -1,6 +1,8 @@ -selected_platform = None -ros_distro = None -skip_testing = None -parsed_args = None -setup_pixi_version = None -pixi_version = None +from argparse import Namespace + +selected_platform: str | None = None +ros_distro: str | None = None +skip_testing: bool | None = None +parsed_args: Namespace | None = None +setup_pixi_version: str | None = None +pixi_version: str | None = None diff --git a/vinca/distro.py b/vinca/distro.py index 8cbce7b..1f5158f 100644 --- a/vinca/distro.py +++ b/vinca/distro.py @@ -237,9 +237,11 @@ def _get_direct_depends(self, pkg: str) -> set[str]: self._direct_depends_cache[pkg] = set(direct) return direct - def _get_snapshot_recursive_depends(self, pkg, ignore_pkgs=None): + def _get_snapshot_recursive_depends( + self, pkg: str, ignore_pkgs: Optional[Iterable[str]] = None + ) -> set[str]: """Return ROS dependencies using only package manifests pinned by the snapshot.""" - dependencies = set() + dependencies: set[str] = set() ignored = set(ignore_pkgs or []) packages_to_check = {pkg} checked_packages = set() @@ -263,7 +265,7 @@ def _get_snapshot_recursive_depends(self, pkg, ignore_pkgs=None): package_xml = self.get_release_package_xml(package_name) package = catkin_pkg.package.parse_package_string(package_xml) package.evaluate_conditions(os.environ) - direct_dependencies = { + direct_dependencies: set[str] = { dependency.name for attribute in dependency_attributes for dependency in getattr(package, attribute) diff --git a/vinca/generate_gha.py b/vinca/generate_gha.py index 5c1e415..4dc60c7 100644 --- a/vinca/generate_gha.py +++ b/vinca/generate_gha.py @@ -97,10 +97,8 @@ def add_additional_recipes(args): os.path.join(args.dir, "..", "additional_recipes") ) - print("Searching additional recipes in ", additional_recipes_path) - if not os.path.exists(additional_recipes_path): - return + return [] with open("vinca.yaml", "r") as vinca_yaml: vinca_conf = yaml.safe_load(vinca_yaml) @@ -222,7 +220,7 @@ def build_unix_pipeline( setup_pixi_version: str = DEFAULT_SETUP_PIXI_VERSION, pixi_version: str = DEFAULT_PIXI_VERSION, ): - blurb = {"jobs": {}, "name": pipeline_name} + blurb: dict[str, Any] = {"jobs": {}, "name": pipeline_name} if workflow is None: workflow = blurb @@ -343,7 +341,7 @@ def build_win_pipeline( ): vm_imagename = "windows-2022" # Build Win pipeline - blurb = {"jobs": {}, "name": "build_win"} + blurb: dict[str, Any] = {"jobs": {}, "name": "build_win"} if workflow is None: workflow = blurb @@ -423,7 +421,10 @@ def build_win_pipeline( def get_full_tree(): - recipes_dir = config.parsed_args.dir + parsed_args = config.parsed_args + if parsed_args is None: + raise RuntimeError("Pipeline arguments must be parsed before generating a tree") + recipes_dir = parsed_args.dir vinca_yaml = os.path.join(os.path.dirname(recipes_dir), "vinca.yaml") @@ -499,7 +500,11 @@ def main(): names_to_build = {pkg["package"]["name"] for pkg in metas} print("Names to build: ", names_to_build) - tg_slimmed = [el for el in tg if el in names_to_build] + tg_slimmed = [ + element + for element in tg + if isinstance(element, str) and element in names_to_build + ] stages = [] current_stage = [] diff --git a/vinca/generate_gitlab.py b/vinca/generate_gitlab.py index 0410c26..b92f0a4 100644 --- a/vinca/generate_gitlab.py +++ b/vinca/generate_gitlab.py @@ -1,6 +1,7 @@ import glob import os import sys +from typing import Any import networkx as nx import yaml @@ -79,7 +80,7 @@ def main(): print(stages) - gitlab_template = {"image": "condaforge/linux-anvil-cos7-x86_64"} + gitlab_template: dict[str, Any] = {"image": "condaforge/linux-anvil-cos7-x86_64"} stage_names = [] for i, s in enumerate(stages): diff --git a/vinca/main.py b/vinca/main.py index 51e16de..87fb9f2 100644 --- a/vinca/main.py +++ b/vinca/main.py @@ -263,6 +263,8 @@ def get_pkg(pkg_name): print(f"Could not generate output for {pkg_shortname}") continue + output: dict[str, Any] | None = None + try: output = generate_output( pkg_shortname, @@ -692,7 +694,7 @@ def main(): ): with open(add_rec) as fi: add_rec_y = yaml.load(fi) - if config.parsed_args.platform == "emscripten-wasm32": + if arguments.platform == "emscripten-wasm32": additional_recipe_names.add(add_rec_y["package"]["name"]) else: if add_rec_y["package"]["name"] not in [ diff --git a/vinca/migrate.py b/vinca/migrate.py index 00c2b62..71f5b29 100644 --- a/vinca/migrate.py +++ b/vinca/migrate.py @@ -78,22 +78,24 @@ def create_migration_instructions(arch, packages_to_migrate, trigger_branch): latest = {} for pkg in ros_pkgs: - current = current_version = None + current = None + current_version: tuple[int, ...] | None = None for pkey in packages: if packages[pkey]["name"] == pkg: - tmp = packages[pkey]["version"].split(".") + parts = packages[pkey]["version"].split(".") version = [] - for el in tmp: - if el.isdecimal(): - version.append(int(el)) + for element in parts: + if element.isdecimal(): + version.append(int(element)) else: - x = re.search(r"[^0-9]", version).start() - version.append(int(el[:x])) + match = re.search(r"[^0-9]", element) + if match is not None: + version.append(int(element[: match.start()])) - version = tuple(version) + parsed_version = tuple(version) - if not current or version > current_version: - current_version = version + if current_version is None or parsed_version > current_version: + current_version = parsed_version current = pkey latest[pkg] = current @@ -132,22 +134,24 @@ def create_migration_instructions(arch, packages_to_migrate, trigger_branch): if os.path.exists("recipes"): shutil.rmtree("recipes") - mutex_path = os.path.join( - config.parsed_args.dir, "additional_recipes/ros-distro-mutex" - ) - if os.path.exists(mutex_path): - goal_folder = os.path.join( - config.parsed_args.dir, "recipes", "ros-distro-mutex" + parsed_args = config.parsed_args + if parsed_args is None: + raise RuntimeError( + "Migration arguments must be parsed before generating instructions" ) + + mutex_path = os.path.join(parsed_args.dir, "additional_recipes/ros-distro-mutex") + if os.path.exists(mutex_path): + goal_folder = os.path.join(parsed_args.dir, "recipes", "ros-distro-mutex") os.makedirs(goal_folder, exist_ok=True) copy_tree(mutex_path, goal_folder) subprocess.check_call( - ["vinca", "-d", config.parsed_args.dir, "--multiple", "--platform", arch] + ["vinca", "-d", parsed_args.dir, "--multiple", "--platform", arch] ) # TODO remove hard coded build branch here! - recipe_dir = os.path.join(config.parsed_args.dir, "recipes") + recipe_dir = os.path.join(parsed_args.dir, "recipes") subprocess.check_call( [ "vinca-gha", diff --git a/vinca/mutex.py b/vinca/mutex.py index afacab2..b69a76f 100644 --- a/vinca/mutex.py +++ b/vinca/mutex.py @@ -59,6 +59,8 @@ def get_mutex_package_dependency( raise ValueError( f"Error parsing mutex_package configuration: {error}" ) from error + if mutex_config is None: + return None version_parts = mutex_config["version"].split(".") pin_depth = len(mutex_config["upper_bound"].split(".")) diff --git a/vinca/pinning.py b/vinca/pinning.py index 3719976..19313f4 100644 --- a/vinca/pinning.py +++ b/vinca/pinning.py @@ -19,7 +19,7 @@ from concurrent.futures import ThreadPoolExecutor, as_completed from contextlib import contextmanager from pathlib import Path -from typing import Any, Iterator, Mapping, Optional, Sequence, Union +from typing import Any, Iterator, Literal, Mapping, Optional, Sequence, Union, cast from urllib.parse import quote import requests @@ -127,7 +127,9 @@ def get_pinning_distribution(version: str) -> str: return url -def _read_tar_members(fileobj: Any, *, mode: str) -> dict[str, bytes]: +def _read_tar_members( + fileobj: Any, *, mode: Literal["r|", "r:bz2"] +) -> dict[str, bytes]: """Extract only the base config and migration YAML files from a package tarball.""" files = {} with tarfile.open(fileobj=fileobj, mode=mode) as archive: @@ -290,7 +292,7 @@ def render_pinning( else: base_payload, migration_payloads = package - rendered = yaml.load(base_payload.decode("utf-8")) or {} + rendered: Any = yaml.load(base_payload.decode("utf-8")) or {} migration_names = {_migration_name(name): name for name in selected_migrations} missing = sorted(set(migration_names) - set(migration_payloads)) if missing: @@ -305,7 +307,7 @@ def render_pinning( for migration in ordered_migrations: migration_config = yaml.load(migration_payloads[migration].decode("utf-8")) try: - rendered = variant_add(rendered, migration_config or {}) + rendered = cast(Any, variant_add(rendered, migration_config or {})) except (KeyError, RuntimeError, TypeError, ValueError) as exc: raise PinningError(f"Could not apply migration {migration}: {exc}") from exc _validate_zipped_overrides(rendered, overrides) @@ -416,6 +418,8 @@ def dependencies_from_vinca( raise PinningError( "Platform selectors must not change the ROS distro snapshots" ) + if group_packages is None: + raise PinningError("No group packages were generated") vinca_config["_selected_pkgs"] = get_selected_packages( distro, vinca_config ) @@ -501,8 +505,9 @@ def _migration_selectors(data: Any) -> Iterator[Optional[str]]: key_comment = data.ca.items.get(key, [None, None, None])[2] key_selector = _comment_selector(key_comment) if isinstance(value, list): + value_with_comments: Any = value for index in range(len(value)): - item_comment = value.ca.items.get(index, [None])[0] + item_comment = value_with_comments.ca.items.get(index, [None])[0] item_selector = _comment_selector(item_comment) if key_selector and item_selector: yield f"({key_selector}) and ({item_selector})" @@ -525,6 +530,7 @@ def _migration_applies_to_platforms(payload: bytes, platforms: Sequence[str]) -> return any( _eval_condition(selector, _platform_flags(platform)) for selector in selectors + if selector is not None for platform in platforms ) diff --git a/vinca/resolve.py b/vinca/resolve.py index e0a36a0..c4317dd 100644 --- a/vinca/resolve.py +++ b/vinca/resolve.py @@ -32,10 +32,15 @@ def get_conda_index(vinca_conf, base_dir): def resolve_pkgname_from_indexes(pkg_shortname, conda_index): for i in conda_index: if pkg_shortname in i: - sys_platform = map_platform_python_to_conda[config.selected_platform] + selected_platform = config.selected_platform + if selected_platform is None: + raise RuntimeError( + "A target platform is required to resolve package names" + ) + sys_platform = map_platform_python_to_conda[selected_platform] if "robostack" in i[pkg_shortname].keys(): - if config.selected_platform in i[pkg_shortname]["robostack"]: - return i[pkg_shortname]["robostack"][config.selected_platform] + if selected_platform in i[pkg_shortname]["robostack"]: + return i[pkg_shortname]["robostack"][selected_platform] elif sys_platform in i[pkg_shortname]["robostack"]: return i[pkg_shortname]["robostack"][sys_platform] else: diff --git a/vinca/snapshot.py b/vinca/snapshot.py index cdf10b3..014ea0d 100644 --- a/vinca/snapshot.py +++ b/vinca/snapshot.py @@ -60,8 +60,10 @@ def main(): deps = distro.get_depends(args.package) deps.add(args.package) + max_len = 0 + if not args.quiet: - max_len = max([len(dep) for dep in deps]) + max_len = max(len(dep) for dep in deps) print("\033[1m{0:{2}} {1}\033[0m".format("Package", "Version", max_len + 2)) output = {} diff --git a/vinca/sources.py b/vinca/sources.py index 0022d85..04014e8 100644 --- a/vinca/sources.py +++ b/vinca/sources.py @@ -89,7 +89,7 @@ def generate_source( url, ref, ref_type = distro.get_released_repo(shortname) package_name = package_names[0] - entry = source_reference(url=url, ref=ref, ref_type=ref_type) + entry: dict[str, Any] = source_reference(url=url, ref=ref, ref_type=ref_type) entry["target_directory"] = f"{package_name}/src/work" patches = _package_patches(package_name, vinca_conf, platform) @@ -127,7 +127,7 @@ def generate_source_version( url, ref, ref_type = distro.get_released_repo(shortname) package_name = package_names[0] - entry = source_reference(url=url, ref=ref, ref_type=ref_type) + entry: dict[str, Any] = source_reference(url=url, ref=ref, ref_type=ref_type) entry["target_directory"] = f"{package_name}/src/work" if patches := _package_patches(package_name, vinca_conf, platform): entry["patches"] = patches @@ -150,7 +150,7 @@ def generate_fat_source( continue url, ref, ref_type = distro.get_released_repo(shortname) package_name = package_names[0] - entry = source_reference(url=url, ref=ref, ref_type=ref_type) + entry: dict[str, Any] = source_reference(url=url, ref=ref, ref_type=ref_type) entry["target_directory"] = f"src/{package_name}" patch_path = Path(vinca_conf["_patch_dir"]) / f"{package_name}.patch" diff --git a/vinca/test_archive_sources.py b/vinca/test_archive_sources.py index 645e39a..b938945 100644 --- a/vinca/test_archive_sources.py +++ b/vinca/test_archive_sources.py @@ -1,6 +1,7 @@ import io import tarfile import zipfile +from typing import Any import pytest import requests @@ -216,7 +217,7 @@ def test_get_forces_no_credentials_of_its_own(monkeypatch): """ distro = Distro.__new__(Distro) monkeypatch.delenv("GITHUB_TOKEN", raising=False) - calls = {} + calls: dict[str, Any] = {} class FakeResponse: content = b"payload" diff --git a/vinca/test_distro_dependencies.py b/vinca/test_distro_dependencies.py index d24f9b0..7961f7d 100644 --- a/vinca/test_distro_dependencies.py +++ b/vinca/test_distro_dependencies.py @@ -1,3 +1,4 @@ +from typing import Any from unittest.mock import Mock from vinca.distro import Distro @@ -11,7 +12,7 @@ def test_dependency_walk_reuses_direct_dependencies_across_roots(): "app-leaf": set(), "common-leaf": set(), } - distro = Distro.__new__(Distro) + distro: Any = Distro.__new__(Distro) distro._depends_cache = {} distro._direct_depends_cache = {} distro.snapshot = None @@ -39,7 +40,7 @@ def test_dependency_walk_honors_ignored_packages(): "ignored": {"hidden"}, "hidden": set(), } - distro = Distro.__new__(Distro) + distro: Any = Distro.__new__(Distro) distro._depends_cache = {} distro._direct_depends_cache = {} distro.snapshot = None @@ -57,7 +58,7 @@ def test_dependency_walk_honors_ignored_packages(): def test_dependency_walk_excludes_root_in_cycles(): graph = {"a": {"b"}, "b": {"a"}} - distro = Distro.__new__(Distro) + distro: Any = Distro.__new__(Distro) distro._depends_cache = {} distro._direct_depends_cache = {} distro.snapshot = None @@ -69,7 +70,7 @@ def test_dependency_walk_excludes_root_in_cycles(): def test_dependency_walk_excludes_root_from_self_dependency(): - distro = Distro.__new__(Distro) + distro: Any = Distro.__new__(Distro) distro._depends_cache = {} distro._direct_depends_cache = {} distro.snapshot = None @@ -82,7 +83,7 @@ def test_dependency_walk_excludes_root_from_self_dependency(): def test_dependency_walk_excludes_root_from_longer_cycle(): graph = {"a": {"b"}, "b": {"c"}, "c": {"a"}} - distro = Distro.__new__(Distro) + distro: Any = Distro.__new__(Distro) distro._depends_cache = {} distro._direct_depends_cache = {} distro.snapshot = None diff --git a/vinca/test_naming_scheme.py b/vinca/test_naming_scheme.py index 3cce89b..00d925b 100644 --- a/vinca/test_naming_scheme.py +++ b/vinca/test_naming_scheme.py @@ -1,5 +1,7 @@ """Tests for the ROS package naming scheme.""" +from typing import Any + import pytest from vinca.distro import Distro @@ -134,7 +136,7 @@ def test_package_name_mode_defaults_to_legacy(): ], ) def test_dependency_resolution_uses_selected_name_mode(mode, expected_name): - distro = make_distro() + distro: Any = make_distro() distro.check_package = lambda _name: True vinca_conf = {"_conda_indexes": [], "package_name_mode": mode} diff --git a/vinca/test_provenance.py b/vinca/test_provenance.py index 93801e4..ef6f4ed 100644 --- a/vinca/test_provenance.py +++ b/vinca/test_provenance.py @@ -4,6 +4,8 @@ directly by the config and which already-selected packages depend on it. """ +from typing import Any + import vinca.main as m @@ -32,7 +34,10 @@ def test_provenance_requested_and_required_by(): {"app": {"libA", "libB"}, "libA": {"libcommon"}, "libB": {"libcommon"}}, ros1=True, # keep ROS2 auto-injection out of this assertion ) - conf = {"packages_select_by_deps": ["app", "libA"], "packages_skip_by_deps": None} + conf: dict[str, Any] = { + "packages_select_by_deps": ["app", "libA"], + "packages_skip_by_deps": None, + } selected = m.get_selected_packages(distro, conf) @@ -52,7 +57,7 @@ def test_provenance_requested_and_required_by(): def test_provenance_skip_by_deps_excludes_package(): distro = FakeDistro({"app": {"libA", "skipme"}}, ros1=True) - conf = { + conf: dict[str, Any] = { "packages_select_by_deps": ["app"], "packages_skip_by_deps": ["skipme"], } @@ -65,7 +70,10 @@ def test_provenance_skip_by_deps_excludes_package(): def test_ros2_workspace_auto_injected_with_reason(): distro = FakeDistro({"app": set()}, ros1=False) - conf = {"packages_select_by_deps": ["app"], "packages_skip_by_deps": None} + conf: dict[str, Any] = { + "packages_select_by_deps": ["app"], + "packages_skip_by_deps": None, + } selected = m.get_selected_packages(distro, conf) @@ -80,7 +88,10 @@ def test_generation_summary_output(monkeypatch, capsys): {"app": {"libA", "libB"}, "libA": {"libcommon"}, "libB": {"libcommon"}}, ros1=True, ) - conf = {"packages_select_by_deps": ["app", "libA"], "packages_skip_by_deps": None} + conf: dict[str, Any] = { + "packages_select_by_deps": ["app", "libA"], + "packages_skip_by_deps": None, + } conf["_selected_pkgs"] = m.get_selected_packages(distro, conf) monkeypatch.setattr( diff --git a/vinca/test_snapshot_metadata.py b/vinca/test_snapshot_metadata.py index 0a57dd8..96ba1e1 100644 --- a/vinca/test_snapshot_metadata.py +++ b/vinca/test_snapshot_metadata.py @@ -1,3 +1,4 @@ +from typing import Any from unittest.mock import Mock, patch import vinca.main as main @@ -74,7 +75,7 @@ def make_snapshot_distro(monkeypatch): def test_snapshot_package_xml_and_dependencies_do_not_follow_live_rosdistro( monkeypatch, ): - distro = make_snapshot_distro(monkeypatch) + distro: Any = make_snapshot_distro(monkeypatch) package_xml_content = distro.get_release_package_xml("snapshot_package") @@ -84,6 +85,8 @@ def test_snapshot_package_xml_and_dependencies_do_not_follow_live_rosdistro( "tag", ) assert distro.get_version("snapshot_package") == "1.0.0" + assert package_xml_content is not None + assert "1.0.0" in package_xml_content assert "snapshot_dependency" in package_xml_content assert "live_dependency" not in package_xml_content @@ -143,7 +146,7 @@ def test_snapshot_package_xml_does_not_use_live_cache_after_snapshot_change( def test_snapshot_metadata_generates_dependency_required_by_pinned_source( monkeypatch, ): - distro = make_snapshot_distro(monkeypatch) + distro: Any = make_snapshot_distro(monkeypatch) dependency_names = { "snapshot_package": "ros2-snapshot-package", "python": "python", @@ -168,6 +171,7 @@ def test_snapshot_metadata_generates_dependency_required_by_pinned_source( distro, distro.get_version("snapshot_package"), ) + assert output is not None assert output["package"] == { "name": "ros2-snapshot-package", @@ -178,7 +182,7 @@ def test_snapshot_metadata_generates_dependency_required_by_pinned_source( def test_snapshot_is_authoritative_for_package_membership(monkeypatch): - distro = make_snapshot_distro(monkeypatch) + distro: Any = make_snapshot_distro(monkeypatch) distro._distro.release_packages = {"live_only": Mock()} assert distro.check_package("snapshot_package") diff --git a/vinca/v1_selectors.py b/vinca/v1_selectors.py index 95539fe..49720c7 100644 --- a/vinca/v1_selectors.py +++ b/vinca/v1_selectors.py @@ -117,26 +117,26 @@ def _process_node(node: Any, ctx: Mapping[str, Any]) -> Any | None: # Recursion – dict --------------------------------------------------------- if isinstance(node, MutableMapping): - out: dict[str, Any] = {} + mapping_out: dict[str, Any] = {} for k, v in node.items(): new = _process_node(v, ctx) if new is not None: - out[k] = new - return out + mapping_out[k] = new + return mapping_out # Recursion – list --------------------------------------------------------- if isinstance(node, Sequence) and not isinstance(node, (str, bytes)): - out: list[Any] = [] + sequence_out: list[Any] = [] for item in node: new = _process_node(item, ctx) if new is None: continue # Flatten lists returned from selectors so the example in docs works if isinstance(new, list): - out.extend(new) + sequence_out.extend(new) else: - out.append(new) - return out + sequence_out.append(new) + return sequence_out # Scalar – nothing to do --------------------------------------------------- return node diff --git a/vinca/variant_algebra.py b/vinca/variant_algebra.py index 1c2321c..f7437ea 100644 --- a/vinca/variant_algebra.py +++ b/vinca/variant_algebra.py @@ -89,15 +89,16 @@ def _selector_platforms(selector: Optional[str]) -> set[str]: """Conservatively map a selector expression to its affected build platforms.""" if selector is None: return set(_PLATFORMS) + selector_expression = selector platforms = set(_PLATFORMS) - tokens = set(re.findall(r"[A-Za-z0-9_]+", selector)) + tokens = set(re.findall(r"[A-Za-z0-9_]+", selector_expression)) if "unix" in tokens: platforms &= {item for item in _PLATFORMS if not item.startswith("win-")} os_scopes = [] def is_positive_os(*names: Any) -> Any: return any(name in tokens for name in names) and not any( - re.search(rf"\bnot\s+{name}\b", selector) for name in names + re.search(rf"\bnot\s+{name}\b", selector_expression) for name in names ) if is_positive_os("linux", "linux64"):