Use snapshotted package metadata

vinca/distro.py

@@ -5,6 +5,7 @@
 import urllib.parse
 import zipfile
 
+import catkin_pkg.package
 import requests
 from rosdistro import get_cached_distribution, get_index, get_index_url
 from rosdistro.dependency_walker import DependencyWalker
@@ -151,6 +152,11 @@ def get_depends(self, pkg, ignore_pkgs=None):
             print(f"{pkg} not in available packages anymore")
             return dependencies
 
+        if self.snapshot:
+            dependencies = self._get_snapshot_recursive_depends(pkg, ignore_pkgs)
+            self._depends_cache[cache_key] = set(dependencies)
+            return dependencies
+
         # if pkg comes from additional_packages_snapshot, extract from its package.xml
         if (
             self.additional_packages_snapshot
@@ -207,25 +213,74 @@ def get_depends(self, pkg, ignore_pkgs=None):
         self._depends_cache[cache_key] = set(dependencies)
         return dependencies
 
+    def _get_snapshot_recursive_depends(self, pkg, ignore_pkgs=None):
+        """Return ROS dependencies using only package manifests pinned by the snapshot."""
+        dependencies = set()
+        ignored = set(ignore_pkgs or [])
+        packages_to_check = {pkg}
+        checked_packages = set()
+        dependency_attributes = (
+            "buildtool_depends",
+            "buildtool_export_depends",
+            "build_depends",
+            "build_export_depends",
+            "run_depends",
+            "test_depends",
+            "exec_depends",
+        )
+
+        while packages_to_check:
+            package_name = sorted(packages_to_check)[0]
+            packages_to_check.remove(package_name)
+            if package_name in ignored or package_name in checked_packages:
+                continue
+            checked_packages.add(package_name)
+
+            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 = {
+                dependency.name
+                for attribute in dependency_attributes
+                for dependency in getattr(package, attribute)
+                if dependency.evaluated_condition is not False
+                and dependency.name not in ignored
+                and self.check_package(dependency.name)
+            }
+            new_dependencies = direct_dependencies - dependencies
+            dependencies |= new_dependencies
+            packages_to_check |= new_dependencies - checked_packages
+
+        return dependencies
+
+    def _get_snapshot_package_info(self, pkg_name):
+        if not self.snapshot:
+            return None
+        for name in (pkg_name, pkg_name.replace("_", "-")):
+            if name in self.snapshot:
+                return self.snapshot[name]
+        return None
+
     def get_released_repo(self, pkg_name):
-        if self.snapshot and pkg_name in self.snapshot:
+        pkg_info = self._get_snapshot_package_info(pkg_name)
+        if pkg_info is not None:
             # In the case of snapshot, for rosdistro_additional_recipes
             # we also support a 'rev' field, so depending on what is available
             # we return either the tag or the rev, and the third argument is either 'rev' or 'tag'
-            url = self.snapshot[pkg_name].get("url", None)
+            url = pkg_info.get("url", None)
             # An archive URL is not a git repository: the reference is its sha256 checksum.
             if is_archive_url(url):
-                sha256 = self.snapshot[pkg_name].get("sha256", None)
+                sha256 = pkg_info.get("sha256", None)
                 if not sha256:
                     raise RuntimeError(
                         f"The archive source of '{pkg_name}' has no sha256 checksum: {url}"
                     )
                 return url, sha256, "sha256"
-            if "tag" in self.snapshot[pkg_name].keys():
-                tag_or_rev = self.snapshot[pkg_name].get("tag", None)
+            if "tag" in pkg_info:
+                tag_or_rev = pkg_info.get("tag", None)
                 ref_type = "tag"
             else:
-                tag_or_rev = self.snapshot[pkg_name].get("rev", None)
+                tag_or_rev = pkg_info.get("rev", None)
                 ref_type = "rev"
 
             return url, tag_or_rev, ref_type
@@ -244,23 +299,27 @@ def check_package(self, pkg_name):
             and pkg_name in self.additional_packages_snapshot
         ):
             return True
-        # the .replace('_', '-') is needed for packages like 'hpp-fcl' that have hypen and not underscore
+        if self.snapshot:
+            return (
+                self._get_snapshot_package_info(pkg_name) is not None
+                or pkg_name in self.build_packages
+            )
+        # the .replace('_', '-') is needed for packages like 'hpp-fcl' that have a hyphen and not underscore
         # in the rosdistro metadata
         if (
             pkg_name in self._distro.release_packages
             or pkg_name.replace("_", "-") in self._distro.release_packages
         ):
