Snapshot rosdistro dependency metadata

vinca/distro.py

@@ -7,7 +7,6 @@
 from concurrent.futures import ThreadPoolExecutor
 from typing import Iterable, Optional
 
-import catkin_pkg.package
 import requests
 from rosdistro import get_cached_distribution, get_index, get_index_url
 from rosdistro.dependency_walker import DependencyWalker
@@ -116,9 +115,12 @@ def __init__(
  python_version=None,
  snapshot=None,
  additional_packages_snapshot=None,
+ distribution_cache=None,
  ):
  index = get_index(get_index_url())
- self._distro = get_cached_distribution(index, distro_name)
+ self._distro = get_cached_distribution(
+ index, distro_name, cache=distribution_cache
+ )
  self.distro_name = distro_name
  self.snapshot = snapshot
  self.additional_packages_snapshot = additional_packages_snapshot
@@ -169,11 +171,6 @@ def get_depends(
 
  ignore_pkgs = set(ignore_pkgs or ())
 
- if self.snapshot:
- dependencies = self._get_snapshot_recursive_depends(pkg, ignore_pkgs)
- self._depends_cache[cache_key] = set(dependencies)
- return dependencies
-
  dependencies = set()
  visited = {pkg}
  packages_to_check = {pkg}
@@ -194,18 +191,32 @@ def get_depends(
  self._depends_cache[cache_key] = set(dependencies)
  return dependencies
 
+ def get_direct_depends(self, pkg: str) -> set[str]:
+ """Return the direct ROS dependencies of a package."""
+ return self._get_direct_depends(pkg)
+
  def _get_direct_depends(self, pkg: str) -> set[str]:
  """Return direct dependencies, caching package metadata across root walks."""
 
  if pkg in self._direct_depends_cache:
  return set(self._direct_depends_cache[pkg])
 
+ snapshot_info = self._get_snapshot_package_info(pkg)
+ additional_packages_snapshot = self.additional_packages_snapshot or {}
+ is_additional_package = pkg in additional_packages_snapshot
+ if snapshot_info is not None and not is_additional_package:
+ if "dependencies" not in snapshot_info:
+ raise RuntimeError(
+ f"Snapshot metadata for '{pkg}' has no dependencies; "
+ "regenerate the rosdistro snapshot"
+ )
+ direct = set(snapshot_info["dependencies"] or [])
+ self._direct_depends_cache[pkg] = set(direct)
+ return direct
+
  # if pkg comes from additional_packages_snapshot, extract from its package.xml
- if (
- self.additional_packages_snapshot
- and pkg in self.additional_packages_snapshot
- ):
- pkg_info = self.additional_packages_snapshot[pkg]
+ if is_additional_package:
+ pkg_info = additional_packages_snapshot[pkg]
  xml_str = self.get_package_xml_for_additional_package(pkg_info)
  # parse XML
  import xml.etree.ElementTree as ET
@@ -249,48 +260,6 @@ 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: str, ignore_pkgs: Optional[Iterable[str]] = None
- ) -> set[str]:
- """Return ROS dependencies using only package manifests pinned by the snapshot."""
- dependencies: set[str] = 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: set[str] = {
- 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
@@ -396,13 +365,7 @@ def get_version(self, pkg_name):
  return repo.version.split("-")[0]
 
  def live_cache_matches_snapshot(self, pkg_name, snapshot_entry):
- """Return whether rosdistro's cached manifest is the pinned snapshot source.
-
- A snapshot pins the release repository URL, the package-specific release
- tag, and the release version. Only an exact match may reuse rosdistro's
- local ``DistributionCache``; otherwise the manifest must be read from the
- immutable snapshot source.
- """
+ """Return whether rosdistro's cached manifest is the pinned snapshot source."""
 
  for live_name in (pkg_name, pkg_name.replace("_", "-")):
  try:

vinca/snapshot.py

@@ -2,6 +2,12 @@
 import datetime
 
 import yaml
+from rosdistro import (
+ DistributionCache,
+ get_distribution_cache_string,
+ get_index,
+ get_index_url,
+)
 
 from .distro import Distro
 
@@ -50,9 +56,14 @@ def main():
  # Get the current UTC time
  utc_time = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
 
- # Note: we intentionally do not pass any kind of additional packages snapshot
- # here, as it would pollute the snapshot with additional packages
- distro = Distro(args.distro)
+ # Fetch the cache once and use that exact data for both the package pins and
+ # the dependency metadata stored in the snapshot.
+ # TODO: Consider referencing immutable cache release assets instead if they
+ # become available through https://github.com/ros/rosdistro/pull/50112.
+ index = get_index(get_index_url())
+ cache_yaml = get_distribution_cache_string(index, args.distro)
+ cache = DistributionCache(args.distro, yaml.safe_load(cache_yaml))
+ distro = Distro(args.distro, distribution_cache=cache)
 
  if args.package is None:
  deps = distro.get_package_names()
@@ -68,7 +79,7 @@ def main():
 
  output = {}
 
- for dep in deps:
+ for dep in sorted(deps):
  try:
  url, tag, _ = distro.get_released_repo(dep)
  version = distro.get_version(dep)
@@ -80,7 +91,12 @@ def main():
  )
  continue
 
- output[dep] = {"url": url, "version": version, "tag": tag}
+ output[dep] = {
+ "url": url,
+ "version": version,
+ "tag": tag,
+ "dependencies": sorted(distro.get_direct_depends(dep)),
+ }
  if repository := distro.get_repository_url(dep):
  output[dep]["repository"] = repository
 

vinca/test_snapshot_metadata.py

@@ -1,6 +1,8 @@
 from typing import Any
 from unittest.mock import Mock, patch
 
+import pytest
+
 import vinca.main as main
 import vinca.recipes as recipes
 from vinca.distro import Distro
@@ -37,12 +39,14 @@ def make_snapshot_distro(monkeypatch):
  "repository": "https://github.com/example/snapshot-package.git",
  "version": "1.0.0",
  "tag": "release/rolling/snapshot_package/1.0.0-1",
+ "dependencies": ["snapshot_dependency"],
  },
  "snapshot_dependency": {
  "url": "https://github.com/example/snapshot-dependency-release.git",
  "repository": "https://github.com/example/snapshot-dependency.git",
  "version": "1.0.0",
  "tag": "release/rolling/snapshot_dependency/1.0.0-1",
+ "dependencies": [],
  },
  }
  distro = Distro.__new__(Distro)
