feat(downloader): resolve latest via latest pointer target

Author: dkirov-dd

datadog_checks_downloader/datadog_checks/downloader/download_v2.py

@@ -7,11 +7,12 @@
 The v2 format stores a JSON pointer file as a TUF target at:
 
     targets/<project>/<version>.json   (versioned pointer)
+    targets/<project>/latest.json      (copy of latest stable version pointer)
 
 Target files are content-addressed on S3 (consistent-snapshot format):
-the actual file at ``targets/<project>/<version>.json`` is stored as
-``targets/<project>/{sha256}.{version}.json``.  The TUF Updater resolves
-the hash-prefixed path automatically via ``N.targets.json`` metadata.
+the actual file at ``targets/<project>/<name>.json`` is stored as
+``targets/<project>/{sha256}.{name}.json``.  The TUF Updater resolves
+the hash-prefixed path automatically via TUF metadata.
 
 Each pointer file contains:
 
@@ -38,7 +39,6 @@
 import urllib.request
 from pathlib import Path
 
-from packaging.version import InvalidVersion, Version
 from tuf.api.metadata import Metadata, Snapshot, Targets, Timestamp
 from tuf.ngclient import Updater
 from tuf.ngclient.config import UpdaterConfig
@@ -99,32 +99,9 @@ def _make_updater(self, metadata_dir: Path, target_dir: Path) -> Updater:
         )
 
     @staticmethod
-    def _resolve_version(project: str, version: str | None, targets: Targets) -> str:
-        """Return the concrete version to fetch.
-
-        If *version* is given, return it as-is (TUF will raise if absent).
-        If None, scan *targets* for all ``<project>/<ver>.json`` entries and
-        return the highest stable PEP 440 version.
-        """
-        if version is not None:
-            return version
-
-        prefix = f"{project}/"
-        stable: list[Version] = []
-        for target_path in targets.targets:
-            if not (target_path.startswith(prefix) and target_path.endswith(".json")):
-                continue
-            stem = target_path[len(prefix): -len(".json")]
-            try:
-                v = Version(stem)
-            except InvalidVersion:
-                continue
-            if not v.is_prerelease:
-                stable.append(v)
-
-        if not stable:
-            raise TargetNotFoundError(f"No stable releases found for {project!r} in targets metadata")
-        return str(max(stable))
+    def _target_path(project: str, version: str | None) -> str:
+        name = version if version is not None else 'latest'
+        return f'{project}/{name}.json'
 
     def _fetch_targets_unverified(self) -> Targets:
         """Fetch targets metadata without verifying TUF signatures.
@@ -152,9 +129,7 @@ def _fetch_targets_unverified(self) -> Targets:
     def get_pointer(self, project: str, version: str | None = None) -> dict:
         """Return the pointer JSON for *project* at *version*.
 
