Skip to content

fix: use copyPackage instead of syncSources for series copies #125

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rtibbles merged 1 commit into from
Apr 1, 2026
Merged

fix: use copyPackage instead of syncSources for series copies #125

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 28 additions & 23 deletions scripts/launchpad_copy.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 = []
Expand Down Expand Up @@ -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."""
Expand All @@ -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:
Expand Down
58 changes: 38 additions & 20 deletions tests/test_launchpad_copy.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -149,52 +149,53 @@ 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):
wrapper = LaunchpadWrapper()
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:
Expand All @@ -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 (
Expand Down
Loading