tests: Add end-to-end test for the examples

examples/repository/repo

@@ -11,6 +11,7 @@ live repository by adding new target files periodically.
 
 import argparse
 import logging
+import socketserver
 import sys
 from datetime import datetime
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
@@ -107,6 +108,17 @@ class RepositoryServer(ThreadingHTTPServer):
         self.timeout = 1
         self.repo = SimpleRepository()
 
+    def server_bind(self):
+        """Bind the socket without resolving the bind address.
+
+        HTTPServer.server_bind() fills in server_name with a reverse DNS
+        lookup of the bind address. Nothing here uses server_name, and the
+        lookup happens before the socket starts listening, so a slow resolver
+        delays startup by however long it takes to answer.
+        """
+        socketserver.TCPServer.server_bind(self)
+        self.server_name, self.server_port = self.server_address[:2]
+
 
 def main(argv: List[str]) -> None:
     """Example repository server"""

tests/test_examples_e2e.py

@@ -0,0 +1,331 @@
+# Copyright 2026, the TUF contributors
+# SPDX-License-Identifier: MIT OR Apache-2.0
+
+"""End-to-end test that runs the repository, uploader and client examples.
+
+Unlike test_examples.py (which execs the self-contained 'manual_repo'
+scripts), this test wires the three interacting examples together the way
+a user would run them from the command line:
+
+* 'examples/repository/repo' is started as a live HTTP server,
+* 'examples/uploader/uploader' performs Trust-On-First-Use, claims a
+  delegation and uploads target files to that server,
+* 'examples/client/client' performs Trust-On-First-Use and downloads the
+  uploaded targets, which verifies them against the repository metadata.
+
+The test also uploads a second target after the client has already been
+initialized, then downloads it: this exercises the case from issue #2228
+where a client with metadata version N must update to version N+M.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+from typing import ClassVar
+from urllib.parse import quote
+
+import urllib3
+
+from tests import utils
+
+# The repository example serves metadata under this URL prefix.
+_METADATA_URL = "metadata"
+
+
+# Number of times to start the server before giving up. A restart is needed
+# when another process grabs the port in the small window between
+# _get_free_port() releasing it and the server binding it.
+_MAX_SERVER_STARTS = 5
+
+# Total budget for getting a server up, across restarts.
+_SERVER_START_TIMEOUT = 10
+
+
+def _get_free_port() -> int:
+    """Return a currently-free localhost TCP port.
+
+    The port is handed to the example repository server via '-p'. There is a
+    small race between closing this socket and the server binding the port;
+    _start_server() retries on a fresh port if the server loses that race.
+    """
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.bind((utils.TEST_HOST_ADDRESS, 0))
+        return int(sock.getsockname()[1])
+
+
+class TestExamplesEnd2End(unittest.TestCase):
+    """Run the repository, uploader and client examples against each other."""
+
+    examples_dir: ClassVar[Path]
+
+    server: subprocess.Popen[bytes]
+    _stopped_output: str | None
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.examples_dir = Path(__file__).resolve().parents[1] / "examples"
+
+    def setUp(self) -> None:
+        # The examples import their private helper modules by name (e.g.
+        # "from _simplerepo import SimpleRepository"), so each example dir
+        # must be importable. The client and uploader also write trusted
+        # metadata and keys under Path.home(): point HOME at a throwaway dir
+        # so the test does not touch the real home directory and starts from
+        # a clean slate every run.
+        self.test_dir = tempfile.mkdtemp()
+        self.addCleanup(self._rmtree, self.test_dir)
+
+        self.env = os.environ.copy()
+        self.env["HOME"] = self.test_dir
+        # The example scripts import their sibling modules, so the example
+        # directories go on PYTHONPATH. Prepend rather than replace: the test
+        # runner may already be using PYTHONPATH to make tuf itself importable,
+        # and dropping it leaves the subprocess unable to import tuf at all.
+        # The repo root goes on too: when tuf is not installed into the
+        # environment the test process imports it from the checkout via the
+        # current directory, and the subprocess runs with cwd=test_dir.
+        example_paths = [
+            str(self.examples_dir / name)
+            for name in ("repository", "uploader", "client")
+        ]
+        example_paths.append(str(self.examples_dir.parent))
+        existing_pythonpath = self.env.get("PYTHONPATH")
+        if existing_pythonpath:
+            example_paths.append(existing_pythonpath)
+        self.env["PYTHONPATH"] = os.pathsep.join(example_paths)
+
+        self._stopped_output = None
+        self._start_server()
+
+    def _start_server(self) -> None:
+        """Start the repository example and wait until it serves metadata.
+
+        The server is started as a subprocess on a free port and is given
+        _SERVER_START_TIMEOUT to bind that port and answer for root metadata.
+        A subprocess that loses the race for the port exits with "Address
+        already in use"; that is retried on a fresh port, as long as there is
+        budget left, rather than reported as a readiness failure. A subprocess
+        that dies for any other reason is reported immediately, since starting
+        it again would only repeat the same crash.
+        """
+        repo_script = self.examples_dir / "repository" / "repo"
+        deadline = time.monotonic() + _SERVER_START_TIMEOUT
+        for attempt in range(_MAX_SERVER_STARTS):
+            self.port = _get_free_port()
+            self.base_url = f"http://{utils.TEST_HOST_ADDRESS}:{self.port}"
+
+            self._stopped_output = None
+            self.server = subprocess.Popen(
+                [sys.executable, "-u", str(repo_script), "-p", str(self.port)],
+                env=self.env,
+                cwd=self.test_dir,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.STDOUT,
+            )
+
+            listening = self._wait_for_port(deadline)
+            if listening and self._wait_for_root(deadline):
+                # Tear the server down at the end of the test.
+                self.addCleanup(self._stop_server)
+                return
+
+            # Stop the server (which drains its output) before deciding
+            # whether to retry.
+            exited_early = self.server.poll() is not None
+            returncode = self.server.returncode
+            self._stop_server()
+            output = self._server_output()
+
+            lost_port_race = exited_early and "Address already in use" in output
+            if (
+                lost_port_race
+                and attempt + 1 < _MAX_SERVER_STARTS
+                and time.monotonic() < deadline
+            ):
+                continue
+
+            if exited_early:
+                reason = f"exited early with code {returncode}"
+            elif not listening:
+                reason = (
+                    f"did not start listening on port {self.port} within "
+                    f"{_SERVER_START_TIMEOUT}s"
+                )
+            else:
+                reason = (
+                    "started listening but did not serve root metadata "
+                    f"within {_SERVER_START_TIMEOUT}s"
+                )
+            self.fail(f"repository example {reason}:\n{output}")
+
+    def _rmtree(self, path: str) -> None:
+        # shutil.rmtree wrapper registered with addCleanup.
+        shutil.rmtree(path, ignore_errors=True)
+
+    def _stop_server(self) -> None:
+        self.server.terminate()
+        try:
+            self.server.wait(timeout=5)
+        except subprocess.TimeoutExpired:
+            self.server.kill()
+            self.server.wait()
+        # Stash any remaining output before closing the pipe so it stays
+        # available to _server_output() after the server has been stopped.
+        if self.server.stdout is not None:
+            self._stopped_output = self.server.stdout.read().decode(
+                "utf-8", errors="replace"
+            )
+            self.server.stdout.close()
+
+    def _wait_for_port(self, deadline: float) -> bool:
+        """Poll until the server accepts connections on its port.
+
+        Returns False if the deadline passes or the server process exits
+        first, so the caller can tell a server that never bound apart from one
+        that bound but never served.
+        """
+        address = (utils.TEST_HOST_ADDRESS, self.port)
+        while time.monotonic() < deadline:
+            if self.server.poll() is not None:
+                # Server exited; let the caller inspect its output.
+                return False
+            try:
+                with socket.create_connection(address, timeout=1):
+                    return True
+            except OSError:
+                # Not listening yet (ECONNREFUSED on every platform we test).
+                time.sleep(0.05)
+        return False
+
+    def _wait_for_root(self, deadline: float) -> bool:
+        """Poll until the server serves 1.root.json.
+
+        Returns True once the server answers with the initial root metadata.
+        Returns False if the server process exits before then or the deadline
+        passes.
+        """
+        url = f"{self.base_url}/{_METADATA_URL}/1.root.json"
+        while time.monotonic() < deadline:
+            if self.server.poll() is not None:
+                # Server exited; let the caller inspect its output.
+                return False
+            try:
+                # The example accepts connections before its request loop
+                # runs, so bound the wait for a response and do not let
+                # urllib3 spend the deadline on its own retries.
+                response = urllib3.request("GET", url, timeout=1, retries=False)
+                if response.status == 200:
+                    return True
+            except urllib3.exceptions.HTTPError:
+                # Connection refused/reset while the server is still starting.
+                pass
+            time.sleep(0.05)
+        return False
+
+    def _server_output(self) -> str:
+        # If the server has been stopped, _stop_server() already drained the
+        # pipe; otherwise read whatever it has produced so far.
+        if self._stopped_output is not None:
+            return self._stopped_output
+        if self.server.stdout is None:
+            return ""
+        return self.server.stdout.read().decode("utf-8", errors="replace")
+
+    def _run_example(self, script: str, *args: str) -> None:
+        """Run an example script as a subprocess and assert it succeeds."""
+        script_path = self.examples_dir / script
+        result = subprocess.run(
+            [sys.executable, str(script_path), "-u", self.base_url, *args],
+            env=self.env,
+            cwd=self.test_dir,
+            capture_output=True,
+            text=True,
+            timeout=60,
+            check=False,
+        )
+        self.assertEqual(
+            result.returncode,
+            0,
+            f"{script} {' '.join(args)} failed (rc={result.returncode}):\n"
+            f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}",
+        )
+
+    def _snapshot_version(self) -> int:
+        """Return the version of the repository's current snapshot metadata."""
+        url = f"{self.base_url}/{_METADATA_URL}/snapshot.json"
+        response = urllib3.request("GET", url)
+        self.assertEqual(response.status, 200)
+        return int(json.loads(response.data)["signed"]["version"])
+
+    def _downloaded_target(self, targetpath: str) -> bytes:
+        """Read a target the client downloaded into its downloads dir.
+
+        The client writes targets into a 'downloads' dir at its CWD, using the
+        URL-quoted targetpath as filename (see Updater file path generation).
+        """
+        path = Path(self.test_dir) / "downloads" / quote(targetpath, "")
+        self.assertTrue(path.exists(), f"missing downloaded target {path}")
+        return path.read_bytes()
+
+    def test_repository_uploader_client(self) -> None:
+        """Upload targets via the uploader and download them via the client.
+
+        Covers the end-to-end "examples" slice of issue #2228: a client can
+        download a target and can update across metadata versions (N -> N+M).
+        """
+        uploader = "uploader/uploader"
+        client = "client/client"
+        role = "myrole"
+        first_target = f"{role}/first"
+        second_target = f"{role}/second"
+
+        # Maintainer side: trust the repo, claim a delegation, add a target.
+        self._run_example(uploader, "tofu")
+        self._run_example(uploader, "add-delegation", role)
+        self._run_example(uploader, "add-target", role, first_target)
+
+        version_after_first = self._snapshot_version()
+
+        # Client side: trust the repo and download the first target. The
+        # Updater verifies the target against the (signed) repository metadata
+        # before download_target() returns, so a successful download means the
+        # metadata chain and the target hash both verified.
+        self._run_example(client, "tofu")
+        self._run_example(client, "download", first_target)
+
+        # The example generates target content equal to the targetpath.
+        self.assertEqual(
+            self._downloaded_target(first_target),
+            first_target.encode(),
+        )
+
+        # Maintainer adds a second target. This publishes new targets,
+        # snapshot and timestamp metadata, so the repository advances past the
+        # version the client currently trusts.
+        self._run_example(uploader, "add-target", role, second_target)
+        version_after_second = self._snapshot_version()
+        self.assertGreater(version_after_second, version_after_first)
+
+        # Client downloads the second target. download() refreshes metadata
+        # first, so the already-initialized client must update from the
+        # version it trusts (N) to the newer version (N+M) to find and verify
+        # the new target.
+        self._run_example(client, "download", second_target)
+        self.assertEqual(
+            self._downloaded_target(second_target),
+            second_target.encode(),
+        )
+
+
+if __name__ == "__main__":
+    utils.configure_test_logging(sys.argv)
+    unittest.main()