-            return self.snapshot is None or (
-                pkg_name in self.snapshot or pkg_name.replace("_", "-") in self.snapshot
-            )
+            return True
         elif pkg_name in self.build_packages:
             return True
         else:
             return False
 
     def get_version(self, pkg_name):
-        if self.snapshot and pkg_name in self.snapshot:
-            return self.snapshot[pkg_name].get("version", None)
+        pkg_info = self._get_snapshot_package_info(pkg_name)
+        if pkg_info is not None:
+            return pkg_info.get("version", None)
 
         pkg = self._distro.release_packages[pkg_name]
         repo = self._distro.repositories[pkg.repository_name].release_repository
@@ -273,6 +332,9 @@ def get_release_package_xml(self, pkg_name):
         ):
             pkg_info = self.additional_packages_snapshot[pkg_name]
             return self.get_package_xml_for_additional_package(pkg_info)
+        pkg_info = self._get_snapshot_package_info(pkg_name)
+        if pkg_info is not None:
+            return self.get_package_xml_for_additional_package(pkg_info)
         return self._distro.get_release_package_xml(pkg_name)
 
     def check_ros1(self):
@@ -294,6 +356,8 @@ def get_python_version(self):
         return self._python_version
 
     def get_package_names(self):
+        if self.snapshot:
+            return self.snapshot.keys()
         return self._distro.release_packages.keys()
 
     def get_package_xml_for_additional_package(self, pkg_info):

vinca/main.py

@@ -405,15 +405,9 @@ def generate_output(pkg_shortname, vinca_conf, distro, version, all_pkgs=None):
 
     xml = distro.get_release_package_xml(pkg_shortname)
 
-    # If the snapshot is not aligned with the latest rosdistro (for example if a package is removed,
-    # see https://github.com/RoboStack/ros-jazzy/pull/107#issuecomment-3338962041), get_release_package_xml can return none,
-    # in that case we can just skip the package, remove if https://github.com/RoboStack/vinca/issues/93 is fixed
+    # Without a snapshot, rosdistro can return no manifest for an unreleased package.
     if not xml:
-        print(
-            "Skip "
-            + pkg_shortname
-            + " as it is present in our snapshot, but not in the latest rosdistro cache."
-        )
+        print(f"Skip {pkg_shortname} because no release package.xml is available.")
         return None
 
     pkg = catkin_pkg.package.parse_package_string(xml)

vinca/test_snapshot_metadata.py

