From c15b57ec877fa185b258120894cd82b99edb2d47 Mon Sep 17 00:00:00 2001 From: Edward Betts Date: Tue, 5 May 2026 09:28:44 +0100 Subject: [PATCH] Various improvements --- debian_todo/__init__.py | 139 +++++++++++++++++++++++++-------------- tests/test_update_pkg.py | 19 ++++++ 2 files changed, 110 insertions(+), 48 deletions(-) diff --git a/debian_todo/__init__.py b/debian_todo/__init__.py index ce2768b..5a9bafd 100644 --- a/debian_todo/__init__.py +++ b/debian_todo/__init__.py @@ -39,25 +39,30 @@ CACHE_VERSION = 4 SourceInfo = dict[str, str] HIDE_UPLOADER = "Edward Betts " DEBIAN_SRC_BASE = Path.home() / "src" / "debian" +UDD_HIGH_LOAD_MARKER = "Current system load" class PackageUpdateError(Exception): """Base exception for package update operations.""" + pass class PackageNotFoundError(PackageUpdateError): """Package not found in source info.""" + pass class RepositoryStateError(PackageUpdateError): """Repository in invalid state (uncommitted changes, etc).""" + pass class ExternalCommandError(PackageUpdateError): """External command failed (salsa, gbp, dch).""" + def __init__(self, command: str, returncode: int, message: str): self.command = command self.returncode = returncode @@ -66,6 +71,7 @@ class ExternalCommandError(PackageUpdateError): class MissingToolError(PackageUpdateError): """Required external tool not available.""" + pass @@ -104,7 +110,10 @@ def run_command( raise ExternalCommandError( command=cmd_str, returncode=result.returncode, - message=f"{description} failed: {cmd_str} (exit code {result.returncode})", + message=( + f"{description} failed: {cmd_str} " + f"(exit code {result.returncode})" + ), ) return result except FileNotFoundError as e: @@ -128,7 +137,8 @@ def check_required_tools() -> None: if missing_tools: tools_str = ", ".join(missing_tools) raise MissingToolError( - f"Required tools not found: {tools_str}. Please install them to continue." + f"Required tools not found: {tools_str}. " + "Please install them to continue." ) @@ -217,15 +227,16 @@ def load_cache(source_paths: list[str]) -> Optional[dict[str, SourceInfo]]: if not isinstance(key, str): return None if isinstance(value, str): - normalized[key] = {"vcs_git": value, "uploaders": ""} + normalized[key] = {"team": value, "uploaders": ""} continue if not isinstance(value, dict): return None - vcs_git = value.get("vcs_git") + team = value.get("team") uploaders = value.get("uploaders") - if not isinstance(vcs_git, str) or not isinstance(uploaders, str): + vcs_git = value.get("vcs_git") + if not isinstance(team, str) or not isinstance(uploaders, str): return None - normalized[key] = {"vcs_git": vcs_git, "uploaders": uploaders} + normalized[key] = {"team": team, "uploaders": uploaders, "vcs_git": vcs_git} return normalized @@ -282,8 +293,9 @@ def load_source_info_map() -> dict[str, SourceInfo]: ) if team or uploaders_text: vcs_by_source[source] = { - "vcs_git": team or "", + "team": team or "", "uploaders": uploaders_text, + "vcs_git": vcs_git, } save_cache(source_paths, vcs_by_source) return vcs_by_source @@ -299,29 +311,41 @@ def fetch_todo_list(timeout: int = 30) -> TodoList: List of TODO items Raises: - PackageUpdateError: If network request fails or JSON is invalid + PackageUpdateError: If network request fails or JSON is + invalid """ try: with urlopen(TODO_URL, timeout=timeout) as response: - payload = response.read().decode("utf-8") + payload = response.read() except urllib.error.HTTPError as e: raise PackageUpdateError( - f"HTTP error {e.code} while fetching TODO list from {TODO_URL}: {e.reason}" + f"HTTP error {e.code} while fetching TODO list from " + f"{TODO_URL}: {e.reason}" ) from e except urllib.error.URLError as e: raise PackageUpdateError( - f"Network error while fetching TODO list from {TODO_URL}: {e.reason}" + f"Network error while fetching TODO list from " f"{TODO_URL}: {e.reason}" ) from e except TimeoutError as e: raise PackageUpdateError( - f"Timeout after {timeout}s while fetching TODO list from {TODO_URL}" + f"Timeout after {timeout}s while fetching TODO list " f"from {TODO_URL}" ) from e try: return cast(TodoList, json.loads(payload)) except json.JSONDecodeError as e: + response_text = payload.decode("utf-8", errors="replace").strip() + if UDD_HIGH_LOAD_MARKER in response_text: + raise PackageUpdateError( + "UDD returned a temporary overload page instead of JSON. " + "Please retry later." + ) from e + + preview = response_text[:120] + if len(response_text) > 120: + preview += "..." raise PackageUpdateError( - f"Invalid JSON in TODO list response: {e}" + f"Invalid JSON in TODO list response: {preview}" ) from e @@ -456,7 +480,10 @@ def list_todos(include_prerelease: bool) -> None: save_todo_list(todo_list) except PackageUpdateError as e: print(f"Error: {e}", file=sys.stderr) - print("Please try again later or check your network connection.", file=sys.stderr) + print( + "Please try again later or check your network connection.", + file=sys.stderr, + ) sys.exit(1) else: with TODO_PATH.open("r", encoding="utf-8") as handle: @@ -480,14 +507,14 @@ def list_todos(include_prerelease: bool) -> None: for todo in filtered: new_version, current_version = parse_details(todo[":details"]) source_info = source_info_map.get(todo[":source"], {}) - vcs_git = source_info.get("vcs_git") + team = source_info.get("team") uploaders = source_info.get("uploaders", "") source = todo[":source"] note = notes_by_source.get(source, "") display_new = new_version if is_prerelease_version(todo[":details"]): display_new = f"[yellow]{new_version} (pre)[/yellow]" - display_team = vcs_git or "-" + display_team = team or "-" if display_team == "homeassistant-team": display_team = "HA" elif display_team.endswith("-team"): @@ -516,7 +543,7 @@ def list_todos(include_prerelease: bool) -> None: console.print(f"Packages: {len(filtered)}") -def update_todos() -> None: +def refresh_todos() -> None: """Fetch latest TODO list from UDD and show changes.""" old_list: TodoList = [] if TODO_PATH.exists(): @@ -609,20 +636,21 @@ def validate_package_info(package: str) -> SourceInfo: Source info dictionary containing vcs_git and uploaders Raises: - PackageNotFoundError: If package not found or has no vcs_git + PackageNotFoundError: If package not found or has no + vcs_git """ source_info_map = load_source_info_map() source_info = source_info_map.get(package) - if not source_info or not source_info.get("vcs_git"): - raise PackageNotFoundError( - f"Could not find team info for package '{package}'" - ) + if not source_info or not source_info.get("team"): + raise PackageNotFoundError(f"Could not find team info for package '{package}'") return source_info -def resolve_package_directories(package: str, source_info: SourceInfo) -> tuple[Path, Path, Path]: +def resolve_package_directories( + package: str, source_info: SourceInfo +) -> tuple[Path, Path, Path]: """Resolve team_dir, pkg_dir, and repo_dir paths. Args: @@ -632,7 +660,7 @@ def resolve_package_directories(package: str, source_info: SourceInfo) -> tuple[ Returns: Tuple of (team_dir, pkg_dir, repo_dir) """ - team_slug = source_info["vcs_git"] + team_slug = source_info["team"] display_team = team_slug_to_display_name(team_slug).lower() team_dir = DEBIAN_SRC_BASE / display_team @@ -642,7 +670,9 @@ def resolve_package_directories(package: str, source_info: SourceInfo) -> tuple[ return team_dir, pkg_dir, repo_dir -def ensure_package_checkout(package: str, source_info: SourceInfo, pkg_dir: Path) -> None: +def ensure_package_checkout( + package: str, source_info: SourceInfo, pkg_dir: Path +) -> None: """Checkout package from salsa. Args: @@ -657,13 +687,12 @@ def ensure_package_checkout(package: str, source_info: SourceInfo, pkg_dir: Path try: pkg_dir.mkdir(parents=True, exist_ok=True) except OSError as e: - raise RepositoryStateError( - f"Failed to create directory {pkg_dir}: {e}" - ) from e + raise RepositoryStateError(f"Failed to create directory {pkg_dir}: {e}") from e print(f"Checking out {package} into {pkg_dir}...") - team_slug = source_info["vcs_git"] - salsa_path = f"{team_slug}/deps/{package}" + assert source_info["vcs_git"].startswith("https://salsa.debian.org/") + salsa_path = source_info["vcs_git"][len("https://salsa.debian.org/") : -4] + print(" ".join(["salsa", "checkout", salsa_path])) run_command( ["salsa", "checkout", salsa_path], pkg_dir, @@ -682,8 +711,7 @@ def validate_repository_state(repo_dir: Path) -> None: """ if has_uncommitted_changes(repo_dir): raise RepositoryStateError( - f"{repo_dir} has uncommitted changes. " - "Please commit or stash them first." + f"{repo_dir} has uncommitted changes. " "Please commit or stash them first." ) @@ -696,6 +724,12 @@ def run_gbp_pq_workflow(repo_dir: Path) -> None: Raises: ExternalCommandError: If gbp pq commands fail """ + run_command( + ["gbp", "pq", "drop"], + repo_dir, + "Import patch queue", + ) + run_command( ["gbp", "pq", "import"], repo_dir, @@ -731,7 +765,7 @@ def update_debian_control(repo_dir: Path) -> None: - Removes obsolete 'Priority: optional' (now default) - Removes obsolete 'Rules-Requires-Root: no' (now default) - - Updates Standards-Version to current (4.7.3) + - Updates Standards-Version to current (4.7.4) Args: repo_dir: Repository directory @@ -744,7 +778,7 @@ def update_debian_control(repo_dir: Path) -> None: return lines = control_path.read_text().splitlines() - current_standards_version = "4.7.3" + current_standards_version = "4.7.4" new_lines = [] for line in lines: @@ -763,7 +797,7 @@ def update_debian_control(repo_dir: Path) -> None: ) continue if line.startswith("Standards-Version: "): - standards_version = line[len("Standards-Version: "):] + standards_version = line[len("Standards-Version: ") :] if standards_version != current_standards_version: line = "Standards-Version: " + current_standards_version run_command( @@ -781,8 +815,10 @@ def update_debian_copyright_year(repo_dir: Path) -> None: """Update Edward Betts' copyright years in debian/copyright. Transforms: - - "Copyright: 2025 Edward Betts <...>" -> "Copyright: 2025-2026 Edward Betts <...>" - - "Copyright: 2022-2024 Edward Betts <...>" -> "Copyright: 2022-2026 Edward Betts <...>" + - "Copyright: 2025 Edward Betts <...>" -> + "Copyright: 2025-2026 Edward Betts <...>" + - "Copyright: 2022-2024 Edward Betts <...>" -> + "Copyright: 2022-2026 Edward Betts <...>" Args: repo_dir: Repository directory @@ -831,7 +867,9 @@ def update_debian_copyright_year(repo_dir: Path) -> None: for line in lines: stripped = line.rstrip("\n") - if field_start_re.match(stripped) and not copyright_field_start_re.match(stripped): + if field_start_re.match(stripped) and not copyright_field_start_re.match( + stripped + ): in_copyright_field = False if copyright_field_start_re.match(stripped): @@ -1024,10 +1062,12 @@ def add_salsa_ci(repo_dir: Path) -> None: if not salsa_repo: return - content = """--- -include: - - https://salsa.debian.org/salsa-ci-team/pipeline/raw/master/recipes/debian.yml -""" + content = ( + "---\n" + "include:\n" + " - https://salsa.debian.org/salsa-ci-team/pipeline" + "/raw/master/recipes/debian.yml\n" + ) salsa_ci_path.write_text(content) @@ -1142,8 +1182,11 @@ def update_package(package: str) -> None: # Run package updates run_package_updates(repo_dir) + run_command(["debuild", "-S"], repo_dir, "debuild -S") + os.remove(repo_dir / "debian/files") + print(f"Successfully updated {package}") - print(repo_dir) + print(pkg_dir) @click.group(invoke_without_command=True) @@ -1164,14 +1207,14 @@ def list_command(show_prerelease: bool) -> None: list_todos(include_prerelease=show_prerelease) -@cli.command("update", help="Fetch the latest todo list and show changes.") -def update_command() -> None: - update_todos() +@cli.command("refresh", help="Fetch the latest todo list and show changes.") +def refresh_command() -> None: + refresh_todos() -@cli.command("update-pkg", help="Update a package to a new upstream version.") +@cli.command("update", help="Update a package to a new upstream version.") @click.argument("package") -def update_pkg_command(package: str) -> None: +def update_command(package: str) -> None: try: update_package(package) except PackageUpdateError as e: diff --git a/tests/test_update_pkg.py b/tests/test_update_pkg.py index 266d310..6cea520 100644 --- a/tests/test_update_pkg.py +++ b/tests/test_update_pkg.py @@ -155,6 +155,25 @@ class TestFetchTodoListErrors: with pytest.raises(debian_todo.PackageUpdateError) as exc_info: debian_todo.fetch_todo_list() assert "Invalid JSON" in str(exc_info.value) + assert "not valid json {" in str(exc_info.value) + + def test_udd_high_load_html_response(self): + """Test known UDD overload HTML response handling.""" + with patch("debian_todo.urlopen") as mock_urlopen: + mock_response = MagicMock() + mock_response.read.return_value = ( + b"

Current system load (26.1) is too high. " + b"Please retry later!

\n" + ) + mock_response.__enter__ = lambda self: self + mock_response.__exit__ = lambda self, *args: None + mock_urlopen.return_value = mock_response + + with pytest.raises(debian_todo.PackageUpdateError) as exc_info: + debian_todo.fetch_todo_list() + + assert "temporary overload page" in str(exc_info.value) + assert "Please retry later" in str(exc_info.value) def test_successful_fetch(self): """Test successful fetch."""