-        Resolves *version* = None to the latest stable release by scanning
-        ``N.targets.json``.  The pointer file is fetched via its hash-prefixed
-        consistent-snapshot path.
+        Resolves *version* = None by fetching ``<project>/latest.json``.
         Raises ``TargetNotFoundError`` if the target is absent from the TUF repo.
         """
         with tempfile.TemporaryDirectory() as tmp:
@@ -163,17 +138,17 @@ def get_pointer(self, project: str, version: str | None = None) -> dict:
             metadata_dir.mkdir()
             target_dir.mkdir()
 
+            target_path = self._target_path(project, version)
+            target_name = Path(target_path).stem
+
             if self._disable_verification:
                 targets = self._fetch_targets_unverified()
-                resolved = self._resolve_version(project, version, targets)
-                target_path = f'{project}/{resolved}.json'
-
                 entry = targets.targets.get(target_path)
                 if entry is None:
                     raise TargetNotFoundError(f'Target not found: {target_path}')
 
                 sha256 = entry.hashes.get('sha256', '')
-                url = f'{self._repository_url}/targets/{project}/{sha256}.{resolved}.json'
+                url = f'{self._repository_url}/targets/{project}/{sha256}.{target_name}.json'
                 try:
                     with urllib.request.urlopen(url) as resp:
                         return json.loads(resp.read())
@@ -186,19 +161,10 @@ def get_pointer(self, project: str, version: str | None = None) -> dict:
             updater = self._make_updater(metadata_dir, target_dir)
             updater.refresh()
 
-            # Parse the downloaded N.targets.json to resolve version and
-            # enumerate the hash-prefixed path the Updater will fetch.
-            candidates = sorted(metadata_dir.glob('*.targets.json'))
-            if not candidates:
-                raise TargetNotFoundError(f'No targets metadata after refresh for {project!r}')
-            targets = Metadata[Targets].from_file(str(candidates[-1])).signed
-
-            resolved = self._resolve_version(project, version, targets)
-            target_path = f'{project}/{resolved}.json'
-
             target_info = updater.get_targetinfo(target_path)
             if target_info is None:
-                raise TargetNotFoundError(f'No TUF target for {project!r} version {resolved!r}')
+                label = version if version is not None else 'latest stable'
+                raise TargetNotFoundError(f'No TUF target for {project!r} version {label!r}')
 
             pointer_path = target_dir / target_path
             pointer_path.parent.mkdir(parents=True, exist_ok=True)

datadog_checks_downloader/tests/test_v2_downloader.py

@@ -15,11 +15,9 @@
 from unittest.mock import MagicMock, patch
 
 import pytest
-from packaging.version import Version
-from tuf.api.metadata import MetaFile, Snapshot, Targets, Timestamp
-
 from datadog_checks.downloader.download_v2 import TUFPointerDownloader
 from datadog_checks.downloader.exceptions import DigestMismatch, TargetNotFoundError
+from tuf.api.metadata import Targets
 
 # ---------------------------------------------------------------------------
 # Shared fixtures
@@ -90,47 +88,16 @@ def fake_from_file(path):
 
 
 # ---------------------------------------------------------------------------
-# _resolve_version unit tests
+# target-path unit tests
 # ---------------------------------------------------------------------------
 
 
-class TestResolveVersion:
-    def test_explicit_version_returned_unchanged(self):
-        targets = _make_targets('1.0.0', '2.0.0')
-        assert TUFPointerDownloader._resolve_version(_PROJECT, '1.0.0', targets) == '1.0.0'
-
-    def test_latest_picks_highest_stable(self):
-        targets = _make_targets('1.0.0', '2.0.0', '1.5.0')
-        result = TUFPointerDownloader._resolve_version(_PROJECT, None, targets)
-        assert result == '2.0.0'
-
-    def test_latest_excludes_prereleases(self):
-        targets = _make_targets('1.0.0', '2.0.0rc1', '1.9.0b2', '1.8.0a1', '1.7.0.dev1')
-        result = TUFPointerDownloader._resolve_version(_PROJECT, None, targets)
-        assert result == '1.0.0'
-
-    def test_latest_includes_post_releases(self):
-        targets = _make_targets('1.0.0', '1.0.0.post1')
-        result = TUFPointerDownloader._resolve_version(_PROJECT, None, targets)
-        assert result == '1.0.0.post1'
+class TestTargetPath:
+    def test_explicit_version_uses_versioned_pointer(self):
+        assert TUFPointerDownloader._target_path(_PROJECT, '1.0.0') == f'{_PROJECT}/1.0.0.json'
 
-    def test_no_stable_versions_raises(self):
-        targets = _make_targets('1.0.0rc1', '2.0.0b1')
-        with pytest.raises(TargetNotFoundError, match=_PROJECT):
-            TUFPointerDownloader._resolve_version(_PROJECT, None, targets)
-
-    def test_ignores_other_projects(self):
-        from tuf.api.metadata import TargetFile
-
-        targets = Targets()
-        for path in ['datadog-mysql/1.0.0.json', f'{_PROJECT}/2.0.0.json']:
-            tf = MagicMock(spec=TargetFile)
-            tf.hashes = {'sha256': 'a' * 64}
-            tf.length = 100
-            targets.targets[path] = tf
-
-        result = TUFPointerDownloader._resolve_version(_PROJECT, None, targets)
-        assert result == '2.0.0'
+    def test_missing_version_uses_latest_pointer(self):
+        assert TUFPointerDownloader._target_path(_PROJECT, None) == f'{_PROJECT}/latest.json'
 
 
 # ---------------------------------------------------------------------------
@@ -163,7 +130,9 @@ def test_download_returns_wheel_path(self, mock_urlopen, mock_metadata_cls, mock
     @patch('datadog_checks.downloader.download_v2.Updater')
     @patch('datadog_checks.downloader.download_v2.Metadata')
     @patch('datadog_checks.downloader.download_v2.urllib.request.urlopen')
-    def test_repository_flag_overrides_pointer_repository(self, mock_urlopen, mock_metadata_cls, mock_updater_cls, tmp_path):
+    def test_repository_flag_overrides_pointer_repository(
+        self, mock_urlopen, mock_metadata_cls, mock_updater_cls, tmp_path
+    ):
         """--repository supersedes the repository field baked into the pointer."""
         staging_url = 'https://agent-integration-wheels-staging.s3.amazonaws.com'
         prod_pointer = {**_POINTER, 'repository': 'https://agent-integration-wheels-prod.s3.amazonaws.com'}
@@ -195,24 +164,20 @@ def fake_urlopen(url):
     @patch('datadog_checks.downloader.download_v2.Updater')
     @patch('datadog_checks.downloader.download_v2.Metadata')
     @patch('datadog_checks.downloader.download_v2.urllib.request.urlopen')
-    def test_version_none_resolves_to_latest_stable(self, mock_urlopen, mock_metadata_cls, mock_updater_cls, tmp_path):
-        """version=None picks the highest stable version from N.targets.json."""
-        targets = _make_targets('1.0.0', '2.0.0', '3.0.0rc1')
+    def test_version_none_fetches_latest_pointer(self, mock_urlopen, mock_metadata_cls, mock_updater_cls, tmp_path):
+        """version=None fetches <project>/latest.json."""
         mock_updater = _mock_tuf_updater(_POINTER)
         mock_updater_cls.return_value = mock_updater
-        mock_metadata_cls.__getitem__.return_value.from_file.return_value.signed = targets
 
         mock_urlopen.return_value.__enter__ = lambda s: s
         mock_urlopen.return_value.__exit__ = MagicMock(return_value=False)
         mock_urlopen.return_value.read.return_value = _WHEEL_CONTENT
 
-        with patch('datadog_checks.downloader.download_v2.Path.glob', return_value=iter(['1.targets.json'])):
-            downloader = TUFPointerDownloader(repository_url=_REPO_URL)
-            downloader.download(_PROJECT, dest_dir=tmp_path)
+        downloader = TUFPointerDownloader(repository_url=_REPO_URL)
+        downloader.download(_PROJECT, dest_dir=tmp_path)
 
-        # Updater should have been asked for the 2.0.0 target, not 3.0.0rc1
         call_args = mock_updater.get_targetinfo.call_args[0][0]
-        assert call_args == f'{_PROJECT}/2.0.0.json'
+        assert call_args == f'{_PROJECT}/latest.json'
 
 
 # ---------------------------------------------------------------------------
@@ -287,8 +252,6 @@ class TestDisableVerification:
     def test_skips_tuf_and_digest(self, mock_urlopen, tmp_path):
         """disable_verification fetches metadata chain to find hash-prefixed path,
         then downloads without digest checks."""
-        from tuf.api.metadata import MetaFile, TargetFile
-
         # Build fake metadata responses
         ts = MagicMock()
         ts.snapshot_meta.version = 1
@@ -300,7 +263,6 @@ def test_skips_tuf_and_digest(self, mock_urlopen, tmp_path):
         snap_md = MagicMock()
         snap_md.signed = snap
 
-        from tuf.api.metadata import TargetFile as TF
         tf = MagicMock()
         tf.hashes = {'sha256': 'a' * 64}
         targets = Targets()
@@ -316,7 +278,6 @@ def fake_from_bytes(data):
             call_idx[0] += 1
             return r
 
-        pointer_bytes = json.dumps(_POINTER).encode()
         bad_pointer_bytes = json.dumps({**_POINTER, 'digest': 'deadbeef' * 8}).encode()
 
         url_call = [0]
@@ -340,8 +301,6 @@ def fake_urlopen(url):
 
     @patch('datadog_checks.downloader.download_v2.urllib.request.urlopen')
     def test_404_raises_target_not_found(self, mock_urlopen):
-        from tuf.api.metadata import TargetFile
-
         ts = MagicMock()
         ts.snapshot_meta.version = 1
         ts_md = MagicMock()