@@ -0,0 +1,157 @@
+from unittest.mock import Mock
+
+import vinca.main as main
+from vinca.distro import Distro
+
+
+def package_xml(name, version, build_dependency=None):
+    dependency = (
+        f"  <build_depend>{build_dependency}</build_depend>\n"
+        if build_dependency
+        else ""
+    )
+    return f"""\
+<package format="3">
+  <name>{name}</name>
+  <version>{version}</version>
+  <description>Test package</description>
+  <maintainer email="test@example.com">Test</maintainer>
+  <license>BSD-3-Clause</license>
+  <buildtool_depend>ament_cmake</buildtool_depend>
+{dependency}  <export><build_type>ament_cmake</build_type></export>
+</package>
+"""
+
+
+SNAPSHOT_PACKAGE_XML = package_xml("snapshot_package", "1.0.0", "snapshot_dependency")
+LIVE_PACKAGE_XML = package_xml("snapshot_package", "2.0.0", "live_dependency")
+SNAPSHOT_DEPENDENCY_XML = package_xml("snapshot_dependency", "1.0.0")
+
+
+def make_snapshot_distro(monkeypatch):
+    snapshot = {
+        "snapshot_package": {
+            "url": "https://github.com/example/snapshot-package-release.git",
+            "version": "1.0.0",
+            "tag": "release/rolling/snapshot_package/1.0.0-1",
+        },
+        "snapshot_dependency": {
+            "url": "https://github.com/example/snapshot-dependency-release.git",
+            "version": "1.0.0",
+            "tag": "release/rolling/snapshot_dependency/1.0.0-1",
+        },
+    }
+    distro = Distro.__new__(Distro)
+    distro.distro_name = "rolling"
+    distro.snapshot = snapshot
+    distro.additional_packages_snapshot = None
+    distro.build_packages = set()
+    distro._distribution_type = "ros2"
+    distro._additional_xml_cache = {}
+    distro._depends_cache = {}
+    distro._distro = Mock()
+    distro._distro.get_release_package_xml.return_value = LIVE_PACKAGE_XML
+    distro._walker = Mock()
+
+    snapshot_xml_by_url = {
+        "https://raw.githubusercontent.com/example/snapshot-package-release/"
+        "release/rolling/snapshot_package/1.0.0-1/package.xml": (SNAPSHOT_PACKAGE_XML),
+        "https://raw.githubusercontent.com/example/snapshot-dependency-release/"
+        "release/rolling/snapshot_dependency/1.0.0-1/package.xml": (
+            SNAPSHOT_DEPENDENCY_XML
+        ),
+    }
+    monkeypatch.setattr(
+        distro,
+        "_download_raw_pkg_xml_or_cached",
+        lambda url: snapshot_xml_by_url[url],
+    )
+    return distro
+
+
+def test_snapshot_package_xml_and_dependencies_do_not_follow_live_rosdistro(
+    monkeypatch,
+):
+    distro = make_snapshot_distro(monkeypatch)
+
+    package_xml_content = distro.get_release_package_xml("snapshot_package")
+
+    assert distro.get_released_repo("snapshot_package") == (
+        "https://github.com/example/snapshot-package-release.git",
+        "release/rolling/snapshot_package/1.0.0-1",
+        "tag",
+    )
+    assert distro.get_version("snapshot_package") == "1.0.0"
+    assert "<version>1.0.0</version>" in package_xml_content
+    assert "snapshot_dependency" in package_xml_content
+    assert "live_dependency" not in package_xml_content
+    assert distro.get_depends("snapshot_package") == {"snapshot_dependency"}
+    distro._distro.get_release_package_xml.assert_not_called()
+    distro._walker.get_recursive_depends.assert_not_called()
+
+
+def test_snapshot_metadata_generates_dependency_required_by_pinned_source(
+    monkeypatch,
+):
+    distro = make_snapshot_distro(monkeypatch)
+    dependency_names = {
+        "snapshot_package": "ros2-snapshot-package",
+        "python": "python",
+        "ament_cmake": "ros2-ament-cmake",
+        "snapshot_dependency": "ros2-snapshot-dependency",
+    }
+    monkeypatch.setattr(
+        main,
+        "resolve_pkgname",
+        lambda name, *_args, **_kwargs: [dependency_names[name]],
+    )
+    config = {
+        "_selected_pkgs": {"snapshot_package"},
+        "_pkg_additional_info": {},
+        "depmods": {},
+        "package_name_mode": "new",
+    }
+
+    output = main.generate_output(
+        "snapshot_package",
+        config,
+        distro,
+        distro.get_version("snapshot_package"),
+    )
+
+    assert output["package"] == {
+        "name": "ros2-snapshot-package",
+        "version": "1.0.0",
+    }
+    assert "ros2-snapshot-dependency" in output["requirements"]["host"]
+    assert "ros2-live-dependency" not in output["requirements"]["host"]
+
+
+def test_snapshot_is_authoritative_for_package_membership(monkeypatch):
+    distro = make_snapshot_distro(monkeypatch)
+    distro._distro.release_packages = {"live_only": Mock()}
+
+    assert distro.check_package("snapshot_package")
+    assert not distro.check_package("live_only")
+    assert set(distro.get_package_names()) == {
+        "snapshot_package",
+        "snapshot_dependency",
+    }
+
+
+def test_empty_snapshot_keeps_live_rosdistro_behavior():
+    distro = Distro.__new__(Distro)
+    distro.snapshot = {}
+    distro.additional_packages_snapshot = None
+    distro.build_packages = set()
+    distro._depends_cache = {}
+    distro._distro = Mock()
+    distro._distro.release_packages = {"live_package": Mock()}
+    distro._distro.get_release_package_xml.return_value = LIVE_PACKAGE_XML
+    distro._walker = Mock()
+    distro._walker.get_recursive_depends.return_value = {"live_dependency"}
+
+    assert distro.check_package("live_package")
+    assert distro.get_release_package_xml("live_package") == LIVE_PACKAGE_XML
+    assert distro.get_depends("live_package") == {"live_dependency"}
+    assert set(distro.get_package_names()) == {"live_package"}