diff --git a/tools/lp-queue/.gitignore b/tools/lp-queue/.gitignore new file mode 100644 index 0000000..1f109d7 --- /dev/null +++ b/tools/lp-queue/.gitignore @@ -0,0 +1,2 @@ +*.egg-info +.pytest_cache diff --git a/tools/lp-queue/README.md b/tools/lp-queue/README.md new file mode 100644 index 0000000..50c158a --- /dev/null +++ b/tools/lp-queue/README.md @@ -0,0 +1,71 @@ +# LP Queue TUI + +A small standalone TUI tool for working with the Ubuntu Launchpad upload queue. + +## Features + +- **List** all items in the upload queue for a given Ubuntu series +- **Review** packages by viewing the debdiff between the current archive version and the new upload +- **Accept** packages into the archive +- **Reject** packages with a comment explaining the reason + +## Installation + +```bash +sudo apt install python3-launchpadlib python3-textual +sudo apt install --no-install-recommends diffoscope-minimal +ln -s "$(realpath lp-queue)" ~/.local/bin/lp-queue +``` +or +```bash +pip install . +``` + +For development: + +```bash +pip install -e ".[dev]" +``` + +## Usage + +Run the TUI: + +```bash +lp-queue +``` + +Or as a Python module: + +```bash +python -m lp_queue +``` + +## Keybindings + +| Key | Action | +|------|-------------------------------------| +| `r` | Review selected package (debdiff) | +| `a` | Accept selected package | +| `j` | Reject selected package (with comment) | +| `F5` | Refresh the queue listing | +| `q` | Quit the application | + +## Requirements + +- Python 3.10+ +- `launchpadlib` for Launchpad API access +- `textual` for the terminal UI framework +- Ubuntu archive access credentials (obtained via Launchpad OAuth) + +## How It Works + +1. On startup, the tool authenticates with Launchpad using `launchpadlib` +2. It fetches the current upload queue for the configured Ubuntu series +3. Items are displayed in a table with package name, version, component, etc. +4. You can review, accept, or reject packages using the keybindings above + +For syncs from Debian, the review shows the debdiff comparing the Debian +version against what is currently in the Ubuntu archive. You can also visit +the Debian tracker at `https://tracker.debian.org/pkg/` for +additional context. diff --git a/tools/lp-queue/lp-queue b/tools/lp-queue/lp-queue new file mode 100755 index 0000000..cf64f75 --- /dev/null +++ b/tools/lp-queue/lp-queue @@ -0,0 +1,6 @@ +#!/usr/bin/bash + +BASE_DIR="$(dirname "$(realpath "$0")")" + +cd "$BASE_DIR" +exec python3 -m lp_queue diff --git a/tools/lp-queue/lp_queue/__init__.py b/tools/lp-queue/lp_queue/__init__.py new file mode 100644 index 0000000..705cce6 --- /dev/null +++ b/tools/lp-queue/lp_queue/__init__.py @@ -0,0 +1 @@ +"""TUI tool for managing the Ubuntu Launchpad upload queue.""" diff --git a/tools/lp-queue/lp_queue/__main__.py b/tools/lp-queue/lp_queue/__main__.py new file mode 100644 index 0000000..88fe5ba --- /dev/null +++ b/tools/lp-queue/lp_queue/__main__.py @@ -0,0 +1,5 @@ +"""Entry point for running the TUI as a module: python -m lp_queue.""" + +from lp_queue.app import main + +main() diff --git a/tools/lp-queue/lp_queue/app.py b/tools/lp-queue/lp_queue/app.py new file mode 100644 index 0000000..0480456 --- /dev/null +++ b/tools/lp-queue/lp_queue/app.py @@ -0,0 +1,738 @@ +"""TUI application for the Ubuntu Launchpad upload queue.""" + +from __future__ import annotations + +from textual import on, work +from textual.app import App, ComposeResult +from textual.binding import Binding +from textual.containers import Vertical +from textual.css.query import QueryError +from textual.screen import ModalScreen +from textual.widgets import ( + Button, + DataTable, + Footer, + Header, + Input, + Label, + OptionList, + RichLog, +) +from textual.widgets.option_list import Option + +from lp_queue.launchpad import ( + QUEUE_STATUS_UNAPPROVED, + QUEUE_STATUSES, + LaunchpadQueue, + QueueItem, +) + + +class ReviewScreen(ModalScreen[None]): + """Modal screen to display a debdiff for review.""" + + BINDINGS = [ + Binding("escape", "dismiss", "Close"), + Binding("q", "dismiss", "Close"), + ] + + DEFAULT_CSS = """ + ReviewScreen { + align: center middle; + } + + ReviewScreen > Vertical { + width: 90%; + height: 90%; + border: thick $accent; + background: $surface; + padding: 1 2; + } + + ReviewScreen RichLog { + height: 1fr; + } + + ReviewScreen .review-title { + dock: top; + text-style: bold; + padding: 1; + background: $accent; + color: $text; + width: 100%; + } + """ + + def __init__(self, item: QueueItem, debdiff: str) -> None: + super().__init__() + self.item = item + self.debdiff = debdiff + + def compose(self) -> ComposeResult: + """Build the review screen layout.""" + with Vertical(): + yield Label(f"Review: {self.item.display_name}", classes="review-title") + yield RichLog(highlight=True, markup=False, auto_scroll=False) + + def on_mount(self) -> None: + """Load the debdiff content into the log widget.""" + from rich.syntax import Syntax + + log = self.query_one(RichLog) + log.write(Syntax(self.debdiff, "diff", line_numbers=False)) + + +class RejectScreen(ModalScreen[str | None]): + """Modal screen to collect a rejection comment.""" + + BINDINGS = [ + Binding("escape", "cancel", "Cancel"), + ] + + DEFAULT_CSS = """ + RejectScreen { + align: center middle; + } + + RejectScreen > Vertical { + width: 60%; + height: auto; + max-height: 50%; + border: thick $error; + background: $surface; + padding: 1 2; + } + + RejectScreen .reject-title { + text-style: bold; + padding: 1; + background: $error; + color: $text; + width: 100%; + } + + RejectScreen Input { + margin: 1 0; + } + """ + + def __init__(self, item: QueueItem) -> None: + super().__init__() + self.item = item + + def compose(self) -> ComposeResult: + """Build the rejection comment input screen.""" + with Vertical(): + yield Label(f"Reject: {self.item.display_name}", classes="reject-title") + yield Label("Enter rejection reason:") + yield Input(placeholder="Reason for rejection...") + + @on(Input.Submitted) + def on_input_submitted(self, event: Input.Submitted) -> None: + """Handle the rejection comment submission.""" + comment = event.value.strip() + if comment: + self.dismiss(comment) + + def action_cancel(self) -> None: + """Cancel the rejection.""" + self.dismiss(None) + + +class ConfirmScreen(ModalScreen[bool]): + """Modal screen asking the user to confirm an action.""" + + BINDINGS = [ + Binding("y", "confirm", "Yes"), + Binding("n", "cancel", "No"), + Binding("escape", "cancel", "Cancel"), + ] + + DEFAULT_CSS = """ + ConfirmScreen { + align: center middle; + } + + ConfirmScreen > Vertical { + width: 50%; + height: auto; + max-height: 50%; + border: thick $accent; + background: $surface; + padding: 1 2; + } + + ConfirmScreen .confirm-title { + text-style: bold; + padding: 1; + width: 100%; + } + + ConfirmScreen .confirm-message { + padding: 0 1 1 1; + } + + ConfirmScreen .confirm-buttons { + height: auto; + width: 100%; + align: center middle; + padding: 1 0 0 0; + } + + ConfirmScreen Button { + margin: 0 1; + } + """ + + def __init__(self, title: str, message: str) -> None: + super().__init__() + self._title = title + self._message = message + + def compose(self) -> ComposeResult: + """Build the confirmation dialog layout.""" + from textual.containers import Horizontal + + with Vertical(): + yield Label(self._title, classes="confirm-title") + yield Label(self._message, classes="confirm-message") + with Horizontal(classes="confirm-buttons"): + yield Button("Yes", variant="success", id="confirm-yes") + yield Button("No", variant="error", id="confirm-no") + + @on(Button.Pressed, "#confirm-yes") + def on_confirm(self) -> None: + """Confirm the action.""" + self.dismiss(True) + + @on(Button.Pressed, "#confirm-no") + def on_deny(self) -> None: + """Cancel the action.""" + self.dismiss(False) + + def action_confirm(self) -> None: + """Confirm via keyboard shortcut.""" + self.dismiss(True) + + def action_cancel(self) -> None: + """Cancel via keyboard shortcut.""" + self.dismiss(False) + + +class SeriesScreen(ModalScreen[str | None]): + """Modal screen to select an Ubuntu series.""" + + BINDINGS = [ + Binding("escape", "cancel", "Cancel"), + ] + + DEFAULT_CSS = """ + SeriesScreen { + align: center middle; + } + + SeriesScreen > Vertical { + width: 50%; + height: 70%; + border: thick $accent; + background: $surface; + padding: 1 2; + } + + SeriesScreen .series-title { + text-style: bold; + padding: 1; + background: $accent; + color: $text; + width: 100%; + } + + SeriesScreen OptionList { + height: 1fr; + margin: 1 0; + } + """ + + def __init__( + self, + series_list: list[tuple[str, str, str]], + current_series: str, + ) -> None: + super().__init__() + self._series_list = series_list + self._current_series = current_series + + def compose(self) -> ComposeResult: + """Build the series selection screen.""" + with Vertical(): + yield Label("Switch Ubuntu Series", classes="series-title") + yield OptionList(*self._build_options()) + + def _build_options(self) -> list[Option]: + """Build OptionList entries from the series data.""" + options: list[Option] = [] + for name, version, status in self._series_list: + marker = " ✦ " if name == self._current_series else " " + options.append(Option(f"{marker} {name} ({version}) {status}", id=name)) + return options + + @on(OptionList.OptionSelected) + def on_option_selected(self, event: OptionList.OptionSelected) -> None: + """Handle series selection.""" + self.dismiss(event.option.id) + + def action_cancel(self) -> None: + """Cancel series selection.""" + self.dismiss(None) + + +class QueueStatusScreen(ModalScreen[str | None]): + """Modal screen to select a queue status filter.""" + + BINDINGS = [ + Binding("escape", "cancel", "Cancel"), + ] + + DEFAULT_CSS = """ + QueueStatusScreen { + align: center middle; + } + + QueueStatusScreen > Vertical { + width: 50%; + height: 70%; + border: thick $accent; + background: $surface; + padding: 1 2; + } + + QueueStatusScreen .status-title { + text-style: bold; + padding: 1; + background: $accent; + color: $text; + width: 100%; + } + + QueueStatusScreen OptionList { + height: 1fr; + margin: 1 0; + } + """ + + def __init__(self, current_status: str) -> None: + super().__init__() + self._current_status = current_status + + def compose(self) -> ComposeResult: + """Build the queue status selection screen.""" + with Vertical(): + yield Label("Switch Queue Status", classes="status-title") + yield OptionList(*self._build_options()) + + def _build_options(self) -> list[Option]: + """Build OptionList entries from the queue status list.""" + options: list[Option] = [] + for status in QUEUE_STATUSES: + marker = " ✦ " if status == self._current_status else " " + options.append(Option(f"{marker} {status}", id=status)) + return options + + @on(OptionList.OptionSelected) + def on_option_selected(self, event: OptionList.OptionSelected) -> None: + """Handle queue status selection.""" + self.dismiss(event.option.id) + + def action_cancel(self) -> None: + """Cancel queue status selection.""" + self.dismiss(None) + + +class QueueApp(App[None]): + """TUI application for managing the Ubuntu upload queue.""" + + TITLE = "Ubuntu Upload Queue" + + AUTO_REFRESH_SECONDS = 300 # 5 minutes + + BINDINGS = [ + Binding("r", "review", "Review"), + Binding("a", "accept", "Accept"), + Binding("j", "reject", "Reject"), + Binding("f2", "switch_series", "Series"), + Binding("f3", "switch_queue_status", "Queue"), + Binding("f5", "refresh", "Refresh"), + Binding("tilde", "toggle_debug", "Debug log"), + Binding("q", "quit", "Quit"), + ] + + DEFAULT_CSS = """ + #debug-panel { + height: 12; + border-top: thick $accent; + background: $surface; + display: none; + } + + #debug-panel.visible { + display: block; + } + + .debug-title { + dock: top; + text-style: bold; + padding: 0 1; + background: $accent; + color: $text; + width: 100%; + height: 1; + } + + #status-bar { + width: 100%; + height: 1; + background: $accent; + color: $text; + padding: 0 1; + } + + #status-bar.busy { + background: $warning; + color: $text; + text-style: bold; + } + + #status-bar.error { + background: $error; + color: $text; + text-style: bold; + } + + #status-bar.success { + background: $success; + color: $text; + } + """ + + def __init__(self, lp_queue: LaunchpadQueue | None = None) -> None: + super().__init__() + self.lp_queue = lp_queue or LaunchpadQueue() + self.queue_items: list[QueueItem] = [] + self.queue_status: str = QUEUE_STATUS_UNAPPROVED + self.username = "" + self._pending_reject_comment = "" + + def compose(self) -> ComposeResult: + """Build the main application layout.""" + yield Header() + with Vertical(id="main-view"): + yield DataTable() + with Vertical(id="debug-panel"): + yield Label("Launchpad Debug Log", classes="debug-title") + yield RichLog(id="debug-log", highlight=True, markup=False) + yield Label("⏳ Connecting to Launchpad…", id="status-bar", classes="busy") + yield Footer() + + def on_mount(self) -> None: + """Set up the data table and load queue items.""" + table = self.query_one(DataTable) + table.cursor_type = "row" + table.add_columns( + "Package", + "Version", + "Component", + "Section", + "Author(s)", + "Status", + "Sync", + "Created", + ) + self.lp_queue.set_log_callback(self._on_lp_log) + self._connect_and_load() + self.set_interval(self.AUTO_REFRESH_SECONDS, self._periodic_refresh) + + def _periodic_refresh(self) -> None: + """Refresh the queue automatically if no modal screen is open.""" + if len(self.screen_stack) == 1: + self._load_queue() + + def _on_lp_log(self, message: str) -> None: + """Receive a log message from LaunchpadQueue and write it to the debug panel.""" + self.call_from_thread(self._write_debug_log, message) + + def _write_debug_log(self, message: str) -> None: + """Append a message to the debug RichLog widget.""" + self.query_one("#debug-log", RichLog).write(message) + + def action_toggle_debug(self) -> None: + """Toggle visibility of the Launchpad debug log panel.""" + panel = self.query_one("#debug-panel") + panel.toggle_class("visible") + + def _set_username(self, username): + self.username = username + + def _set_status(self, message: str, state: str = "") -> None: + """Update the status bar message and visual state. + + Args: + message: The status text to display. + state: Visual state — ``"busy"``, ``"error"``, ``"success"``, or + ``""`` for the default/idle look. + + """ + try: + bar = self.query_one("#status-bar", Label) + except QueryError: + return None + + bar.remove_class("busy", "error", "success") + if state: + bar.add_class(state) + msg = message.replace("(", r"\(").replace("[", r"\[").replace("{", r"\{") + if self.username: + msg = f"Connected as '{self.username}' - {msg}" + bar.update(msg) + + def _get_selected_item(self) -> QueueItem | None: + """Return the currently selected queue item, or None.""" + table = self.query_one(DataTable) + if table.row_count == 0: + return None + row_idx = table.cursor_row + if 0 <= row_idx < len(self.queue_items): + return self.queue_items[row_idx] + return None + + @work(thread=True) + def _connect_and_load(self) -> None: + """Connect to Launchpad and load the queue (runs in a worker thread).""" + try: + self.app.call_from_thread(self._set_status, "⏳ Connecting to Launchpad…", "busy") + self.lp_queue.connect() + self.app.call_from_thread(self._set_status, "⏳ Getting user name…", "busy") + self.app.call_from_thread(self._set_username, self.lp_queue.lp_user_name()) + self.app.call_from_thread(self._set_status, "⏳ Loading queue items…", "busy") + self.queue_items = self.lp_queue.get_queue_items(self.queue_status) + self.app.call_from_thread(self._populate_table) + except Exception as exc: + self.app.call_from_thread(self._set_status, f"❌ Error: {exc}", "error") + + @work(thread=True) + def _load_queue(self) -> None: + """Refresh queue items from Launchpad (runs in a worker thread).""" + try: + self.app.call_from_thread(self._set_status, "⏳ Refreshing queue…", "busy") + self.queue_items = self.lp_queue.get_queue_items(self.queue_status) + self.app.call_from_thread(self._populate_table) + except Exception as exc: + self.app.call_from_thread(self._set_status, f"❌ Error: {exc}", "error") + + def _populate_table(self) -> None: + """Populate the data table with current queue items.""" + table = self.query_one(DataTable) + table.clear() + for item in self.queue_items: + table.add_row( + item.source_name, + item.version, + item.component, + item.section, + item.authors, + item.status, + "Yes" if item.is_sync else "No", + item.date_created, + ) + self._set_status(f"{len(self.queue_items)} items in queue") + + def action_refresh(self) -> None: + """Refresh the queue listing.""" + self._load_queue() + + def action_switch_series(self) -> None: + """Open the series selection screen.""" + self._fetch_series() + + @work(thread=True) + def _fetch_series(self) -> None: + """Fetch the list of Ubuntu series in a worker thread.""" + self.app.call_from_thread(self._set_status, "⏳ Loading series list…", "busy") + try: + series_list = getattr(self, "series_list", None) + if not series_list: + series_list = self.lp_queue.get_all_series() + self.series_list = series_list + self.app.call_from_thread( + self.push_screen, + SeriesScreen(series_list, self.lp_queue.series), + self._handle_series_result, + ) + self.app.call_from_thread(self._set_status, "Select a series") + except Exception as exc: + self.app.call_from_thread(self._set_status, f"❌ Error: {exc}", "error") + + def _handle_series_result(self, series_name: str | None) -> None: + """Process the result from the series selection screen.""" + if series_name is None: + self._set_status("Series switch cancelled") + return + if series_name == self.lp_queue.series: + self._set_status(f"Already on {series_name}") + return + self._do_switch_series(series_name) + + @work(thread=True) + def _do_switch_series(self, series_name: str) -> None: + """Switch the active series and reload the queue.""" + self.app.call_from_thread( + self._set_status, f"⏳ Switching to {series_name}…", "busy" + ) + try: + self.lp_queue.switch_series(series_name) + self.queue_items = self.lp_queue.get_queue_items(self.queue_status) + self.app.call_from_thread(self._populate_table) + except Exception as exc: + self.app.call_from_thread(self._set_status, f"❌ Error: {exc}", "error") + + def action_switch_queue_status(self) -> None: + """Open the queue status selection screen.""" + self.push_screen( + QueueStatusScreen(self.queue_status), + self._handle_queue_status_result, + ) + + def _handle_queue_status_result(self, status: str | None) -> None: + """Process the result from the queue status selection screen.""" + if status is None: + self._set_status("Queue status switch cancelled") + return + if status == self.queue_status: + self._set_status(f"Already showing {status}") + return + self.queue_status = status + self._load_queue() + + def action_review(self) -> None: + """Review the selected queue item by showing its debdiff.""" + item = self._get_selected_item() + if item is None: + self._set_status("No item selected") + return + self._fetch_debdiff(item) + + @work(thread=True) + def _fetch_debdiff(self, item: QueueItem) -> None: + """Fetch the debdiff in a worker thread and push the review screen.""" + self.app.call_from_thread( + self._set_status, f"⏳ Fetching debdiff for {item.display_name}…", "busy" + ) + try: + debdiff = self.lp_queue.get_debdiff(item) + self.app.call_from_thread(self.push_screen, ReviewScreen(item, debdiff)) + self.app.call_from_thread( + self._set_status, f"Reviewing {item.display_name}" + ) + except Exception as exc: + self.app.call_from_thread(self._set_status, f"❌ Error: {exc}", "error") + + def action_accept(self) -> None: + """Accept the selected queue item after confirmation.""" + item = self._get_selected_item() + if item is None: + self._set_status("No item selected") + return + self.push_screen( + ConfirmScreen( + "Accept Package", + f"Are you sure you want to accept {item.display_name}?", + ), + self._handle_accept_confirm, + ) + + def _handle_accept_confirm(self, confirmed: bool) -> None: + """Process the result from the accept confirmation dialog.""" + if not confirmed: + self._set_status("Accept cancelled") + return + item = self._get_selected_item() + if item is None: + return + self._do_accept(item) + + @work(thread=True) + def _do_accept(self, item: QueueItem) -> None: + """Accept the item in a worker thread.""" + self.app.call_from_thread( + self._set_status, f"⏳ Accepting {item.display_name}…", "busy" + ) + try: + self.lp_queue.accept(item) + self.app.call_from_thread( + self._set_status, f"✔ Accepted {item.display_name}", "success" + ) + self.queue_items = self.lp_queue.get_queue_items(self.queue_status) + self.app.call_from_thread(self._populate_table) + except Exception as exc: + self.app.call_from_thread(self._set_status, f"❌ Error: {exc}", "error") + + def action_reject(self) -> None: + """Reject the selected queue item, asking for a comment.""" + item = self._get_selected_item() + if item is None: + self._set_status("No item selected") + return + self.push_screen(RejectScreen(item), self._handle_reject_result) + + def _handle_reject_result(self, comment: str | None) -> None: + """Process the result from the reject screen.""" + if comment is None: + self._set_status("Rejection cancelled") + return + item = self._get_selected_item() + if item is None: + return + self._pending_reject_comment = comment + self.push_screen( + ConfirmScreen( + "Reject Package", + f"Are you sure you want to reject {item.display_name}?", + ), + self._handle_reject_confirm, + ) + + def _handle_reject_confirm(self, confirmed: bool) -> None: + """Process the result from the reject confirmation dialog.""" + if not confirmed: + self._set_status("Rejection cancelled") + return + item = self._get_selected_item() + if item is None: + return + comment = self._pending_reject_comment + self._do_reject(item, comment) + + @work(thread=True) + def _do_reject(self, item: QueueItem, comment: str) -> None: + """Reject the item in a worker thread.""" + self.app.call_from_thread( + self._set_status, f"⏳ Rejecting {item.display_name}…", "busy" + ) + try: + self.lp_queue.reject(item, comment) + self.app.call_from_thread( + self._set_status, f"✔ Rejected {item.display_name}", "success" + ) + self.queue_items = self.lp_queue.get_queue_items(self.queue_status) + self.app.call_from_thread(self._populate_table) + except Exception as exc: + self.app.call_from_thread(self._set_status, f"❌ Error: {exc}", "error") + + +def main() -> None: + """Run the TUI application.""" + app = QueueApp() + app.run() + + +if __name__ == "__main__": + main() diff --git a/tools/lp-queue/lp_queue/launchpad.py b/tools/lp-queue/lp_queue/launchpad.py new file mode 100644 index 0000000..1716277 --- /dev/null +++ b/tools/lp-queue/lp_queue/launchpad.py @@ -0,0 +1,399 @@ +"""Launchpad API interactions for the upload queue.""" + +from __future__ import annotations + +import subprocess +import tempfile +import urllib.error +import urllib.request +from collections.abc import Callable +from dataclasses import dataclass +from pathlib import Path + +# Default Ubuntu series to operate on. +DEFAULT_SERIES = "resolute" + +# Launchpad queue status strings expected by the API. +QUEUE_STATUS_NEW = "New" +QUEUE_STATUS_UNAPPROVED = "Unapproved" +QUEUE_STATUS_ACCEPTED = "Accepted" +QUEUE_STATUS_DONE = "Done" +QUEUE_STATUS_REJECTED = "Rejected" + +# All queue statuses in display order. +QUEUE_STATUSES = [ + QUEUE_STATUS_NEW, + QUEUE_STATUS_UNAPPROVED, + QUEUE_STATUS_ACCEPTED, + QUEUE_STATUS_DONE, + QUEUE_STATUS_REJECTED, +] + + +@dataclass +class QueueItem: + """Represents a single item in the upload queue.""" + + source_name: str + version: str + component: str + section: str + archive_url: str + date_created: str + status: str + is_sync: bool + changes_file_url: str | None + authors: str = "" + lp_item: object | None = None + + @property + def display_name(self) -> str: + """Return a display-friendly name for the item.""" + return f"{self.source_name}/{self.version}" + + +class LaunchpadQueue: + """Interface to the Launchpad upload queue.""" + + def __init__(self, series: str = DEFAULT_SERIES): + self.series = series + self._lp = None + self._ubuntu = None + self._archive = None + self._series = None + self._debian_archive = None + self._log_callback: Callable[[str], None] | None = None + self.work_dir = Path().home() / "lp-queue" + self.work_dir.mkdir(parents=True, exist_ok=True) + + def set_log_callback(self, callback: Callable[[str], None]) -> None: + """Register a callback to receive debug log messages. + + Args: + callback: A callable that accepts a single log-message string. + """ + self._log_callback = callback + + def _log(self, message: str) -> None: + """Send *message* to the registered log callback (if any).""" + if self._log_callback is not None: + self._log_callback(message) + + def connect(self) -> None: + """Authenticate and connect to Launchpad. + + Uses launchpadlib to obtain OAuth credentials and connect + to the production Launchpad instance. The first run will open a + browser window for OAuth authorization; subsequent runs reuse the + stored credentials. + """ + from launchpadlib.launchpad import Launchpad + + self._lp = Launchpad.login_with( + "lp-queue-tui", + "production", + version="devel", + ) + self._ubuntu = self._lp.distributions["ubuntu"] + self._archive = self._ubuntu.main_archive + self._series = self._ubuntu.getSeries(name_or_version=self.series) + + def lp_user_name(self) -> str: + return self._lp.me.name + + def get_queue_items(self, status: str = QUEUE_STATUS_UNAPPROVED) -> list[QueueItem]: + """Fetch all items in the upload queue for the configured series. + + Args: + status: The Launchpad queue status string. Defaults to Unapproved. + + Returns: + A list of QueueItem dataclass instances. + + """ + self._log(f"series.getPackageUploads(status={status!r})") + uploads = self._series.getPackageUploads(status=status) + items: list[QueueItem] = [] + for upload in uploads: + name = upload.package_name or upload.display_name + version = upload.package_version or "" + is_sync = _is_sync(upload) + authors = _build_authors(upload, is_sync) + item = QueueItem( + source_name=name, + version=version, + component=upload.component_name or "", + section=upload.section_name or "", + archive_url=upload.self_link, + date_created=str(upload.date_created), + status=upload.status, + is_sync=is_sync, + changes_file_url=upload.changes_file_url, + authors=authors, + lp_item=upload, + ) + items.append(item) + self._log("Adding %s" % item) + return items + + def get_debdiff(self, item: QueueItem) -> str: + """Compute the debdiff for a queue item. + + Compares the currently published version in the archive with the + new version in the queue. Falls back to displaying the changes + file content when no previous version exists or when debdiff is + not installed. + + For sync'd packages whose source files are not available via + Launchpad, the new version is fetched from the Debian archive. + + Args: + item: The queue item to get the debdiff for. + + Returns: + The debdiff output as a string, or the changes file content. + + """ + current = self._get_current_source(item.source_name) + if current is None: + return self._get_changes_content(item) + + self._log( + f"Found current source package: {current.source_package_name}/{current.source_package_version}" + ) + + work_dir = self.work_dir / item.source_name + work_dir.mkdir(parents=True, exist_ok=True) + work_dir = str(work_dir) + + old_dsc = _download_source_files(current.sourceFileUrls(), work_dir) + + if not item.is_sync: + # Try LP source files first; silently fall back for syncs. + try: + self._log(f"Fetching {item.display_name} from Ubuntu archive") + new_dsc = _download_source_files(item.lp_item.sourceFileUrls(), work_dir) + except (OSError, urllib.error.URLError) as e: + self._log("LP source files unavailable for %s (%s)" % (item.display_name, e)) + new_dsc = None + + # For synced packages: fetch from the Debian archive via LP. + if item.is_sync: + try: + self._log(f"Fetching {item.display_name} from Debian archive") + new_dsc = self._get_debian_source(item.source_name, item.version, work_dir) + except (OSError, urllib.error.URLError) as e: + self._log("LP source files unavailable for %s (%s)" % (item.display_name, e)) + new_dsc = None + + if old_dsc is None or new_dsc is None: + return self._get_changes_content(item) + return self._run_debdiff(old_dsc, new_dsc) + + def _run_debdiff(self, old_dsc: str, new_dsc: str) -> str: + """Run some diff between two .dsc files and return the output.""" + # Get a diffoscope HTML output for very rich diffing + try: + output = self.work_dir / ( + Path(old_dsc).stem + "_" + Path(new_dsc).stem + "_debdiff.html" + ) + subprocess.run(["diffoscope", "--html", str(output), old_dsc, new_dsc], timeout=120) + subprocess.run(["xdg-open", str(output)], timeout=120) + except FileNotFoundError: + self._log( + "(diffoscope not found — install at least the 'diffoscope-minimal' package: sudo apt install diffoscope-minimal)" + ) + # Display debdiff in the main console output + try: + result = subprocess.run( + ["debdiff", old_dsc, new_dsc], + capture_output=True, + text=True, + timeout=120, + ) + # debdiff returns 0 for no diff, 1 for diff found (both are OK) + if result.returncode <= 1: + return result.stdout or "(no differences found)" + return result.stderr or f"(debdiff exited with code {result.returncode})" + except FileNotFoundError: + return "(debdiff not found — install the 'devscripts' package: sudo apt install devscripts)" + except subprocess.TimeoutExpired: + return "(debdiff timed out after 120 seconds)" + + def accept(self, item: QueueItem) -> None: + """Accept a queue item. + + Args: + item: The queue item to accept. + + """ + self._log(f"item.acceptFromQueue() [{item.display_name}]") + item.lp_item.acceptFromQueue() + + def reject(self, item: QueueItem, comment: str) -> None: + """Reject a queue item with a comment. + + Args: + item: The queue item to reject. + comment: The rejection reason/comment. + + """ + self._log(f"item.rejectFromQueue(comment=...) [{item.display_name}]") + item.lp_item.rejectFromQueue(comment=comment) + + def get_all_series(self) -> list[tuple[str, str, str]]: + """Return all Ubuntu series as ``(name, version, status)`` tuples.""" + result: list[tuple[str, str, str]] = [] + for s in self._ubuntu.series: + self._log(f"{s.name}: {s.status}") + if s.status != "Obsolete": + result.append((s.name, s.version, s.status)) + return result + + def switch_series(self, series_name: str) -> None: + """Change the active series to *series_name*. + + After calling this the next :meth:`get_queue_items` call will + operate on the new series. + + """ + self._series = self._ubuntu.getSeries(name_or_version=series_name) + self.series = series_name + + def get_debian_tracker_url(self, source_name: str) -> str: + """Return the Debian tracker URL for a source package. + + Args: + source_name: The source package name. + + Returns: + The URL to the Debian tracker page. + + """ + return f"https://tracker.debian.org/pkg/{source_name}" + + def _get_current_source(self, source_name: str) -> object | None: + """Return the currently published source in the archive, or None.""" + sources = self._archive.getPublishedSources( + source_name=source_name, + exact_match=True, + status="Published", + distro_series=self._series, + ) + return sources[0] if sources else None + + def _get_changes_content(self, item: QueueItem) -> str: + """Fetch and return the changes file content for an item. + + This is used as a fallback when debdiff is not possible (e.g., + new package with no previous version in the archive). + """ + if item.changes_file_url is None: + return "(no changes file available)" + try: + with urllib.request.urlopen(item.changes_file_url) as resp: # noqa: S310 + return resp.read().decode("utf-8", errors="replace") + except Exception as e: + self._log("Failed to fetch changes file (%s)" % e) + return "(failed to fetch changes file)" + + def _ensure_debian_archive(self) -> None: + """Lazily initialise the Debian archive reference via Launchpad.""" + if self._debian_archive is None: + debian = self._lp.distributions["debian"] + self._debian_archive = debian.main_archive + + def _get_debian_source(self, source_name: str, version: str, dest: str) -> str | None: + """Download source files for *version* from Debian via the Launchpad API. + + Uses ``lp.distributions["debian"].main_archive.getPublishedSources()`` + to locate the published source package, then downloads the files + referenced by ``sourceFileUrls()``. + + Returns: + The local path to the downloaded ``.dsc``, or ``None`` on failure. + + """ + self._ensure_debian_archive() + try: + sources = self._debian_archive.getPublishedSources( + source_name=source_name, + version=version, + exact_match=True, + status="Published", + ) + except Exception: + self._log("Failed to query Debian archive for %s %s" % (source_name, version)) + return None + + if not sources: + self._log("No published Debian source found for %s %s" % (source_name, version)) + return None + + try: + urls = sources[0].sourceFileUrls() + except Exception as e: + self._log("Failed to get source file URLs for %s %s (%s)" % (source_name, version, e)) + return None + + return _download_source_files(urls, dest) + + +def _is_sync(upload: object) -> bool: + """Detect whether an upload is a sync from Debian.""" + return upload.contains_copy + + +def _extract_lp_username(link: str | None) -> str | None: + """Extract a Launchpad username from an API person link. + + Launchpad person links have the form + ``https://api.launchpad.net/devel/~username``. + + Returns: + The username string, or ``None`` when *link* is ``None`` or does + not contain the expected ``/~`` separator. + + """ + if link and "/~" in link: + return link.rsplit("/~", 1)[-1] + return None + + +def _build_authors(upload: object, is_sync: bool) -> str: + """Build the "Author(s)" string for a queue upload. + + For syncs the requestor is shown. For regular uploads the signer + (uploader) is shown, plus the sponsor when the two differ. + + """ + if is_sync: + requestor = _extract_lp_username( + getattr(upload, "package_copy_requestor_link", None), + ) + return requestor or "" + + signer = _extract_lp_username( + getattr(upload, "signing_key_owner_link", None), + ) + sponsor = _extract_lp_username( + getattr(upload, "sponsor_link", None), + ) + + if signer and sponsor and signer != sponsor: + return f"{signer}, sponsor: {sponsor}" + return signer or "" + + +def _download_source_files(urls: list[str], dest: str) -> str | None: + """Download source files and return the path to the .dsc file.""" + dsc_path = None + for url in urls: + filename = url.rsplit("/", 1)[-1] + filepath = Path(dest) / filename + if filename.endswith(".dsc"): + dsc_path = str(filepath) + if filepath.exists(): + continue + urllib.request.urlretrieve(url, filepath) # noqa: S310 + return dsc_path diff --git a/tools/lp-queue/pyproject.toml b/tools/lp-queue/pyproject.toml new file mode 100644 index 0000000..4c3f4a9 --- /dev/null +++ b/tools/lp-queue/pyproject.toml @@ -0,0 +1,45 @@ +[project] +name = "lp-queue" +version = "0.1.0" +description = "A TUI tool for managing the Ubuntu Launchpad upload queue" +requires-python = ">=3.10" +dependencies = [ + "launchpadlib", + "textual>=3.0.0", +] + +[project.optional-dependencies] +dev = [ + "pytest", + "pytest-asyncio", + "ruff", + "textual-dev", +] + +[project.scripts] +lp-queue = "lp_queue.app:main" + +[tool.ruff] +line-length = 99 + +[tool.ruff.lint] +select = ["E", "W", "F", "C", "N", "D", "I001"] +extend-ignore = [ + "D100", + "D101", + "D102", + "D103", + "D104", + "D105", + "D107", + "D203", + "D213", +] + +[tool.ruff.lint.per-file-ignores] +"tests/*" = ["D"] + +[tool.pytest.ini_options] +minversion = "6.0" +log_cli_level = "INFO" +asyncio_mode = "auto" diff --git a/tools/lp-queue/tests/__init__.py b/tools/lp-queue/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tools/lp-queue/tests/test_app.py b/tools/lp-queue/tests/test_app.py new file mode 100644 index 0000000..f7a8afe --- /dev/null +++ b/tools/lp-queue/tests/test_app.py @@ -0,0 +1,556 @@ +"""Tests for the TUI application.""" + +from unittest.mock import MagicMock + +from textual.widgets import DataTable, OptionList, RichLog + +from lp_queue.app import ( + ConfirmScreen, + QueueApp, + QueueStatusScreen, + RejectScreen, + ReviewScreen, + SeriesScreen, +) +from lp_queue.launchpad import QUEUE_STATUS_UNAPPROVED, LaunchpadQueue, QueueItem + + +def _make_item(**overrides): + """Create a QueueItem with sensible defaults.""" + defaults = { + "source_name": "hello", + "version": "2.10-3", + "component": "main", + "section": "devel", + "archive_url": "", + "date_created": "2025-01-01", + "status": "Unapproved", + "is_sync": False, + "changes_file_url": None, + "authors": "", + "lp_item": MagicMock(), + } + defaults.update(overrides) + return QueueItem(**defaults) + + +class TestQueueAppTable: + """Tests for the main QueueApp table population.""" + + async def test_populate_table(self): + """Test that _populate_table fills the DataTable correctly.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + app.queue_items = [ + _make_item(source_name="hello", version="2.10-3"), + _make_item(source_name="bash", version="5.2-1ubuntu1", is_sync=False), + ] + app._populate_table() + table = app.query_one(DataTable) + assert table.row_count == 2 + + async def test_empty_queue(self): + """Test that an empty queue displays zero rows.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + app.queue_items = [] + app._populate_table() + table = app.query_one(DataTable) + assert table.row_count == 0 + + async def test_get_selected_item_empty(self): + """Test that _get_selected_item returns None for an empty table.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + result = app._get_selected_item() + assert result is None + + +class TestReviewScreen: + """Tests for the ReviewScreen modal.""" + + async def test_review_screen_displays_content(self): + """Test that the ReviewScreen shows the debdiff text.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + item = _make_item() + screen = ReviewScreen(item, "--- a/file\n+++ b/file\n") + app.push_screen(screen) + await _pilot.pause() + assert app.screen is screen + + async def test_review_screen_uses_diff_highlighter(self): + """Test that the ReviewScreen writes content as a Syntax renderable with diff lexer.""" + from unittest.mock import patch + + from rich.syntax import Syntax + + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + item = _make_item() + debdiff = "--- a/file\n+++ b/file\n@@ -1 +1 @@\n-old\n+new\n" + screen = ReviewScreen(item, debdiff) + written_objects = [] + original_write = RichLog.write + + def capture_write(self_log, content, *args, **kwargs): + written_objects.append(content) + return original_write(self_log, content, *args, **kwargs) + + with patch.object(RichLog, "write", capture_write): + app.push_screen(screen) + await _pilot.pause() + + assert len(written_objects) >= 1 + assert isinstance(written_objects[0], Syntax) + assert written_objects[0].lexer.name == "Diff" + + +class TestRejectScreen: + """Tests for the RejectScreen modal.""" + + async def test_reject_screen_dismiss_on_escape(self): + """Test that pressing escape dismisses the reject screen.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + item = _make_item() + screen = RejectScreen(item) + app.push_screen(screen) + await _pilot.pause() + assert app.screen is screen + await _pilot.press("escape") + await _pilot.pause() + assert app.screen is not screen + + +class TestDebugPanel: + """Tests for the toggleable debug log panel.""" + + async def test_debug_panel_hidden_by_default(self): + """Test that the debug panel is hidden on startup.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + panel = app.query_one("#debug-panel") + assert "visible" not in panel.classes + + async def test_toggle_debug_panel(self): + """Test that pressing ~ toggles the debug panel visibility.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + panel = app.query_one("#debug-panel") + assert "visible" not in panel.classes + + # Toggle on + await _pilot.press("~") + await _pilot.pause() + assert "visible" in panel.classes + + # Toggle off + await _pilot.press("~") + await _pilot.pause() + assert "visible" not in panel.classes + + async def test_write_debug_log(self): + """Test that _write_debug_log appends to the debug RichLog.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + # Make the panel visible so RichLog knows its size and renders immediately + app.action_toggle_debug() + await _pilot.pause() + log = app.query_one("#debug-log", RichLog) + initial = len(log.lines) + app._write_debug_log("test log message") + await _pilot.pause() + assert len(log.lines) > initial + + async def test_log_callback_registered(self): + """Test that the log callback is set on LaunchpadQueue during mount.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + assert lp._log_callback is not None + + +class TestConfirmScreen: + """Tests for the ConfirmScreen modal.""" + + async def test_confirm_with_y(self): + """Test that pressing 'y' confirms and dismisses the screen.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + results = [] + screen = ConfirmScreen("Title", "Are you sure?") + app.push_screen(screen, results.append) + await _pilot.pause() + assert app.screen is screen + await _pilot.press("y") + await _pilot.pause() + assert app.screen is not screen + assert results == [True] + + async def test_cancel_with_n(self): + """Test that pressing 'n' cancels and dismisses the screen.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + results = [] + screen = ConfirmScreen("Title", "Are you sure?") + app.push_screen(screen, results.append) + await _pilot.pause() + assert app.screen is screen + await _pilot.press("n") + await _pilot.pause() + assert app.screen is not screen + assert results == [False] + + async def test_cancel_with_escape(self): + """Test that pressing escape cancels and dismisses the screen.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + results = [] + screen = ConfirmScreen("Title", "Are you sure?") + app.push_screen(screen, results.append) + await _pilot.pause() + assert app.screen is screen + await _pilot.press("escape") + await _pilot.pause() + assert app.screen is not screen + assert results == [False] + + +class TestAcceptConfirmation: + """Tests for the accept confirmation flow.""" + + async def test_accept_shows_confirm(self): + """Test that action_accept pushes a ConfirmScreen when an item is selected.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + app.queue_items = [_make_item(source_name="hello", version="2.10-3")] + app._populate_table() + app.action_accept() + await _pilot.pause() + assert isinstance(app.screen, ConfirmScreen) + + async def test_accept_cancelled(self): + """Test that cancelling accept does not proceed.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + app.queue_items = [_make_item(source_name="hello", version="2.10-3")] + app._populate_table() + app.action_accept() + await _pilot.pause() + assert isinstance(app.screen, ConfirmScreen) + await _pilot.press("n") + await _pilot.pause() + assert not isinstance(app.screen, ConfirmScreen) + + +class TestRejectConfirmation: + """Tests for the reject confirmation flow.""" + + async def test_reject_shows_reject_screen_first(self): + """Test that action_reject pushes RejectScreen first.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + app.queue_items = [_make_item(source_name="hello", version="2.10-3")] + app._populate_table() + app.action_reject() + await _pilot.pause() + assert isinstance(app.screen, RejectScreen) + + async def test_reject_shows_confirm_after_comment(self): + """Test that submitting a comment shows ConfirmScreen.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + app.queue_items = [_make_item(source_name="hello", version="2.10-3")] + app._populate_table() + app.action_reject() + await _pilot.pause() + assert isinstance(app.screen, RejectScreen) + await _pilot.press("F", "T", "B", "F", "S") + await _pilot.press("enter") + await _pilot.pause() + assert isinstance(app.screen, ConfirmScreen) + + async def test_reject_cancelled_at_confirm(self): + """Test that cancelling at ConfirmScreen does not reject.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + app.queue_items = [_make_item(source_name="hello", version="2.10-3")] + app._populate_table() + app.action_reject() + await _pilot.pause() + await _pilot.press("F", "T", "B", "F", "S") + await _pilot.press("enter") + await _pilot.pause() + assert isinstance(app.screen, ConfirmScreen) + await _pilot.press("n") + await _pilot.pause() + assert not isinstance(app.screen, ConfirmScreen) + + +class TestSeriesScreen: + """Tests for the SeriesScreen modal.""" + + async def test_series_screen_displays_options(self): + """Test that SeriesScreen renders the series list.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + series_list = [ + ("resolute", "26.04", True), + ("noble", "24.04", True), + ("jammy", "22.04", True), + ("focal", "20.04", False), + ] + screen = SeriesScreen(series_list, "resolute") + app.push_screen(screen) + await _pilot.pause() + assert app.screen is screen + option_list = screen.query_one(OptionList) + assert option_list.option_count == 4 + + async def test_series_screen_dismiss_on_escape(self): + """Test that pressing escape cancels the series screen.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + results = [] + series_list = [("resolute", "26.04", True)] + screen = SeriesScreen(series_list, "resolute") + app.push_screen(screen, results.append) + await _pilot.pause() + assert app.screen is screen + await _pilot.press("escape") + await _pilot.pause() + assert app.screen is not screen + assert results == [None] + + async def test_series_screen_current_marker(self): + """Test that the current series is marked with a marker character.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + series_list = [ + ("resolute", "26.04", True), + ("noble", "24.04", True), + ] + screen = SeriesScreen(series_list, "resolute") + app.push_screen(screen) + await _pilot.pause() + option_list = screen.query_one(OptionList) + first_option = option_list.get_option_at_index(0) + second_option = option_list.get_option_at_index(1) + assert "resolute" in str(first_option.prompt) + assert "noble" in str(second_option.prompt) + + async def test_handle_series_result_cancel(self): + """Test that None result from series screen sets cancel status.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + app._handle_series_result(None) + + async def test_handle_series_result_same_series(self): + """Test that selecting the same series shows 'already on' message.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + app._handle_series_result("resolute") + + +class TestQueueStatusScreen: + """Tests for the QueueStatusScreen modal.""" + + async def test_queue_status_screen_displays_options(self): + """Test that QueueStatusScreen renders all queue statuses.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + screen = QueueStatusScreen("Unapproved") + app.push_screen(screen) + await _pilot.pause() + assert app.screen is screen + option_list = screen.query_one(OptionList) + assert option_list.option_count == 5 + + async def test_queue_status_screen_dismiss_on_escape(self): + """Test that pressing escape cancels the queue status screen.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + results = [] + screen = QueueStatusScreen("Unapproved") + app.push_screen(screen, results.append) + await _pilot.pause() + assert app.screen is screen + await _pilot.press("escape") + await _pilot.pause() + assert app.screen is not screen + assert results == [None] + + async def test_queue_status_screen_current_marker(self): + """Test that the current queue status is marked with a marker character.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + screen = QueueStatusScreen("Unapproved") + app.push_screen(screen) + await _pilot.pause() + option_list = screen.query_one(OptionList) + # The second option (index 1) should be Unapproved with the marker + unapproved_option = option_list.get_option_at_index(1) + assert "✦" in str(unapproved_option.prompt) + # The first option (index 0) should be New without the marker + new_option = option_list.get_option_at_index(0) + assert "✦" not in str(new_option.prompt) + + async def test_default_queue_status_is_unapproved(self): + """Test that the default queue status is Unapproved.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + assert app.queue_status == QUEUE_STATUS_UNAPPROVED + + async def test_handle_queue_status_result_cancel(self): + """Test that None result from queue status screen sets cancel status.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + app._handle_queue_status_result(None) + assert app.queue_status == QUEUE_STATUS_UNAPPROVED + + async def test_handle_queue_status_result_same_status(self): + """Test that selecting the same status shows 'already showing' message.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + app._handle_queue_status_result("Unapproved") + assert app.queue_status == QUEUE_STATUS_UNAPPROVED + + async def test_handle_queue_status_result_changes_status(self): + """Test that selecting a different status updates queue_status.""" + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + app._handle_queue_status_result("New") + assert app.queue_status == "New" + + +class TestAutoRefresh: + """Tests for the automatic queue refresh timer.""" + + async def test_periodic_refresh_calls_load_queue_on_main_screen(self): + """Test that _periodic_refresh triggers _load_queue when on the main screen.""" + from unittest.mock import patch + + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + with patch.object(app, "_load_queue") as mock_load: + app._periodic_refresh() + mock_load.assert_called_once() + + async def test_periodic_refresh_skips_when_modal_open(self): + """Test that _periodic_refresh does not refresh when a modal screen is open.""" + from unittest.mock import patch + + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + screen = ConfirmScreen("Title", "Message") + app.push_screen(screen) + await _pilot.pause() + with patch.object(app, "_load_queue") as mock_load: + app._periodic_refresh() + mock_load.assert_not_called() + + async def test_periodic_refresh_skips_when_review_screen_open(self): + """Test that _periodic_refresh does not refresh when ReviewScreen is open.""" + from unittest.mock import patch + + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + item = _make_item() + screen = ReviewScreen(item, "--- diff content ---") + app.push_screen(screen) + await _pilot.pause() + with patch.object(app, "_load_queue") as mock_load: + app._periodic_refresh() + mock_load.assert_not_called() + + async def test_periodic_refresh_resumes_after_modal_dismissed(self): + """Test that _periodic_refresh works again after a modal is dismissed.""" + from unittest.mock import patch + + lp = LaunchpadQueue() + app = QueueApp(lp_queue=lp) + + async with app.run_test(size=(120, 30)) as _pilot: + screen = ConfirmScreen("Title", "Message") + app.push_screen(screen) + await _pilot.pause() + + # Should not refresh while modal is open + with patch.object(app, "_load_queue") as mock_load: + app._periodic_refresh() + mock_load.assert_not_called() + + # Dismiss the modal + await _pilot.press("escape") + await _pilot.pause() + + # Should refresh now that we're back to the main screen + with patch.object(app, "_load_queue") as mock_load: + app._periodic_refresh() + mock_load.assert_called_once() diff --git a/tools/lp-queue/tests/test_launchpad.py b/tools/lp-queue/tests/test_launchpad.py new file mode 100644 index 0000000..01a8a90 --- /dev/null +++ b/tools/lp-queue/tests/test_launchpad.py @@ -0,0 +1,686 @@ +"""Tests for the launchpad module.""" + +import subprocess +from unittest.mock import MagicMock, patch + +from lp_queue.launchpad import ( + LaunchpadQueue, + QueueItem, + _build_authors, + _download_source_files, + _extract_lp_username, + _is_sync, +) + +# --------------------------------------------------------------------------- +# QueueItem tests +# --------------------------------------------------------------------------- + + +class TestQueueItem: + """Tests for the QueueItem dataclass.""" + + def test_display_name(self): + """Test that display_name returns name and version.""" + item = QueueItem( + source_name="hello", + version="2.10-3", + component="main", + section="devel", + archive_url="https://launchpad.net/ubuntu/+archive/primary", + date_created="2025-01-01", + status="Unapproved", + is_sync=False, + changes_file_url=None, + ) + assert item.display_name == "hello/2.10-3" + + def test_display_name_with_ubuntu_version(self): + item = QueueItem( + source_name="bash", + version="5.2-1ubuntu1", + component="main", + section="shells", + archive_url="", + date_created="2025-06-01", + status="Unapproved", + is_sync=False, + changes_file_url=None, + ) + assert item.display_name == "bash/5.2-1ubuntu1" + + +# --------------------------------------------------------------------------- +# LaunchpadQueue tests +# --------------------------------------------------------------------------- + + +class TestLaunchpadQueue: + """Tests for the LaunchpadQueue class.""" + + def test_default_series(self): + """Test that the default series is set correctly.""" + lp = LaunchpadQueue() + assert lp.series == "resolute" + + def test_custom_series(self): + """Test that a custom series can be specified.""" + lp = LaunchpadQueue(series="noble") + assert lp.series == "noble" + + def test_debian_tracker_url(self): + """Test that Debian tracker URLs are built correctly.""" + lp = LaunchpadQueue() + url = lp.get_debian_tracker_url("hello") + assert url == "https://tracker.debian.org/pkg/hello" + + def test_connect(self): + """Test that connect() calls Launchpad.login_with correctly.""" + mock_lp_class = MagicMock() + mock_lp = MagicMock() + mock_lp_class.login_with.return_value = mock_lp + + lp_mod = MagicMock(Launchpad=mock_lp_class) + with patch.dict("sys.modules", {"launchpadlib": lp_mod, "launchpadlib.launchpad": lp_mod}): + lp = LaunchpadQueue() + lp.connect() + + mock_lp_class.login_with.assert_called_once_with( + "lp-queue-tui", "production", version="devel" + ) + + def test_get_queue_items(self): + """Test get_queue_items converts LP objects to QueueItems.""" + lp = LaunchpadQueue() + + mock_upload = MagicMock() + mock_upload.package_name = "hello" + mock_upload.display_name = "hello" + mock_upload.package_version = "2.10-3ubuntu1" + mock_upload.component_name = "main" + mock_upload.section_name = "devel" + mock_upload.self_link = "https://api.launchpad.net/devel/..." + mock_upload.date_created = "2025-01-01T00:00:00+00:00" + mock_upload.status = "Unapproved" + mock_upload.changes_file_url = "https://example.com/changes" + mock_upload.contains_copy = False + mock_upload.signing_key_owner_link = "https://api.launchpad.net/devel/~uploader" + + mock_series = MagicMock() + mock_series.getPackageUploads.return_value = [mock_upload] + lp._series = mock_series + + items = lp.get_queue_items() + + assert len(items) == 1 + assert items[0].source_name == "hello" + assert items[0].version == "2.10-3ubuntu1" + assert items[0].component == "main" + assert items[0].status == "Unapproved" + assert items[0].authors == "uploader" + + def test_get_queue_items_sync_author(self): + """Test that sync uploads extract the requestor as the author.""" + lp = LaunchpadQueue() + + mock_upload = MagicMock() + mock_upload.package_name = "hello" + mock_upload.display_name = "hello" + mock_upload.package_version = "2.10-3" + mock_upload.component_name = "main" + mock_upload.section_name = "devel" + mock_upload.self_link = "https://api.launchpad.net/devel/..." + mock_upload.date_created = "2025-01-01T00:00:00+00:00" + mock_upload.status = "Unapproved" + mock_upload.changes_file_url = None + mock_upload.contains_copy = True + mock_upload.package_copy_requestor_link = "https://api.launchpad.net/devel/~syncer" + + mock_series = MagicMock() + mock_series.getPackageUploads.return_value = [mock_upload] + lp._series = mock_series + + items = lp.get_queue_items() + + assert len(items) == 1 + assert items[0].is_sync is True + assert items[0].authors == "syncer" + + def test_accept(self): + """Test accept delegates to the LP item.""" + mock_lp_item = MagicMock() + item = QueueItem( + source_name="hello", + version="2.10-3", + component="main", + section="devel", + archive_url="", + date_created="2025-01-01", + status="Unapproved", + is_sync=False, + changes_file_url=None, + lp_item=mock_lp_item, + ) + lp = LaunchpadQueue() + lp.accept(item) + mock_lp_item.acceptFromQueue.assert_called_once() + + def test_reject(self): + """Test reject delegates to the LP item with comment.""" + mock_lp_item = MagicMock() + item = QueueItem( + source_name="hello", + version="2.10-3", + component="main", + section="devel", + archive_url="", + date_created="2025-01-01", + status="Unapproved", + is_sync=False, + changes_file_url=None, + lp_item=mock_lp_item, + ) + lp = LaunchpadQueue() + lp.reject(item, "FTBFS on amd64") + mock_lp_item.rejectFromQueue.assert_called_once_with(comment="FTBFS on amd64") + + def test_get_changes_content_no_url(self): + """Test _get_changes_content with no URL returns a message.""" + lp = LaunchpadQueue() + item = QueueItem( + source_name="hello", + version="2.10-3", + component="main", + section="devel", + archive_url="", + date_created="", + status="Unapproved", + is_sync=False, + changes_file_url=None, + ) + result = lp._get_changes_content(item) + assert "(no changes file available)" in result + + def test_log_callback(self): + """Test that _log invokes the registered callback.""" + lp = LaunchpadQueue() + messages: list[str] = [] + lp.set_log_callback(messages.append) + lp._log("hello") + assert messages == ["hello"] + + def test_log_no_callback(self): + """Test that _log is a no-op when no callback is registered.""" + lp = LaunchpadQueue() + # Should not raise + lp._log("ignored") + + def test_get_all_series(self): + """Test get_all_series returns non-Obsolete series tuples.""" + lp = LaunchpadQueue() + + mock_series_1 = MagicMock() + mock_series_1.name = "resolute" + mock_series_1.version = "26.04" + mock_series_1.status = "Active Development" + + mock_series_2 = MagicMock() + mock_series_2.name = "noble" + mock_series_2.version = "24.04" + mock_series_2.status = "Supported" + + mock_series_3 = MagicMock() + mock_series_3.name = "trusty" + mock_series_3.version = "14.04" + mock_series_3.status = "Obsolete" + + mock_ubuntu = MagicMock() + mock_ubuntu.series = [mock_series_1, mock_series_2, mock_series_3] + lp._ubuntu = mock_ubuntu + + result = lp.get_all_series() + + # Obsolete series are filtered out + assert len(result) == 2 + assert result[0] == ("resolute", "26.04", "Active Development") + assert result[1] == ("noble", "24.04", "Supported") + + def test_switch_series(self): + """Test switch_series updates the active series.""" + lp = LaunchpadQueue() + + mock_ubuntu = MagicMock() + mock_new_series = MagicMock() + mock_ubuntu.getSeries.return_value = mock_new_series + lp._ubuntu = mock_ubuntu + + lp.switch_series("noble") + + mock_ubuntu.getSeries.assert_called_once_with(name_or_version="noble") + assert lp.series == "noble" + assert lp._series is mock_new_series + + +# --------------------------------------------------------------------------- +# Helper function tests +# --------------------------------------------------------------------------- + + +class TestIsSyncHelper: + """Tests for the _is_sync helper.""" + + def test_ubuntu_version_sync(self): + upload = MagicMock() + upload.contains_copy = True + assert _is_sync(upload) is True + + def test_ubuntu_version_not_sync(self): + upload = MagicMock() + upload.contains_copy = False + assert _is_sync(upload) is False + + +class TestExtractLpUsername: + """Tests for the _extract_lp_username helper.""" + + def test_extracts_username(self): + link = "https://api.launchpad.net/devel/~johndoe" + assert _extract_lp_username(link) == "johndoe" + + def test_none_returns_none(self): + assert _extract_lp_username(None) is None + + def test_no_tilde_returns_none(self): + assert _extract_lp_username("https://example.com/nousername") is None + + def test_empty_string_returns_none(self): + assert _extract_lp_username("") is None + + def test_username_with_hyphen(self): + link = "https://api.launchpad.net/devel/~john-doe" + assert _extract_lp_username(link) == "john-doe" + + +class TestBuildAuthors: + """Tests for the _build_authors helper.""" + + def test_sync_with_requestor(self): + upload = MagicMock(spec=[]) + upload.package_copy_requestor_link = ( + "https://api.launchpad.net/devel/~sync-requester" + ) + assert _build_authors(upload, is_sync=True) == "sync-requester" + + def test_sync_without_requestor(self): + upload = MagicMock(spec=[]) + assert _build_authors(upload, is_sync=True) == "" + + def test_regular_upload_signer_only(self): + upload = MagicMock(spec=[]) + upload.signing_key_owner_link = ( + "https://api.launchpad.net/devel/~uploader" + ) + assert _build_authors(upload, is_sync=False) == "uploader" + + def test_regular_upload_with_sponsor(self): + upload = MagicMock(spec=[]) + upload.signing_key_owner_link = ( + "https://api.launchpad.net/devel/~uploader" + ) + upload.sponsor_link = ( + "https://api.launchpad.net/devel/~sponsor-dev" + ) + assert _build_authors(upload, is_sync=False) == "uploader, sponsor: sponsor-dev" + + def test_regular_upload_signer_equals_sponsor(self): + upload = MagicMock(spec=[]) + upload.signing_key_owner_link = ( + "https://api.launchpad.net/devel/~same-person" + ) + upload.sponsor_link = ( + "https://api.launchpad.net/devel/~same-person" + ) + assert _build_authors(upload, is_sync=False) == "same-person" + + def test_regular_upload_no_signer(self): + upload = MagicMock(spec=[]) + assert _build_authors(upload, is_sync=False) == "" + + +class TestRunDebdiff: + """Tests for the LaunchpadQueue._run_debdiff instance method.""" + + @patch("lp_queue.launchpad.subprocess.run") + def test_success(self, mock_run, tmp_path): + mock_run.return_value = MagicMock(returncode=0, stdout="diff output", stderr="") + lp = LaunchpadQueue() + lp.work_dir = tmp_path + assert lp._run_debdiff("/a.dsc", "/b.dsc") == "diff output" + + @patch("lp_queue.launchpad.subprocess.run") + def test_diff_found(self, mock_run, tmp_path): + mock_run.return_value = MagicMock(returncode=1, stdout="differences", stderr="") + lp = LaunchpadQueue() + lp.work_dir = tmp_path + assert lp._run_debdiff("/a.dsc", "/b.dsc") == "differences" + + @patch("lp_queue.launchpad.subprocess.run") + def test_error(self, mock_run, tmp_path): + mock_run.return_value = MagicMock(returncode=2, stdout="", stderr="error msg") + lp = LaunchpadQueue() + lp.work_dir = tmp_path + assert "error msg" in lp._run_debdiff("/a.dsc", "/b.dsc") + + @patch("lp_queue.launchpad.subprocess.run", side_effect=FileNotFoundError) + def test_not_installed(self, mock_run, tmp_path): + lp = LaunchpadQueue() + lp.work_dir = tmp_path + result = lp._run_debdiff("/a.dsc", "/b.dsc") + assert "devscripts" in result + + @patch("lp_queue.launchpad.subprocess.run") + def test_diffoscope_missing_debdiff_succeeds(self, mock_run, tmp_path): + # diffoscope not found, but debdiff works fine + mock_run.side_effect = [ + FileNotFoundError, # diffoscope + MagicMock(returncode=0, stdout="diff output", stderr=""), # debdiff + ] + lp = LaunchpadQueue() + lp.work_dir = tmp_path + result = lp._run_debdiff("/a.dsc", "/b.dsc") + assert result == "diff output" + + @patch("lp_queue.launchpad.subprocess.run") + def test_timeout(self, mock_run, tmp_path): + # First call is diffoscope (succeeds), second is debdiff (times out) + mock_run.side_effect = [ + MagicMock(returncode=0), # diffoscope + MagicMock(returncode=0), # xdg-open + subprocess.TimeoutExpired(cmd="debdiff", timeout=120), # debdiff + ] + lp = LaunchpadQueue() + lp.work_dir = tmp_path + result = lp._run_debdiff("/a.dsc", "/b.dsc") + assert "timed out" in result + + +class TestDownloadSourceFiles: + """Tests for the _download_source_files helper.""" + + @patch("lp_queue.launchpad.urllib.request.urlretrieve") + def test_returns_dsc_path(self, mock_urlretrieve, tmp_path): + urls = [ + "https://example.com/hello_2.10-3.dsc", + "https://example.com/hello_2.10.orig.tar.gz", + ] + result = _download_source_files(urls, str(tmp_path)) + assert result is not None + assert result.endswith(".dsc") + + @patch("lp_queue.launchpad.urllib.request.urlretrieve") + def test_no_dsc(self, mock_urlretrieve, tmp_path): + urls = ["https://example.com/hello_2.10.orig.tar.gz"] + result = _download_source_files(urls, str(tmp_path)) + assert result is None + + +class TestGetDebianSource: + """Tests for the LaunchpadQueue._get_debian_source method.""" + + def test_success(self, tmp_path): + """Successfully fetch source files from Debian via LP API.""" + lp = LaunchpadQueue() + + mock_lp = MagicMock() + mock_debian = MagicMock() + mock_debian_archive = MagicMock() + mock_lp.distributions.__getitem__.return_value = mock_debian + mock_debian.main_archive = mock_debian_archive + + mock_source = MagicMock() + mock_source.sourceFileUrls.return_value = [ + "https://example.com/hello_2.10.orig.tar.gz", + "https://example.com/hello_2.10-3.debian.tar.xz", + "https://example.com/hello_2.10-3.dsc", + ] + mock_debian_archive.getPublishedSources.return_value = [mock_source] + lp._lp = mock_lp + + with patch("lp_queue.launchpad.urllib.request.urlretrieve"): + result = lp._get_debian_source("hello", "2.10-3", str(tmp_path)) + + assert result is not None + assert result.endswith("hello_2.10-3.dsc") + mock_debian_archive.getPublishedSources.assert_called_once_with( + source_name="hello", + version="2.10-3", + exact_match=True, + status="Published", + ) + + def test_no_sources_found(self, tmp_path): + """Return None when no Debian source is found.""" + lp = LaunchpadQueue() + + mock_lp = MagicMock() + mock_debian = MagicMock() + mock_debian_archive = MagicMock() + mock_lp.distributions.__getitem__.return_value = mock_debian + mock_debian.main_archive = mock_debian_archive + mock_debian_archive.getPublishedSources.return_value = [] + lp._lp = mock_lp + + result = lp._get_debian_source("nonexistent", "1.0-1", str(tmp_path)) + assert result is None + + def test_api_error(self, tmp_path): + """Return None when the LP API call raises an exception.""" + lp = LaunchpadQueue() + + mock_lp = MagicMock() + mock_debian = MagicMock() + mock_debian_archive = MagicMock() + mock_lp.distributions.__getitem__.return_value = mock_debian + mock_debian.main_archive = mock_debian_archive + mock_debian_archive.getPublishedSources.side_effect = Exception("API error") + lp._lp = mock_lp + + result = lp._get_debian_source("hello", "2.10-3", str(tmp_path)) + assert result is None + + def test_source_file_urls_error(self, tmp_path): + """Return None when sourceFileUrls() raises an exception.""" + lp = LaunchpadQueue() + + mock_lp = MagicMock() + mock_debian = MagicMock() + mock_debian_archive = MagicMock() + mock_lp.distributions.__getitem__.return_value = mock_debian + mock_debian.main_archive = mock_debian_archive + + mock_source = MagicMock() + mock_source.sourceFileUrls.side_effect = Exception("URL error") + mock_debian_archive.getPublishedSources.return_value = [mock_source] + lp._lp = mock_lp + + result = lp._get_debian_source("hello", "2.10-3", str(tmp_path)) + assert result is None + + def test_caches_debian_archive(self, tmp_path): + """The Debian archive reference is lazily cached after first use.""" + lp = LaunchpadQueue() + + mock_lp = MagicMock() + mock_debian = MagicMock() + mock_debian_archive = MagicMock() + mock_lp.distributions.__getitem__.return_value = mock_debian + mock_debian.main_archive = mock_debian_archive + mock_debian_archive.getPublishedSources.return_value = [] + lp._lp = mock_lp + + lp._get_debian_source("hello", "1.0-1", str(tmp_path)) + lp._get_debian_source("world", "2.0-1", str(tmp_path)) + + # distributions["debian"] should only be accessed once + mock_lp.distributions.__getitem__.assert_called_once_with("debian") + + def test_version_with_epoch(self, tmp_path): + """Epochs are passed through to the LP API as-is.""" + lp = LaunchpadQueue() + + mock_lp = MagicMock() + mock_debian = MagicMock() + mock_debian_archive = MagicMock() + mock_lp.distributions.__getitem__.return_value = mock_debian + mock_debian.main_archive = mock_debian_archive + mock_debian_archive.getPublishedSources.return_value = [] + lp._lp = mock_lp + + lp._get_debian_source("hello", "2:1.0-1", str(tmp_path)) + + mock_debian_archive.getPublishedSources.assert_called_once_with( + source_name="hello", + version="2:1.0-1", + exact_match=True, + status="Published", + ) + + +class TestDebdiffSyncFallback: + """Tests for the Debian archive fallback in get_debdiff.""" + + @patch("lp_queue.launchpad.subprocess.run") + @patch("lp_queue.launchpad._download_source_files") + def test_sync_fallback_used(self, mock_dl_source, mock_subproc_run, tmp_path): + """When item is a sync, Debian LP fallback is used directly.""" + lp = LaunchpadQueue() + lp.work_dir = tmp_path + + mock_current = MagicMock() + mock_current.sourceFileUrls.return_value = [ + "https://lp.example.com/hello_2.10-2ubuntu1.dsc", + ] + lp._archive = MagicMock() + lp._archive.getPublishedSources.return_value = [mock_current] + lp._series = MagicMock() + + # Set up the Debian archive mock + mock_lp_obj = MagicMock() + mock_debian = MagicMock() + mock_debian_archive = MagicMock() + mock_lp_obj.distributions.__getitem__.return_value = mock_debian + mock_debian.main_archive = mock_debian_archive + + mock_deb_source = MagicMock() + mock_deb_source.sourceFileUrls.return_value = [ + "https://example.com/hello_2.10-3.dsc", + ] + mock_debian_archive.getPublishedSources.return_value = [mock_deb_source] + lp._lp = mock_lp_obj + + mock_lp_item = MagicMock() + item = QueueItem( + source_name="hello", + version="2.10-3", + component="main", + section="devel", + archive_url="", + date_created="2025-01-01", + status="Unapproved", + is_sync=True, + changes_file_url=None, + lp_item=mock_lp_item, + ) + + # First call downloads old Ubuntu source, second call is from + # _get_debian_source which internally calls _download_source_files + # with the Debian source file URLs. + mock_dl_source.side_effect = ["/tmp/old.dsc", "/tmp/new.dsc"] + mock_subproc_run.return_value = MagicMock(returncode=0, stdout="diff output", stderr="") + + result = lp.get_debdiff(item) + + assert result == "diff output" + mock_debian_archive.getPublishedSources.assert_called_once_with( + source_name="hello", + version="2.10-3", + exact_match=True, + status="Published", + ) + + @patch("lp_queue.launchpad._download_source_files") + def test_non_sync_no_fallback(self, mock_dl_source, tmp_path): + """Non-sync packages should NOT trigger the Debian fallback.""" + lp = LaunchpadQueue() + lp.work_dir = tmp_path + + mock_current = MagicMock() + mock_current.sourceFileUrls.return_value = [] + lp._archive = MagicMock() + lp._archive.getPublishedSources.return_value = [mock_current] + lp._series = MagicMock() + + mock_lp_item = MagicMock() + mock_lp_item.sourceFileUrls.return_value = [] + item = QueueItem( + source_name="hello", + version="2.10-3ubuntu1", + component="main", + section="devel", + archive_url="", + date_created="2025-01-01", + status="Unapproved", + is_sync=False, + changes_file_url=None, + lp_item=mock_lp_item, + ) + + mock_dl_source.return_value = None + + result = lp.get_debdiff(item) + + assert "(no changes file available)" in result + + @patch("lp_queue.launchpad._download_source_files") + def test_sync_fallback_no_debian_source(self, mock_dl_source, tmp_path): + """When Debian LP API returns no published sources, falls back to changes content.""" + lp = LaunchpadQueue() + lp.work_dir = tmp_path + + mock_current = MagicMock() + mock_current.sourceFileUrls.return_value = [] + lp._archive = MagicMock() + lp._archive.getPublishedSources.return_value = [mock_current] + lp._series = MagicMock() + + # Set up the Debian archive mock that returns no sources + mock_lp_obj = MagicMock() + mock_debian = MagicMock() + mock_debian_archive = MagicMock() + mock_lp_obj.distributions.__getitem__.return_value = mock_debian + mock_debian.main_archive = mock_debian_archive + mock_debian_archive.getPublishedSources.return_value = [] + lp._lp = mock_lp_obj + + mock_lp_item = MagicMock() + item = QueueItem( + source_name="hello", + version="2.10-3", + component="main", + section="devel", + archive_url="", + date_created="2025-01-01", + status="Unapproved", + is_sync=True, + changes_file_url=None, + lp_item=mock_lp_item, + ) + + mock_dl_source.return_value = "/tmp/old.dsc" + + result = lp.get_debdiff(item) + + # Debian fallback was attempted via LP API + mock_debian_archive.getPublishedSources.assert_called_once() + # But it returned no sources, so falls back to changes content + assert "(no changes file available)" in result