diff --git a/README.md b/README.md index a32efa5..420a736 100644 --- a/README.md +++ b/README.md @@ -10,7 +10,6 @@ This tool fetches TODO items from the [Debian UDD (Ultimate Debian Database)](ht - Python 3.10+ - System packages: `python3-click`, `python3-debian`, `python3-rich` -- Package update workflow tools: `salsa`, `gbp`, `dch`, `debuild`, `update-debian-package` Install dependencies on Debian/Ubuntu: @@ -28,10 +27,7 @@ apt install python3-click python3-debian python3-rich ./todo list --show-prerelease # Fetch latest data and show changes -./todo refresh - -# Update a package checkout to the new upstream version -./todo update +./todo update ``` Running `./todo` without arguments is equivalent to `./todo list`. @@ -40,7 +36,7 @@ Running `./todo` without arguments is equivalent to `./todo list`. - `todo.json` - Cached TODO list from UDD (auto-downloaded) - `notes` - Per-package notes (one package per line: ` `) -- `.debian_todo.sqlite3` - SQLite cache for source metadata and package discovery state +- `.vcs_git_cache.json` - Cache of Vcs-Git and uploader info from APT sources ## Notes File Format @@ -74,11 +70,8 @@ PYTHONPATH=. pytest tests/ -v - Compares normalized versions to avoid false positives - Shows team from Vcs-Git (e.g., `python-team`, `HA` for homeassistant-team) - Displays uploaders when no team is set -- Shows the first-seen timestamp for current upstream versions when available -- Tracks when each upstream version was first seen and sorts `./todo list` by that date -- Updates package checkouts with `salsa`, `gbp import-orig`, changelog entries, `update-debian-package`, and `debuild -S` - Responsive output: table view on wide terminals, compact view on narrow -- Caches APT source parsing and package state in SQLite for faster subsequent runs +- Caches APT source parsing for faster subsequent runs ## License diff --git a/debian_todo/__init__.py b/debian_todo/__init__.py index d3825e7..ce2768b 100644 --- a/debian_todo/__init__.py +++ b/debian_todo/__init__.py @@ -11,7 +11,6 @@ import re import glob import os import shutil -import sqlite3 import subprocess import sys import urllib.error @@ -35,39 +34,30 @@ PRERELEASE_RE = re.compile(r"(?i)(?:^|[0-9.\-])(?:alpha|beta|rc|a|b)\d*") CURRENTLY_RE = re.compile( r"^(?P.+?)\s*\(currently in unstable:\s*(?P.+?)\)\s*$" ) -CACHE_PATH = Path(".debian_todo.sqlite3") +CACHE_PATH = Path(".vcs_git_cache.json") CACHE_VERSION = 4 -DISCOVERY_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%SZ" -DISCOVERY_DISPLAY_CUTOFF = datetime.datetime( - 2026, 5, 5, 12, 0, tzinfo=datetime.timezone.utc -) 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 @@ -76,7 +66,6 @@ class ExternalCommandError(PackageUpdateError): class MissingToolError(PackageUpdateError): """Required external tool not available.""" - pass @@ -115,10 +104,7 @@ def run_command( raise ExternalCommandError( command=cmd_str, returncode=result.returncode, - message=( - f"{description} failed: {cmd_str} " - f"(exit code {result.returncode})" - ), + message=f"{description} failed: {cmd_str} (exit code {result.returncode})", ) return result except FileNotFoundError as e: @@ -142,8 +128,7 @@ 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." ) @@ -197,178 +182,74 @@ def normalize_uploaders(uploaders: str) -> str: return "\n".join(cleaned) -def get_db_connection() -> sqlite3.Connection: - """Open the SQLite database and ensure the schema exists.""" - conn = sqlite3.connect(CACHE_PATH) - conn.row_factory = sqlite3.Row - conn.executescript( - """ - CREATE TABLE IF NOT EXISTS metadata ( - key TEXT PRIMARY KEY, - value TEXT NOT NULL - ); - - CREATE TABLE IF NOT EXISTS source_file_mtimes ( - path TEXT PRIMARY KEY, - mtime REAL NOT NULL - ); - - CREATE TABLE IF NOT EXISTS source_info_cache ( - source TEXT PRIMARY KEY, - team TEXT NOT NULL DEFAULT '', - uploaders TEXT NOT NULL DEFAULT '', - vcs_git TEXT - ); - - CREATE TABLE IF NOT EXISTS package_state ( - source TEXT PRIMARY KEY, - shortname TEXT, - todo_type TEXT, - details TEXT, - current_new_version TEXT, - current_version TEXT, - is_upstream INTEGER NOT NULL DEFAULT 0, - is_prerelease INTEGER NOT NULL DEFAULT 0, - discovered_version TEXT, - discovered_at TEXT, - discovered_by_refresh INTEGER NOT NULL DEFAULT 0, - watch_content TEXT, - last_seen_at TEXT, - active INTEGER NOT NULL DEFAULT 1 - ); - """ - ) - package_state_columns = { - row["name"] - for row in conn.execute("PRAGMA table_info(package_state)").fetchall() - } - if "discovered_by_refresh" not in package_state_columns: - conn.execute( - """ - ALTER TABLE package_state - ADD COLUMN discovered_by_refresh INTEGER NOT NULL DEFAULT 0 - """ - ) - return conn - - -def format_discovery_timestamp( - when: Optional[datetime.datetime] = None, -) -> str: - """Return a UTC timestamp string for persisted discovery times.""" - if when is None: - when = datetime.datetime.now(datetime.timezone.utc) - if when.tzinfo is None: - when = when.replace(tzinfo=datetime.timezone.utc) - else: - when = when.astimezone(datetime.timezone.utc) - return when.replace(microsecond=0).strftime(DISCOVERY_TIMESTAMP_FORMAT) - - -def parse_discovery_timestamp(timestamp: str) -> Optional[datetime.datetime]: - """Parse a stored discovery timestamp.""" - try: - return datetime.datetime.strptime( - timestamp, DISCOVERY_TIMESTAMP_FORMAT - ).replace(tzinfo=datetime.timezone.utc) - except ValueError: - return None - - def load_cache(source_paths: list[str]) -> Optional[dict[str, SourceInfo]]: - """Load cached source info from SQLite if still valid. + """Load cached Vcs-Git and uploader info if still valid. - Returns None if cache is missing, unavailable, or stale. + Returns None if cache is missing, corrupted, or stale (based on + Sources file mtimes or cache version mismatch). """ try: - conn = get_db_connection() - except sqlite3.Error: + with CACHE_PATH.open("r", encoding="utf-8") as handle: + data = json.load(handle) + except (FileNotFoundError, json.JSONDecodeError, OSError): + return None + if not isinstance(data, dict): return None - try: - row = conn.execute( - "SELECT value FROM metadata WHERE key = 'source_cache_version'" - ).fetchone() - if row is None or row["value"] != str(CACHE_VERSION): + if data.get("cache_version") != CACHE_VERSION: + return None + cached_mtimes = data.get("sources_mtimes", {}) + if not isinstance(cached_mtimes, dict): + return None + for path in source_paths: + try: + mtime = os.path.getmtime(path) + except OSError: + return None + if str(mtime) != str(cached_mtimes.get(path)): return None - cached_mtimes = { - row["path"]: row["mtime"] - for row in conn.execute( - "SELECT path, mtime FROM source_file_mtimes" - ).fetchall() - } - for path in source_paths: - try: - mtime = os.path.getmtime(path) - except OSError: - return None - if str(mtime) != str(cached_mtimes.get(path)): - return None - - normalized: dict[str, SourceInfo] = {} - for row in conn.execute( - "SELECT source, team, uploaders, vcs_git FROM source_info_cache" - ).fetchall(): - normalized[row["source"]] = { - "team": row["team"], - "uploaders": row["uploaders"], - "vcs_git": row["vcs_git"], - } - return normalized - except sqlite3.Error: + vcs_by_source = data.get("vcs_by_source") + if not isinstance(vcs_by_source, dict): return None - finally: - conn.close() + normalized: dict[str, SourceInfo] = {} + for key, value in vcs_by_source.items(): + if not isinstance(key, str): + return None + if isinstance(value, str): + normalized[key] = {"vcs_git": value, "uploaders": ""} + continue + if not isinstance(value, dict): + return None + vcs_git = value.get("vcs_git") + uploaders = value.get("uploaders") + if not isinstance(vcs_git, str) or not isinstance(uploaders, str): + return None + normalized[key] = {"vcs_git": vcs_git, "uploaders": uploaders} + return normalized def save_cache(source_paths: list[str], vcs_by_source: dict[str, SourceInfo]) -> None: - """Save source info cache to SQLite.""" + """Save Vcs-Git and uploader info to cache file. + + Stores current mtimes of Sources files for cache invalidation. + """ sources_mtimes: dict[str, float] = {} for path in source_paths: try: sources_mtimes[path] = os.path.getmtime(path) except OSError: - return None + return + data = { + "cache_version": CACHE_VERSION, + "sources_mtimes": sources_mtimes, + "vcs_by_source": vcs_by_source, + } try: - conn = get_db_connection() - except sqlite3.Error: + with CACHE_PATH.open("w", encoding="utf-8") as handle: + json.dump(data, handle, sort_keys=True) + except OSError: return - try: - with conn: - conn.execute( - """ - INSERT INTO metadata (key, value) - VALUES ('source_cache_version', ?) - ON CONFLICT(key) DO UPDATE SET value = excluded.value - """, - (str(CACHE_VERSION),), - ) - conn.execute("DELETE FROM source_file_mtimes") - conn.executemany( - "INSERT INTO source_file_mtimes (path, mtime) VALUES (?, ?)", - sources_mtimes.items(), - ) - conn.execute("DELETE FROM source_info_cache") - conn.executemany( - """ - INSERT INTO source_info_cache (source, team, uploaders, vcs_git) - VALUES (?, ?, ?, ?) - """, - [ - ( - source, - info.get("team", ""), - info.get("uploaders", ""), - info.get("vcs_git"), - ) - for source, info in vcs_by_source.items() - ], - ) - except sqlite3.Error: - return - finally: - conn.close() def load_source_info_map() -> dict[str, SourceInfo]: @@ -401,9 +282,8 @@ def load_source_info_map() -> dict[str, SourceInfo]: ) if team or uploaders_text: vcs_by_source[source] = { - "team": team or "", + "vcs_git": team or "", "uploaders": uploaders_text, - "vcs_git": vcs_git, } save_cache(source_paths, vcs_by_source) return vcs_by_source @@ -419,41 +299,29 @@ 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() + payload = response.read().decode("utf-8") except urllib.error.HTTPError as e: raise PackageUpdateError( - f"HTTP error {e.code} while fetching TODO list from " - f"{TODO_URL}: {e.reason}" + f"HTTP error {e.code} while fetching TODO list from {TODO_URL}: {e.reason}" ) from e except urllib.error.URLError as e: raise PackageUpdateError( - f"Network error while fetching TODO list from " f"{TODO_URL}: {e.reason}" + f"Network error while fetching TODO list from {TODO_URL}: {e.reason}" ) from e except TimeoutError as e: raise PackageUpdateError( - f"Timeout after {timeout}s while fetching TODO list " f"from {TODO_URL}" + f"Timeout after {timeout}s while fetching TODO list 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: {preview}" + f"Invalid JSON in TODO list response: {e}" ) from e @@ -464,253 +332,6 @@ def save_todo_list(todo_list: TodoList) -> None: handle.write("\n") -def sync_package_state( - todo_list: TodoList, - seen_at: Optional[datetime.datetime] = None, - record_discovery: bool = True, -) -> None: - """Sync package rows from todo.json into SQLite state.""" - timestamp = format_discovery_timestamp(seen_at) - try: - conn = get_db_connection() - except sqlite3.Error: - return - - present_sources: set[str] = set() - try: - with conn: - for item in todo_list: - source = item.get(":source") - if not isinstance(source, str): - continue - present_sources.add(source) - - shortname = item.get(":shortname") if isinstance( - item.get(":shortname"), str - ) else None - todo_type = item.get(":type") if isinstance(item.get(":type"), str) else None - details = item.get(":details") if isinstance(item.get(":details"), str) else None - is_upstream = bool(shortname and shortname.startswith("newupstream_")) - - new_version = None - current_version = None - is_prerelease = 0 - if details is not None: - new_version, current_version = parse_details(details) - is_prerelease = int(is_prerelease_version(details)) - - existing = conn.execute( - """ - SELECT - discovered_version, - discovered_at, - discovered_by_refresh, - watch_content - FROM package_state WHERE source = ? - """, - (source,), - ).fetchone() - discovered_version = ( - existing["discovered_version"] if existing is not None else None - ) - discovered_at = existing["discovered_at"] if existing is not None else None - discovered_by_refresh = ( - existing["discovered_by_refresh"] if existing is not None else 0 - ) - watch_content = existing["watch_content"] if existing is not None else None - - if record_discovery and is_upstream and new_version: - if discovered_version != new_version or not discovered_at: - discovered_version = new_version - discovered_at = timestamp - discovered_by_refresh = 1 - - conn.execute( - """ - INSERT INTO package_state ( - source, - shortname, - todo_type, - details, - current_new_version, - current_version, - is_upstream, - is_prerelease, - discovered_version, - discovered_at, - discovered_by_refresh, - watch_content, - last_seen_at, - active - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1) - ON CONFLICT(source) DO UPDATE SET - shortname = excluded.shortname, - todo_type = excluded.todo_type, - details = excluded.details, - current_new_version = excluded.current_new_version, - current_version = excluded.current_version, - is_upstream = excluded.is_upstream, - is_prerelease = excluded.is_prerelease, - discovered_version = excluded.discovered_version, - discovered_at = excluded.discovered_at, - discovered_by_refresh = excluded.discovered_by_refresh, - watch_content = excluded.watch_content, - last_seen_at = excluded.last_seen_at, - active = excluded.active - """, - ( - source, - shortname, - todo_type, - details, - new_version, - current_version, - int(is_upstream), - is_prerelease, - discovered_version, - discovered_at, - discovered_by_refresh, - watch_content, - timestamp, - ), - ) - - if present_sources: - placeholders = ",".join("?" for _ in present_sources) - conn.execute( - f"UPDATE package_state SET active = 0 WHERE source NOT IN ({placeholders})", - tuple(sorted(present_sources)), - ) - else: - conn.execute("UPDATE package_state SET active = 0") - except sqlite3.Error: - return - finally: - conn.close() - - -def load_package_discovery_map() -> dict[str, tuple[Optional[str], Optional[str], bool]]: - """Load per-package discovery data keyed by source package.""" - try: - conn = get_db_connection() - except sqlite3.Error: - return {} - try: - return { - row["source"]: ( - row["discovered_version"], - row["discovered_at"], - bool(row["discovered_by_refresh"]), - ) - for row in conn.execute( - """ - SELECT - source, - discovered_version, - discovered_at, - discovered_by_refresh - FROM package_state - """ - ).fetchall() - } - except sqlite3.Error: - return {} - finally: - conn.close() - - -def sort_todo_list_by_discovery(todo_list: TodoList) -> TodoList: - """Sort TODO items by first-seen upstream version date, oldest first.""" - discovery_map = load_package_discovery_map() - - def sort_key(item: TodoItem) -> tuple[int, str, str]: - source = item.get(":source") - details = item.get(":details") - if not isinstance(source, str) or not isinstance(details, str): - return (1, "", "") - discovered_version, discovered_at, discovered_by_refresh = discovery_map.get( - source, (None, None, False) - ) - new_version, _ = parse_details(details) - parsed = ( - parse_discovery_timestamp(discovered_at) if discovered_at is not None else None - ) - if ( - discovered_version == new_version - and discovered_at - and parsed is not None - and parsed >= DISCOVERY_DISPLAY_CUTOFF - ): - return (0, discovered_at, source) - return (1, "", source) - - return sorted(todo_list, key=sort_key) - - -def get_todo_discovery_timestamp( - todo: TodoItem, - discovery_map: dict[str, tuple[Optional[str], Optional[str], bool]], -) -> Optional[str]: - """Return the recorded discovery timestamp for the todo's current version.""" - source = todo.get(":source") - details = todo.get(":details") - if not isinstance(source, str) or not isinstance(details, str): - return None - discovered_version, discovered_at, discovered_by_refresh = discovery_map.get( - source, (None, None, False) - ) - new_version, _ = parse_details(details) - parsed = parse_discovery_timestamp(discovered_at) if discovered_at else None - if ( - discovered_version == new_version - and discovered_at - and parsed is not None - and parsed >= DISCOVERY_DISPLAY_CUTOFF - ): - return discovered_at - return None - - -def format_display_discovery_timestamp(timestamp: Optional[str]) -> str: - """Format a stored discovery timestamp for list output.""" - if not timestamp: - return "-" - parsed = parse_discovery_timestamp(timestamp) - if parsed is None: - return timestamp - return parsed.strftime("%Y-%m-%d %H:%M UTC") - - -def save_watch_snapshot(package: str, repo_dir: Path) -> None: - """Persist a copy of debian/watch for a package when available.""" - watch_path = repo_dir / "debian" / "watch" - watch_content = None - if watch_path.exists(): - try: - watch_content = watch_path.read_text(encoding="utf-8") - except OSError: - watch_content = None - - try: - conn = get_db_connection() - except sqlite3.Error: - return - try: - with conn: - conn.execute( - """ - INSERT INTO package_state (source, watch_content, active) - VALUES (?, ?, 1) - ON CONFLICT(source) DO UPDATE SET watch_content = excluded.watch_content - """, - (package, watch_content), - ) - except sqlite3.Error: - return - finally: - conn.close() - - def summarize_sources(todo_list: TodoList) -> set[str]: """Extract set of source package names from TODO list.""" sources: set[str] = set() @@ -835,22 +456,16 @@ 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: todo_list = cast(TodoList, json.load(handle)) - sync_package_state(todo_list, record_discovery=False) source_info_map = load_source_info_map() notes_by_source = load_notes() - discovery_map = load_package_discovery_map() console = Console() filtered = filter_todo_list(todo_list, include_prerelease=include_prerelease) - filtered = sort_todo_list_by_discovery(filtered) is_narrow = console.width < 100 if is_narrow: console.print("Debian New Upstream TODOs") @@ -859,23 +474,20 @@ def list_todos(include_prerelease: bool) -> None: table.add_column("Source", style="bold") table.add_column("New", style="green", justify="right") table.add_column("Current", style="dim", justify="right") - table.add_column("Seen", style="cyan") table.add_column("Team", justify="right") table.add_column("Note/Uploaders", overflow="fold") for todo in filtered: new_version, current_version = parse_details(todo[":details"]) - discovered_at = get_todo_discovery_timestamp(todo, discovery_map) - display_discovered_at = format_display_discovery_timestamp(discovered_at) source_info = source_info_map.get(todo[":source"], {}) - team = source_info.get("team") + vcs_git = source_info.get("vcs_git") 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 = team or "-" + display_team = vcs_git or "-" if display_team == "homeassistant-team": display_team = "HA" elif display_team.endswith("-team"): @@ -888,7 +500,6 @@ def list_todos(include_prerelease: bool) -> None: parts.append(f"[dim]{display_team}[/dim]") parts.append(f"N: {display_new}") parts.append(f"C: {current_version or '-'}") - parts.append(f"D: {display_discovered_at}") console.print(" ".join(parts)) if display_note != "-": console.print(" " + display_note) @@ -897,7 +508,6 @@ def list_todos(include_prerelease: bool) -> None: source, display_new, current_version or "-", - display_discovered_at, display_team, display_note, ) @@ -906,7 +516,7 @@ def list_todos(include_prerelease: bool) -> None: console.print(f"Packages: {len(filtered)}") -def refresh_todos() -> None: +def update_todos() -> None: """Fetch latest TODO list from UDD and show changes.""" old_list: TodoList = [] if TODO_PATH.exists(): @@ -921,7 +531,6 @@ def refresh_todos() -> None: sys.exit(1) save_todo_list(todo_list) - sync_package_state(todo_list, record_discovery=True) print_changes(filter_todo_list(old_list), filter_todo_list(todo_list)) @@ -1000,21 +609,20 @@ 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("team"): - raise PackageNotFoundError(f"Could not find team info for package '{package}'") + if not source_info or not source_info.get("vcs_git"): + 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: @@ -1024,7 +632,7 @@ def resolve_package_directories( Returns: Tuple of (team_dir, pkg_dir, repo_dir) """ - team_slug = source_info["team"] + team_slug = source_info["vcs_git"] display_team = team_slug_to_display_name(team_slug).lower() team_dir = DEBIAN_SRC_BASE / display_team @@ -1034,9 +642,7 @@ def resolve_package_directories( 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: @@ -1051,12 +657,13 @@ def ensure_package_checkout( 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}...") - 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])) + team_slug = source_info["vcs_git"] + salsa_path = f"{team_slug}/deps/{package}" run_command( ["salsa", "checkout", salsa_path], pkg_dir, @@ -1075,7 +682,8 @@ 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." ) @@ -1088,12 +696,6 @@ 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, @@ -1129,7 +731,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.4) + - Updates Standards-Version to current (4.7.3) Args: repo_dir: Repository directory @@ -1142,7 +744,7 @@ def update_debian_control(repo_dir: Path) -> None: return lines = control_path.read_text().splitlines() - current_standards_version = "4.7.4" + current_standards_version = "4.7.3" new_lines = [] for line in lines: @@ -1161,7 +763,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( @@ -1179,10 +781,8 @@ 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 @@ -1231,9 +831,7 @@ 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): @@ -1426,12 +1024,10 @@ def add_salsa_ci(repo_dir: Path) -> None: if not salsa_repo: return - content = ( - "---\n" - "include:\n" - " - https://salsa.debian.org/salsa-ci-team/pipeline" - "/raw/master/recipes/debian.yml\n" - ) + content = """--- +include: + - https://salsa.debian.org/salsa-ci-team/pipeline/raw/master/recipes/debian.yml +""" salsa_ci_path.write_text(content) @@ -1481,8 +1077,6 @@ def run_package_updates(repo_dir: Path) -> None: ExternalCommandError: If any update command fails """ print("Updating package files...") - subprocess.run(["update-debian-package"], check=True, text=True, cwd=repo_dir) - return update_debian_control(repo_dir) update_debian_copyright_year(repo_dir) update_debian_watch(repo_dir) @@ -1547,15 +1141,9 @@ def update_package(package: str) -> None: # Run package updates run_package_updates(repo_dir) - save_watch_snapshot(package, repo_dir) - - run_command(["debuild", "-S"], repo_dir, "debuild -S") - debian_files = repo_dir / "debian/files" - if debian_files.exists(): - os.remove(debian_files) print(f"Successfully updated {package}") - print(pkg_dir) + print(repo_dir) @click.group(invoke_without_command=True) @@ -1576,14 +1164,14 @@ def list_command(show_prerelease: bool) -> None: list_todos(include_prerelease=show_prerelease) -@cli.command("refresh", help="Fetch the latest todo list and show changes.") -def refresh_command() -> None: - refresh_todos() +@cli.command("update", help="Fetch the latest todo list and show changes.") +def update_command() -> None: + update_todos() -@cli.command("update", help="Update a package to a new upstream version.") +@cli.command("update-pkg", help="Update a package to a new upstream version.") @click.argument("package") -def update_command(package: str) -> None: +def update_pkg_command(package: str) -> None: try: update_package(package) except PackageUpdateError as e: diff --git a/notes b/notes index 25c9976..f3b3b26 100644 --- a/notes +++ b/notes @@ -1,16 +1,10 @@ +pysma needs uv_build backend +pystiebeleltron ImportError: cannot import name 'ModbusDeviceContext' from 'pymodbus.datastore' pytedee-async has been renamed aiotedee python-ftfy bad version +bleak needs new library, bumble - package uploaded flux-led needs new version of webcolors package voluptuous-openapi bad version -hass-nabucasa needs newer version of python-icmplib +hass-nabucasa needs litellm pyjvcprojector AttributeError: 'Channel' object has no attribute 'gethostbyname'. Did you mean: 'gethostbyaddr'? python-volvooncall bad version -python-goodwe RuntimeError: There is no current event loop in thread 'MainThread'. -pygti missing tag during convert to dgit -pyrisco has merge conflicts -sphinxcontrib-typer needs newer typer -zigpy-znp convert to dgit tricky because of exclude files -pyisy bad version -python-voip-utils needs opuslib-next, not in debian -auth0-python need to figure out git-debrebase -pystiebeleltron needs newer modbus diff --git a/tests/test_cache.py b/tests/test_cache.py index 0a19212..f36ad4a 100644 --- a/tests/test_cache.py +++ b/tests/test_cache.py @@ -1,6 +1,7 @@ -"""Tests for SQLite-backed cache and package state.""" +"""Tests for cache loading and saving.""" -import datetime +import json +import pytest from pathlib import Path import debian_todo @@ -9,22 +10,29 @@ import debian_todo class TestLoadCache: """Tests for load_cache function.""" - def test_returns_none_when_db_missing(self, tmp_path, monkeypatch): - monkeypatch.setattr(debian_todo, "CACHE_PATH", tmp_path / "cache.sqlite3") + def test_returns_none_when_file_missing(self, tmp_path, monkeypatch): + cache_file = tmp_path / "cache.json" + monkeypatch.setattr(debian_todo, "CACHE_PATH", cache_file) + + result = debian_todo.load_cache([]) + assert result is None + + def test_returns_none_on_invalid_json(self, tmp_path, monkeypatch): + cache_file = tmp_path / "cache.json" + cache_file.write_text("not valid json") + monkeypatch.setattr(debian_todo, "CACHE_PATH", cache_file) result = debian_todo.load_cache([]) assert result is None def test_returns_none_on_version_mismatch(self, tmp_path, monkeypatch): - db_path = tmp_path / "cache.sqlite3" - monkeypatch.setattr(debian_todo, "CACHE_PATH", db_path) - conn = debian_todo.get_db_connection() - with conn: - conn.execute( - "INSERT INTO metadata (key, value) VALUES ('source_cache_version', ?)", - (str(debian_todo.CACHE_VERSION - 1),), - ) - conn.close() + cache_file = tmp_path / "cache.json" + cache_file.write_text(json.dumps({ + "cache_version": debian_todo.CACHE_VERSION - 1, + "sources_mtimes": {}, + "vcs_by_source": {}, + })) + monkeypatch.setattr(debian_todo, "CACHE_PATH", cache_file) result = debian_todo.load_cache([]) assert result is None @@ -32,19 +40,13 @@ class TestLoadCache: def test_returns_none_when_mtime_mismatch(self, tmp_path, monkeypatch): source_file = tmp_path / "Sources" source_file.write_text("dummy") - db_path = tmp_path / "cache.sqlite3" - monkeypatch.setattr(debian_todo, "CACHE_PATH", db_path) - conn = debian_todo.get_db_connection() - with conn: - conn.execute( - "INSERT INTO metadata (key, value) VALUES ('source_cache_version', ?)", - (str(debian_todo.CACHE_VERSION),), - ) - conn.execute( - "INSERT INTO source_file_mtimes (path, mtime) VALUES (?, ?)", - (str(source_file), 0.0), - ) - conn.close() + cache_file = tmp_path / "cache.json" + cache_file.write_text(json.dumps({ + "cache_version": debian_todo.CACHE_VERSION, + "sources_mtimes": {str(source_file): 0.0}, + "vcs_by_source": {}, + })) + monkeypatch.setattr(debian_todo, "CACHE_PATH", cache_file) result = debian_todo.load_cache([str(source_file)]) assert result is None @@ -55,396 +57,64 @@ class TestLoadCache: source_file = tmp_path / "Sources" source_file.write_text("dummy") mtime = os.path.getmtime(str(source_file)) - db_path = tmp_path / "cache.sqlite3" - monkeypatch.setattr(debian_todo, "CACHE_PATH", db_path) - conn = debian_todo.get_db_connection() - with conn: - conn.execute( - "INSERT INTO metadata (key, value) VALUES ('source_cache_version', ?)", - (str(debian_todo.CACHE_VERSION),), - ) - conn.execute( - "INSERT INTO source_file_mtimes (path, mtime) VALUES (?, ?)", - (str(source_file), mtime), - ) - conn.execute( - """ - INSERT INTO source_info_cache (source, team, uploaders, vcs_git) - VALUES (?, ?, ?, ?) - """, - ("foo", "python-team", "John ", "https://example.invalid/foo"), - ) - conn.close() + + cache_file = tmp_path / "cache.json" + cache_file.write_text(json.dumps({ + "cache_version": debian_todo.CACHE_VERSION, + "sources_mtimes": {str(source_file): mtime}, + "vcs_by_source": { + "foo": {"vcs_git": "python-team", "uploaders": "John "} + }, + })) + monkeypatch.setattr(debian_todo, "CACHE_PATH", cache_file) result = debian_todo.load_cache([str(source_file)]) - assert result == { - "foo": { - "team": "python-team", - "uploaders": "John ", - "vcs_git": "https://example.invalid/foo", - } - } + assert result == {"foo": {"vcs_git": "python-team", "uploaders": "John "}} + + def test_normalizes_legacy_string_format(self, tmp_path, monkeypatch): + import os + + source_file = tmp_path / "Sources" + source_file.write_text("dummy") + mtime = os.path.getmtime(str(source_file)) + + cache_file = tmp_path / "cache.json" + cache_file.write_text(json.dumps({ + "cache_version": debian_todo.CACHE_VERSION, + "sources_mtimes": {str(source_file): mtime}, + "vcs_by_source": {"foo": "python-team"}, # legacy string format + })) + monkeypatch.setattr(debian_todo, "CACHE_PATH", cache_file) + + result = debian_todo.load_cache([str(source_file)]) + assert result == {"foo": {"vcs_git": "python-team", "uploaders": ""}} class TestSaveCache: """Tests for save_cache function.""" - def test_saves_cache_to_sqlite(self, tmp_path, monkeypatch): + def test_saves_cache_file(self, tmp_path, monkeypatch): + import os + source_file = tmp_path / "Sources" source_file.write_text("dummy") - db_path = tmp_path / "cache.sqlite3" - monkeypatch.setattr(debian_todo, "CACHE_PATH", db_path) + cache_file = tmp_path / "cache.json" + monkeypatch.setattr(debian_todo, "CACHE_PATH", cache_file) - vcs_by_source = { - "foo": { - "team": "python-team", - "uploaders": "", - "vcs_git": "https://example.invalid/foo", - } - } + vcs_by_source = {"foo": {"vcs_git": "python-team", "uploaders": ""}} debian_todo.save_cache([str(source_file)], vcs_by_source) - conn = debian_todo.get_db_connection() - row = conn.execute( - "SELECT value FROM metadata WHERE key = 'source_cache_version'" - ).fetchone() - cached = conn.execute( - "SELECT source, team, uploaders, vcs_git FROM source_info_cache" - ).fetchone() - conn.close() - - assert row["value"] == str(debian_todo.CACHE_VERSION) - assert dict(cached) == { - "source": "foo", - "team": "python-team", - "uploaders": "", - "vcs_git": "https://example.invalid/foo", - } + assert cache_file.exists() + data = json.loads(cache_file.read_text()) + assert data["cache_version"] == debian_todo.CACHE_VERSION + assert data["vcs_by_source"] == vcs_by_source + assert str(source_file) in data["sources_mtimes"] def test_handles_missing_source_file(self, tmp_path, monkeypatch): - db_path = tmp_path / "cache.sqlite3" - monkeypatch.setattr(debian_todo, "CACHE_PATH", db_path) + cache_file = tmp_path / "cache.json" + monkeypatch.setattr(debian_todo, "CACHE_PATH", cache_file) debian_todo.save_cache(["/nonexistent/file"], {}) - result = debian_todo.load_cache([]) - assert result is None - - -class TestPackageState: - """Tests for package discovery tracking.""" - - def test_sync_records_first_seen_upstream_version(self, tmp_path, monkeypatch): - db_path = tmp_path / "cache.sqlite3" - monkeypatch.setattr(debian_todo, "CACHE_PATH", db_path) - seen_at = datetime.datetime(2026, 5, 5, 12, 0, tzinfo=datetime.timezone.utc) - - debian_todo.sync_package_state( - [ - { - ":source": "foo", - ":shortname": "newupstream_foo", - ":type": "New upstream", - ":details": "1.2.3 (currently in unstable: 1.2.2-1)", - } - ], - seen_at=seen_at, - ) - - discovery_map = debian_todo.load_package_discovery_map() - assert discovery_map == { - "foo": ("1.2.3", "2026-05-05T12:00:00Z", True) - } - - def test_sync_preserves_discovery_time_for_same_version( - self, tmp_path, monkeypatch - ): - db_path = tmp_path / "cache.sqlite3" - monkeypatch.setattr(debian_todo, "CACHE_PATH", db_path) - - debian_todo.sync_package_state( - [ - { - ":source": "foo", - ":shortname": "newupstream_foo", - ":details": "1.2.3", - } - ], - seen_at=datetime.datetime( - 2026, 5, 5, 12, 0, tzinfo=datetime.timezone.utc - ), - ) - debian_todo.sync_package_state( - [ - { - ":source": "foo", - ":shortname": "newupstream_foo", - ":details": "1.2.3", - } - ], - seen_at=datetime.datetime( - 2026, 5, 6, 12, 0, tzinfo=datetime.timezone.utc - ), - ) - - discovery_map = debian_todo.load_package_discovery_map() - assert discovery_map["foo"] == ("1.2.3", "2026-05-05T12:00:00Z", True) - - def test_sync_updates_discovery_time_when_version_changes( - self, tmp_path, monkeypatch - ): - db_path = tmp_path / "cache.sqlite3" - monkeypatch.setattr(debian_todo, "CACHE_PATH", db_path) - - debian_todo.sync_package_state( - [ - { - ":source": "foo", - ":shortname": "newupstream_foo", - ":details": "1.2.3", - } - ], - seen_at=datetime.datetime( - 2026, 5, 5, 12, 0, tzinfo=datetime.timezone.utc - ), - ) - debian_todo.sync_package_state( - [ - { - ":source": "foo", - ":shortname": "newupstream_foo", - ":details": "1.2.4", - } - ], - seen_at=datetime.datetime( - 2026, 5, 7, 8, 30, tzinfo=datetime.timezone.utc - ), - ) - - discovery_map = debian_todo.load_package_discovery_map() - assert discovery_map["foo"] == ("1.2.4", "2026-05-07T08:30:00Z", True) - - def test_sync_without_discovery_recording_leaves_timestamp_empty( - self, tmp_path, monkeypatch - ): - db_path = tmp_path / "cache.sqlite3" - monkeypatch.setattr(debian_todo, "CACHE_PATH", db_path) - - debian_todo.sync_package_state( - [ - { - ":source": "foo", - ":shortname": "newupstream_foo", - ":details": "1.2.3", - } - ], - seen_at=datetime.datetime( - 2026, 5, 5, 12, 0, tzinfo=datetime.timezone.utc - ), - record_discovery=False, - ) - - discovery_map = debian_todo.load_package_discovery_map() - assert discovery_map["foo"] == (None, None, False) - - def test_sorts_by_oldest_discovery_date(self, tmp_path, monkeypatch): - db_path = tmp_path / "cache.sqlite3" - monkeypatch.setattr(debian_todo, "CACHE_PATH", db_path) - debian_todo.sync_package_state( - [ - { - ":source": "older", - ":shortname": "newupstream_older", - ":details": "1.0.0", - }, - ], - seen_at=datetime.datetime( - 2026, 5, 6, 12, 0, tzinfo=datetime.timezone.utc - ), - ) - debian_todo.sync_package_state( - [ - { - ":source": "newer", - ":shortname": "newupstream_newer", - ":details": "2.0.0", - }, - { - ":source": "older", - ":shortname": "newupstream_older", - ":details": "1.0.0", - }, - ], - seen_at=datetime.datetime( - 2026, 5, 5, 12, 0, tzinfo=datetime.timezone.utc - ), - ) - - sorted_items = debian_todo.sort_todo_list_by_discovery( - [ - { - ":source": "newer", - ":shortname": "newupstream_newer", - ":details": "2.0.0", - }, - { - ":source": "older", - ":shortname": "newupstream_older", - ":details": "1.0.0", - }, - ] - ) - - assert [item[":source"] for item in sorted_items] == ["newer", "older"] - - def test_save_watch_snapshot(self, tmp_path, monkeypatch): - db_path = tmp_path / "cache.sqlite3" - monkeypatch.setattr(debian_todo, "CACHE_PATH", db_path) - repo_dir = tmp_path / "pkg" - debian_dir = repo_dir / "debian" - debian_dir.mkdir(parents=True) - (debian_dir / "watch").write_text("Version: 5\n") - - debian_todo.save_watch_snapshot("mypkg", repo_dir) - - conn = debian_todo.get_db_connection() - row = conn.execute( - "SELECT watch_content FROM package_state WHERE source = ?", - ("mypkg",), - ).fetchone() - conn.close() - assert row["watch_content"] == "Version: 5\n" - - -class TestListOutput: - """Tests for list output using discovery timestamps.""" - - def test_list_shows_discovery_datetime(self, tmp_path, monkeypatch): - db_path = tmp_path / "cache.sqlite3" - todo_path = tmp_path / "todo.json" - monkeypatch.setattr(debian_todo, "CACHE_PATH", db_path) - monkeypatch.setattr(debian_todo, "TODO_PATH", todo_path) - monkeypatch.setattr(debian_todo, "load_source_info_map", lambda: {}) - monkeypatch.setattr(debian_todo, "load_notes", lambda: {}) - - todo_list = [ - { - ":source": "foo", - ":shortname": "newupstream_foo", - ":details": "1.2.3 (currently in unstable: 1.2.2-1)", - } - ] - debian_todo.save_todo_list(todo_list) - debian_todo.sync_package_state( - todo_list, - seen_at=datetime.datetime( - 2026, 5, 5, 12, 0, tzinfo=datetime.timezone.utc - ), - record_discovery=True, - ) - - class FakeConsole: - def __init__(self) -> None: - self.width = 80 - self.output: list[str] = [] - - def print(self, message) -> None: - self.output.append(str(message)) - - fake_console = FakeConsole() - monkeypatch.setattr(debian_todo, "Console", lambda: fake_console) - - debian_todo.list_todos(include_prerelease=False) - - assert any("D: 2026-05-05 12:00 UTC" in line for line in fake_console.output) - - def test_list_hides_pre_cutoff_discovery_datetime(self, tmp_path, monkeypatch): - db_path = tmp_path / "cache.sqlite3" - todo_path = tmp_path / "todo.json" - monkeypatch.setattr(debian_todo, "CACHE_PATH", db_path) - monkeypatch.setattr(debian_todo, "TODO_PATH", todo_path) - monkeypatch.setattr(debian_todo, "load_source_info_map", lambda: {}) - monkeypatch.setattr(debian_todo, "load_notes", lambda: {}) - - todo_list = [ - { - ":source": "foo", - ":shortname": "newupstream_foo", - ":details": "1.2.3 (currently in unstable: 1.2.2-1)", - } - ] - debian_todo.save_todo_list(todo_list) - debian_todo.sync_package_state( - todo_list, - seen_at=datetime.datetime( - 2026, 5, 5, 12, 0, tzinfo=datetime.timezone.utc - ), - record_discovery=False, - ) - - class FakeConsole: - def __init__(self) -> None: - self.width = 80 - self.output: list[str] = [] - - def print(self, message) -> None: - self.output.append(str(message)) - - fake_console = FakeConsole() - monkeypatch.setattr(debian_todo, "Console", lambda: fake_console) - - debian_todo.list_todos(include_prerelease=False) - - assert any("D: -" in line for line in fake_console.output) - - def test_list_shows_post_cutoff_discovery_datetime_even_without_flag( - self, tmp_path, monkeypatch - ): - db_path = tmp_path / "cache.sqlite3" - todo_path = tmp_path / "todo.json" - monkeypatch.setattr(debian_todo, "CACHE_PATH", db_path) - monkeypatch.setattr(debian_todo, "TODO_PATH", todo_path) - monkeypatch.setattr(debian_todo, "load_source_info_map", lambda: {}) - monkeypatch.setattr(debian_todo, "load_notes", lambda: {}) - - todo_list = [ - { - ":source": "foo", - ":shortname": "newupstream_foo", - ":details": "1.2.3 (currently in unstable: 1.2.2-1)", - } - ] - debian_todo.save_todo_list(todo_list) - debian_todo.sync_package_state( - todo_list, - seen_at=datetime.datetime( - 2026, 5, 5, 12, 1, tzinfo=datetime.timezone.utc - ), - record_discovery=False, - ) - conn = debian_todo.get_db_connection() - with conn: - conn.execute( - """ - UPDATE package_state - SET discovered_version = ?, discovered_at = ?, discovered_by_refresh = 0 - WHERE source = ? - """, - ("1.2.3", "2026-05-05T12:01:00Z", "foo"), - ) - conn.close() - - class FakeConsole: - def __init__(self) -> None: - self.width = 80 - self.output: list[str] = [] - - def print(self, message) -> None: - self.output.append(str(message)) - - fake_console = FakeConsole() - monkeypatch.setattr(debian_todo, "Console", lambda: fake_console) - - debian_todo.list_todos(include_prerelease=False) - - assert any("D: 2026-05-05 12:01 UTC" in line for line in fake_console.output) + # Should not create cache file if source file is missing + assert not cache_file.exists() diff --git a/tests/test_update_pkg.py b/tests/test_update_pkg.py index 59760cf..266d310 100644 --- a/tests/test_update_pkg.py +++ b/tests/test_update_pkg.py @@ -155,25 +155,6 @@ 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.""" @@ -296,20 +277,10 @@ class TestValidatePackageInfo: monkeypatch.setattr( debian_todo, "load_source_info_map", - lambda: { - "mypkg": { - "team": "python-team", - "vcs_git": "https://salsa.debian.org/python-team/deps/mypkg.git", - "uploaders": "", - } - }, + lambda: {"mypkg": {"vcs_git": "python-team", "uploaders": ""}}, ) result = debian_todo.validate_package_info("mypkg") - assert result == { - "team": "python-team", - "vcs_git": "https://salsa.debian.org/python-team/deps/mypkg.git", - "uploaders": "", - } + assert result == {"vcs_git": "python-team", "uploaders": ""} def test_package_not_found(self, monkeypatch): """Test package not in source info map.""" @@ -323,7 +294,7 @@ class TestValidatePackageInfo: monkeypatch.setattr( debian_todo, "load_source_info_map", - lambda: {"mypkg": {"team": "", "vcs_git": "", "uploaders": ""}}, + lambda: {"mypkg": {"vcs_git": "", "uploaders": ""}}, ) with pytest.raises(debian_todo.PackageNotFoundError) as exc_info: debian_todo.validate_package_info("mypkg") @@ -336,11 +307,7 @@ class TestResolvePackageDirectories: def test_python_team_package(self, tmp_path, monkeypatch): """Test resolving directories for python-team package.""" monkeypatch.setattr(debian_todo, "DEBIAN_SRC_BASE", tmp_path) - source_info = { - "team": "python-team", - "vcs_git": "https://salsa.debian.org/python-team/deps/mypkg.git", - "uploaders": "", - } + source_info = {"vcs_git": "python-team", "uploaders": ""} team_dir, pkg_dir, repo_dir = debian_todo.resolve_package_directories( "mypkg", source_info ) @@ -351,11 +318,7 @@ class TestResolvePackageDirectories: def test_homeassistant_team_package(self, tmp_path, monkeypatch): """Test resolving directories for homeassistant-team package.""" monkeypatch.setattr(debian_todo, "DEBIAN_SRC_BASE", tmp_path) - source_info = { - "team": "homeassistant-team", - "vcs_git": "https://salsa.debian.org/homeassistant-team/deps/ha-pkg.git", - "uploaders": "", - } + source_info = {"vcs_git": "homeassistant-team", "uploaders": ""} team_dir, pkg_dir, repo_dir = debian_todo.resolve_package_directories( "ha-pkg", source_info ) @@ -370,11 +333,7 @@ class TestEnsurePackageCheckout: def test_successful_checkout(self, tmp_path): """Test successful package checkout.""" pkg_dir = tmp_path / "python" / "mypkg" - source_info = { - "team": "python-team", - "vcs_git": "https://salsa.debian.org/python-team/deps/mypkg.git", - "uploaders": "", - } + source_info = {"vcs_git": "python-team", "uploaders": ""} with patch("debian_todo.run_command") as mock_run: debian_todo.ensure_package_checkout("mypkg", source_info, pkg_dir) @@ -389,11 +348,7 @@ class TestEnsurePackageCheckout: """Test checkout when parent directory already exists.""" pkg_dir = tmp_path / "python" / "mypkg" pkg_dir.mkdir(parents=True) - source_info = { - "team": "python-team", - "vcs_git": "https://salsa.debian.org/python-team/deps/mypkg.git", - "uploaders": "", - } + source_info = {"vcs_git": "python-team", "uploaders": ""} with patch("debian_todo.run_command") as mock_run: debian_todo.ensure_package_checkout("mypkg", source_info, pkg_dir) @@ -425,10 +380,7 @@ class TestRunGbpPqWorkflow: """Test successful gbp pq import and switch.""" with patch("debian_todo.run_command") as mock_run: debian_todo.run_gbp_pq_workflow(tmp_path) - assert mock_run.call_count == 3 - mock_run.assert_any_call( - ["gbp", "pq", "drop"], tmp_path, "Import patch queue" - ) + assert mock_run.call_count == 2 mock_run.assert_any_call( ["gbp", "pq", "import"], tmp_path, "Import patch queue" ) @@ -475,12 +427,16 @@ class TestRunPackageUpdates: """Tests for run_package_updates function.""" def test_runs_all_updates(self, tmp_path): - """Test that the external package update helper is called.""" - with patch("debian_todo.subprocess.run") as mock_run: - debian_todo.run_package_updates(tmp_path) - mock_run.assert_called_once_with( - ["update-debian-package"], check=True, text=True, cwd=tmp_path - ) + """Test that all update functions are called.""" + with patch("debian_todo.update_debian_control") as mock_control: + with patch("debian_todo.update_debian_copyright_year") as mock_copyright: + with patch("debian_todo.update_debian_watch") as mock_watch: + with patch("debian_todo.add_salsa_ci") as mock_salsa: + debian_todo.run_package_updates(tmp_path) + mock_control.assert_called_once_with(tmp_path) + mock_copyright.assert_called_once_with(tmp_path) + mock_watch.assert_called_once_with(tmp_path) + mock_salsa.assert_called_once_with(tmp_path) class TestUpdateDebianControl: @@ -516,14 +472,14 @@ class TestUpdateDebianControl: # Verify Standards-Version was updated content = control_path.read_text() - assert "Standards-Version: 4.7.4" in content + assert "Standards-Version: 4.7.3" in content def test_no_changes_needed(self, tmp_path): """Test when no changes are needed.""" debian_dir = tmp_path / "debian" debian_dir.mkdir() control_path = debian_dir / "control" - control_path.write_text("Source: mypkg\nStandards-Version: 4.7.4\n") + control_path.write_text("Source: mypkg\nStandards-Version: 4.7.3\n") with patch("debian_todo.run_command") as mock_run: debian_todo.update_debian_control(tmp_path) @@ -761,13 +717,7 @@ class TestUpdatePackage: monkeypatch.setattr( debian_todo, "load_source_info_map", - lambda: { - "mypkg": { - "team": "python-team", - "vcs_git": "https://salsa.debian.org/python-team/deps/mypkg.git", - "uploaders": "", - } - }, + lambda: {"mypkg": {"vcs_git": "python-team", "uploaders": ""}}, ) monkeypatch.setattr(debian_todo, "DEBIAN_SRC_BASE", tmp_path) @@ -792,36 +742,27 @@ class TestUpdatePackage: monkeypatch.setattr(debian_todo.shutil, "which", lambda _: "/usr/bin/tool") # Mock the update functions to avoid needing debian/ directory - with patch("debian_todo.run_package_updates"), patch( - "debian_todo.save_watch_snapshot" - ): + with patch("debian_todo.run_package_updates"): debian_todo.update_package("mypkg") # Check salsa checkout was called assert subprocess_calls[0][0] == ["salsa", "checkout", "python-team/deps/mypkg"] # Check gbp pq commands were called - assert subprocess_calls[1][0] == ["gbp", "pq", "drop"] - assert subprocess_calls[2][0] == ["gbp", "pq", "import"] - assert subprocess_calls[3][0] == ["gbp", "pq", "switch"] + assert subprocess_calls[1][0] == ["gbp", "pq", "import"] + assert subprocess_calls[2][0] == ["gbp", "pq", "switch"] # Check gbp import-orig was called - assert subprocess_calls[4][0] == [ + assert subprocess_calls[3][0] == [ "gbp", "import-orig", "--uscan", "--pristine-tar", "--no-interactive" ] # Check dch was called - assert subprocess_calls[5][0][0] == "dch" + assert subprocess_calls[4][0][0] == "dch" def test_existing_checkout_clean(self, tmp_path, monkeypatch): """Test updating a package that's already checked out and clean.""" monkeypatch.setattr( debian_todo, "load_source_info_map", - lambda: { - "mypkg": { - "team": "homeassistant-team", - "vcs_git": "https://salsa.debian.org/homeassistant-team/deps/mypkg.git", - "uploaders": "", - } - }, + lambda: {"mypkg": {"vcs_git": "homeassistant-team", "uploaders": ""}}, ) monkeypatch.setattr(debian_todo, "DEBIAN_SRC_BASE", tmp_path) @@ -846,9 +787,7 @@ class TestUpdatePackage: monkeypatch.setattr(debian_todo.shutil, "which", lambda _: "/usr/bin/tool") # Mock the update functions to avoid needing debian/ directory - with patch("debian_todo.run_package_updates"), patch( - "debian_todo.save_watch_snapshot" - ): + with patch("debian_todo.run_package_updates"): debian_todo.update_package("mypkg") # Should not call salsa checkout @@ -857,24 +796,17 @@ class TestUpdatePackage: # Should call git status, gbp pq, gbp import-orig, dch assert subprocess_calls[0][0] == ["git", "status", "--porcelain"] - assert subprocess_calls[1][0] == ["gbp", "pq", "drop"] - assert subprocess_calls[2][0] == ["gbp", "pq", "import"] - assert subprocess_calls[3][0] == ["gbp", "pq", "switch"] - assert subprocess_calls[4][0][0] == "gbp" - assert subprocess_calls[5][0][0] == "dch" + assert subprocess_calls[1][0] == ["gbp", "pq", "import"] + assert subprocess_calls[2][0] == ["gbp", "pq", "switch"] + assert subprocess_calls[3][0][0] == "gbp" + assert subprocess_calls[4][0][0] == "dch" def test_existing_checkout_dirty(self, tmp_path, monkeypatch): """Test that update aborts if there are uncommitted changes.""" monkeypatch.setattr( debian_todo, "load_source_info_map", - lambda: { - "mypkg": { - "team": "python-team", - "vcs_git": "https://salsa.debian.org/python-team/deps/mypkg.git", - "uploaders": "", - } - }, + lambda: {"mypkg": {"vcs_git": "python-team", "uploaders": ""}}, ) monkeypatch.setattr(debian_todo, "DEBIAN_SRC_BASE", tmp_path) @@ -900,13 +832,7 @@ class TestUpdatePackage: monkeypatch.setattr( debian_todo, "load_source_info_map", - lambda: { - "mypkg": { - "team": "python-team", - "vcs_git": "https://salsa.debian.org/python-team/deps/mypkg.git", - "uploaders": "", - } - }, + lambda: {"mypkg": {"vcs_git": "python-team", "uploaders": ""}}, ) monkeypatch.setattr(debian_todo, "DEBIAN_SRC_BASE", tmp_path) @@ -928,13 +854,7 @@ class TestUpdatePackage: monkeypatch.setattr( debian_todo, "load_source_info_map", - lambda: { - "mypkg": { - "team": "python-team", - "vcs_git": "https://salsa.debian.org/python-team/deps/mypkg.git", - "uploaders": "", - } - }, + lambda: {"mypkg": {"vcs_git": "python-team", "uploaders": ""}}, ) monkeypatch.setattr(debian_todo, "DEBIAN_SRC_BASE", tmp_path) @@ -967,13 +887,7 @@ class TestUpdatePackage: monkeypatch.setattr( debian_todo, "load_source_info_map", - lambda: { - "mypkg": { - "team": "python-team", - "vcs_git": "https://salsa.debian.org/python-team/deps/mypkg.git", - "uploaders": "", - } - }, + lambda: {"mypkg": {"vcs_git": "python-team", "uploaders": ""}}, ) monkeypatch.setattr(debian_todo, "DEBIAN_SRC_BASE", tmp_path)