From b0f90627aea2b0eb54f46e1ee0a03b15126feba3 Mon Sep 17 00:00:00 2001 From: Richard Tibbles Date: Tue, 31 Mar 2026 20:48:44 -0700 Subject: [PATCH] fix: use copyPackage instead of syncSources for series copies syncSources doesn't specify a source series or version, so Launchpad can't correctly resolve which binaries to copy. copyPackage takes explicit source_name and version, matching how promote() already works. Also fixes: - has_published_binaries: 'not builds' returned True when no builds existed, treating "no builds" as "has binaries" - Handle obsolete series gracefully (like promote does) Co-Authored-By: Claude Opus 4.6 (1M context) --- scripts/launchpad_copy.py | 51 +++++++++++++++++-------------- tests/test_launchpad_copy.py | 58 +++++++++++++++++++++++------------- 2 files changed, 66 insertions(+), 43 deletions(-) diff --git a/scripts/launchpad_copy.py b/scripts/launchpad_copy.py index bb33fdc..3487871 100644 --- a/scripts/launchpad_copy.py +++ b/scripts/launchpad_copy.py @@ -202,7 +202,7 @@ def get_builds_for(self, ppa, name, version, series_name): def has_published_binaries(self, ppa, name, version, series_name): builds = self.get_builds_for(ppa, name, version, series_name) - return not builds or builds[0].buildstate == "Successfully built" + return bool(builds) and builds[0].buildstate == "Successfully built" def get_usable_sources(self, ppa, package_names, series_name): res = [] @@ -232,31 +232,36 @@ def get_usable_sources(self, ppa, package_names, series_name): res.append((name, version)) return res - def queue_copy(self, name, source_series, target_series, pocket): - self.queue[source_series, target_series, pocket].add(name) + def queue_copy(self, name, version, source_series, target_series, pocket): + self.queue[source_series, target_series, pocket].add((name, version)) def perform_queued_copies(self, ppa): first = True - for (source_series, target_series, pocket), names in self.queue.items(): - if not names: + for (source_series, target_series, pocket), packages in self.queue.items(): + if not packages: continue - if first: - log.info("") - first = False - log.info("Copying %s to %s", ", ".join(sorted(names)), target_series) - try: - ppa.syncSources( - from_archive=ppa, - to_series=target_series, - to_pocket=pocket, - include_binaries=True, - source_names=sorted(names), - ) - except lre.BadRequest as e: - if "same version already published" in str(e): - log.info("Already copied to %s — skipping", target_series) - else: - raise + for name, version in sorted(packages): + if first: + log.info("") + first = False + log.info("Copying %s %s to %s", name, version, target_series) + try: + ppa.copyPackage( + from_archive=ppa, + include_binaries=True, + to_series=target_series, + to_pocket=pocket, + source_name=name, + version=version, + ) + except lre.BadRequest as e: + msg = str(e) + if "same version already published" in msg: + log.info("Already copied to %s — skipping", target_series) + elif "is obsolete and will not accept new uploads" in msg: + log.info("Skip obsolete series %s for %s %s", target_series, name, version) + else: + raise def copy_to_series(self): """Copy packages from source series to all other supported Ubuntu series.""" @@ -279,7 +284,7 @@ def copy_to_series(self): mentioned = True log.info("%s %s missing from %s", name, version, target_series_name) if self.has_published_binaries(ppa, name, version, source_series): - self.queue_copy(name, source_series, target_series_name, POCKET) + self.queue_copy(name, version, source_series, target_series_name, POCKET) else: builds = self.get_builds_for(ppa, name, version, source_series) if builds: diff --git a/tests/test_launchpad_copy.py b/tests/test_launchpad_copy.py index 7810e75..a705b06 100644 --- a/tests/test_launchpad_copy.py +++ b/tests/test_launchpad_copy.py @@ -149,32 +149,33 @@ def test_raises_on_missing_command(self): class TestLaunchpadWrapper: """Test LaunchpadWrapper queue and filtering logic.""" - def test_queue_copy_accumulates_names(self): + def test_queue_copy_accumulates_packages(self): wrapper = LaunchpadWrapper() - wrapper.queue_copy("kolibri-server", "jammy", "noble", "Release") - wrapper.queue_copy("kolibri-server", "jammy", "focal", "Release") + wrapper.queue_copy("kolibri-server", "0.5.1-0ubuntu1", "jammy", "noble", "Release") + wrapper.queue_copy("kolibri-server", "0.5.1-0ubuntu1", "jammy", "focal", "Release") assert ("jammy", "noble", "Release") in wrapper.queue assert ("jammy", "focal", "Release") in wrapper.queue - assert "kolibri-server" in wrapper.queue[("jammy", "noble", "Release")] + assert ("kolibri-server", "0.5.1-0ubuntu1") in wrapper.queue[("jammy", "noble", "Release")] def test_queue_starts_empty(self): wrapper = LaunchpadWrapper() assert len(wrapper.queue) == 0 - def test_perform_queued_copies_calls_sync_sources(self): + def test_perform_queued_copies_calls_copy_package(self): wrapper = LaunchpadWrapper() - wrapper.queue_copy("kolibri-server", "jammy", "noble", "Release") + wrapper.queue_copy("kolibri-server", "0.5.1-0ubuntu1", "jammy", "noble", "Release") mock_ppa = MagicMock() wrapper.perform_queued_copies(mock_ppa) - mock_ppa.syncSources.assert_called_once_with( + mock_ppa.copyPackage.assert_called_once_with( from_archive=mock_ppa, + include_binaries=True, to_series="noble", to_pocket="Release", - include_binaries=True, - source_names=["kolibri-server"], + source_name="kolibri-server", + version="0.5.1-0ubuntu1", ) def test_perform_queued_copies_skips_empty_queues(self): @@ -182,19 +183,19 @@ def test_perform_queued_copies_skips_empty_queues(self): mock_ppa = MagicMock() wrapper.perform_queued_copies(mock_ppa) - mock_ppa.syncSources.assert_not_called() + mock_ppa.copyPackage.assert_not_called() - def test_perform_queued_copies_handles_already_synced(self): - """Idempotency: syncSources errors for already-copied packages are handled gracefully.""" + def test_perform_queued_copies_handles_already_published(self): + """Idempotency: copyPackage errors for already-copied packages are handled gracefully.""" wrapper = LaunchpadWrapper() - wrapper.queue_copy("kolibri-server", "jammy", "noble", "Release") + wrapper.queue_copy("kolibri-server", "0.5.1-0ubuntu1", "jammy", "noble", "Release") class MockBadRequest(Exception): pass mock_ppa = MagicMock() - mock_ppa.syncSources.side_effect = MockBadRequest( - "kolibri-server 0.9.0 in noble (same version already published)" + mock_ppa.copyPackage.side_effect = MockBadRequest( + "kolibri-server 0.5.1-0ubuntu1 in noble (same version already published)" ) with patch("launchpad_copy.lre") as mock_lre: @@ -203,17 +204,34 @@ class MockBadRequest(Exception): # Should not raise — the error is handled gracefully - def test_perform_queued_copies_logs_already_synced(self, caplog): - """Idempotency: logs a message when syncSources finds package already exists.""" + def test_perform_queued_copies_handles_obsolete_series(self): + """copyPackage errors for obsolete series are handled gracefully.""" + wrapper = LaunchpadWrapper() + wrapper.queue_copy("kolibri-server", "0.5.1-0ubuntu1", "jammy", "trusty", "Release") + + class MockBadRequest(Exception): + pass + + mock_ppa = MagicMock() + mock_ppa.copyPackage.side_effect = MockBadRequest("trusty is obsolete and will not accept new uploads") + + with patch("launchpad_copy.lre") as mock_lre: + mock_lre.BadRequest = MockBadRequest + wrapper.perform_queued_copies(mock_ppa) + + # Should not raise — the error is handled gracefully + + def test_perform_queued_copies_logs_already_published(self, caplog): + """Idempotency: logs a message when copyPackage finds package already exists.""" wrapper = LaunchpadWrapper() - wrapper.queue_copy("kolibri-server", "jammy", "noble", "Release") + wrapper.queue_copy("kolibri-server", "0.5.1-0ubuntu1", "jammy", "noble", "Release") class MockBadRequest(Exception): pass mock_ppa = MagicMock() - mock_ppa.syncSources.side_effect = MockBadRequest( - "kolibri-server 0.9.0 in noble (same version already published)" + mock_ppa.copyPackage.side_effect = MockBadRequest( + "kolibri-server 0.5.1-0ubuntu1 in noble (same version already published)" ) with (