@@ -55,9 +59,6 @@ def make_snapshot_distro(monkeypatch):
  distro._depends_cache = {}
  distro._direct_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),
@@ -71,6 +72,7 @@ def make_snapshot_distro(monkeypatch):
  "_download_raw_pkg_xml_or_cached",
  lambda url: snapshot_xml_by_url[url],
  )
+ distro._walker = Mock()
  return distro
 
 
@@ -98,7 +100,7 @@ def test_snapshot_package_xml_and_dependencies_do_not_follow_live_rosdistro(
  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()
+ distro._walker.get_depends.assert_not_called()
 
 
 def test_snapshot_package_xml_uses_matching_live_distribution_cache(monkeypatch):
@@ -113,6 +115,7 @@ def test_snapshot_package_xml_uses_matching_live_distribution_cache(monkeypatch)
  distro._distro.repositories = {
  "snapshot-package": Mock(release_repository=release_repository)
  }
+ distro._distro.get_release_package_xml.return_value = LIVE_PACKAGE_XML
 
  with patch(
  "vinca.distro.get_release_tag",
@@ -149,6 +152,14 @@ def test_snapshot_package_xml_does_not_use_live_cache_after_snapshot_change(
  distro._distro.get_release_package_xml.assert_not_called()
 
 
+def test_snapshot_without_dependencies_requires_regeneration(monkeypatch):
+ distro = make_snapshot_distro(monkeypatch)
+ del distro.snapshot["snapshot_package"]["dependencies"]
+
+ with pytest.raises(RuntimeError, match="regenerate the rosdistro snapshot"):
+ distro.get_depends("snapshot_package")
+
+
 def test_snapshot_metadata_generates_dependency_required_by_pinned_source(
  monkeypatch,
 ):
@@ -307,3 +318,31 @@ def test_empty_snapshot_keeps_live_rosdistro_behavior():
  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"}
+
+
+def test_read_snapshot_merges_additional_packages(tmp_path, monkeypatch):
+ snapshot_path = tmp_path / "rosdistro_snapshot.yaml"
+ additional_path = tmp_path / "rosdistro_additional_recipes.yaml"
+ snapshot_path.write_text(
+ """\
+snapshot_package:
+ version: 1.0.0
+"""
+ )
+ additional_path.write_text(
+ """\
+additional_package:
+ version: 2.0.0
+"""
+ )
+ monkeypatch.chdir(tmp_path)
+
+ snapshot, additional = main.read_snapshot(
+ {
+ "rosdistro_snapshot": snapshot_path.name,
+ "rosdistro_additional_recipes": additional_path.name,
+ }
+ )
+
+ assert set(snapshot) == {"snapshot_package", "additional_package"}
+ assert additional == {"additional_package": {"version": "2.0.0"}}