From 0d9148e336ec00490bb37fd0909ffe68893b4ed3 Mon Sep 17 00:00:00 2001 From: Edward Betts Date: Sun, 19 Jul 2026 11:42:42 +0100 Subject: [PATCH] Track upstream discovery state --- README.md | 13 +- debian_todo/__init__.py | 475 ++++++++++++++++++++++++++++++++++----- notes | 9 +- tests/test_cache.py | 474 ++++++++++++++++++++++++++++++++------ tests/test_update_pkg.py | 139 +++++++++--- 5 files changed, 945 insertions(+), 165 deletions(-) diff --git a/README.md b/README.md index 420a736..a32efa5 100644 --- a/README.md +++ b/README.md @@ -10,6 +10,7 @@ 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: @@ -27,7 +28,10 @@ apt install python3-click python3-debian python3-rich ./todo list --show-prerelease # Fetch latest data and show changes -./todo update +./todo refresh + +# Update a package checkout to the new upstream version +./todo update ``` Running `./todo` without arguments is equivalent to `./todo list`. @@ -36,7 +40,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: ` `) -- `.vcs_git_cache.json` - Cache of Vcs-Git and uploader info from APT sources +- `.debian_todo.sqlite3` - SQLite cache for source metadata and package discovery state ## Notes File Format @@ -70,8 +74,11 @@ 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 for faster subsequent runs +- Caches APT source parsing and package state in SQLite for faster subsequent runs ## License diff --git a/debian_todo/__init__.py b/debian_todo/__init__.py index 5a9bafd..d3825e7 100644 --- a/debian_todo/__init__.py +++ b/debian_todo/__init__.py @@ -11,6 +11,7 @@ import re import glob import os import shutil +import sqlite3 import subprocess import sys import urllib.error @@ -34,8 +35,12 @@ 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(".vcs_git_cache.json") +CACHE_PATH = Path(".debian_todo.sqlite3") 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" @@ -192,75 +197,178 @@ def normalize_uploaders(uploaders: str) -> str: return "\n".join(cleaned) -def load_cache(source_paths: list[str]) -> Optional[dict[str, SourceInfo]]: - """Load cached Vcs-Git and uploader info if still valid. +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 + ); - Returns None if cache is missing, corrupted, or stale (based on - Sources file mtimes or cache version mismatch). + 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. + + Returns None if cache is missing, unavailable, or stale. """ try: - 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): + conn = get_db_connection() + except sqlite3.Error: return None - 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)): + try: + row = conn.execute( + "SELECT value FROM metadata WHERE key = 'source_cache_version'" + ).fetchone() + if row is None or row["value"] != str(CACHE_VERSION): return None - vcs_by_source = data.get("vcs_by_source") - if not isinstance(vcs_by_source, dict): + 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: return None - 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] = {"team": value, "uploaders": ""} - continue - if not isinstance(value, dict): - return None - team = value.get("team") - uploaders = value.get("uploaders") - vcs_git = value.get("vcs_git") - if not isinstance(team, str) or not isinstance(uploaders, str): - return None - normalized[key] = {"team": team, "uploaders": uploaders, "vcs_git": vcs_git} - return normalized + finally: + conn.close() def save_cache(source_paths: list[str], vcs_by_source: dict[str, SourceInfo]) -> None: - """Save Vcs-Git and uploader info to cache file. - - Stores current mtimes of Sources files for cache invalidation. - """ + """Save source info cache to SQLite.""" sources_mtimes: dict[str, float] = {} for path in source_paths: try: sources_mtimes[path] = os.path.getmtime(path) except OSError: - return - data = { - "cache_version": CACHE_VERSION, - "sources_mtimes": sources_mtimes, - "vcs_by_source": vcs_by_source, - } + return None try: - with CACHE_PATH.open("w", encoding="utf-8") as handle: - json.dump(data, handle, sort_keys=True) - except OSError: + conn = get_db_connection() + except sqlite3.Error: 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]: @@ -356,6 +464,253 @@ 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() @@ -489,10 +844,13 @@ def list_todos(include_prerelease: bool) -> None: 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") @@ -501,11 +859,14 @@ 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") uploaders = source_info.get("uploaders", "") @@ -527,6 +888,7 @@ 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) @@ -535,6 +897,7 @@ def list_todos(include_prerelease: bool) -> None: source, display_new, current_version or "-", + display_discovered_at, display_team, display_note, ) @@ -558,6 +921,7 @@ 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)) @@ -1117,6 +1481,8 @@ 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) @@ -1181,9 +1547,12 @@ 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") - os.remove(repo_dir / "debian/files") + debian_files = repo_dir / "debian/files" + if debian_files.exists(): + os.remove(debian_files) print(f"Successfully updated {package}") print(pkg_dir) diff --git a/notes b/notes index ac5d3ab..25c9976 100644 --- a/notes +++ b/notes @@ -6,4 +6,11 @@ hass-nabucasa needs newer version of python-icmplib 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'. -aiorussound needs new package serialx (rust) +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 f36ad4a..0a19212 100644 --- a/tests/test_cache.py +++ b/tests/test_cache.py @@ -1,7 +1,6 @@ -"""Tests for cache loading and saving.""" +"""Tests for SQLite-backed cache and package state.""" -import json -import pytest +import datetime from pathlib import Path import debian_todo @@ -10,29 +9,22 @@ import debian_todo class TestLoadCache: """Tests for load_cache function.""" - 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) + def test_returns_none_when_db_missing(self, tmp_path, monkeypatch): + monkeypatch.setattr(debian_todo, "CACHE_PATH", tmp_path / "cache.sqlite3") result = debian_todo.load_cache([]) assert result is None def test_returns_none_on_version_mismatch(self, tmp_path, monkeypatch): - 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) + 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() result = debian_todo.load_cache([]) assert result is None @@ -40,13 +32,19 @@ class TestLoadCache: def test_returns_none_when_mtime_mismatch(self, tmp_path, monkeypatch): source_file = tmp_path / "Sources" source_file.write_text("dummy") - 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) + 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() result = debian_todo.load_cache([str(source_file)]) assert result is None @@ -57,64 +55,396 @@ class TestLoadCache: 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": {"vcs_git": "python-team", "uploaders": "John "} - }, - })) - monkeypatch.setattr(debian_todo, "CACHE_PATH", cache_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() result = debian_todo.load_cache([str(source_file)]) - 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": ""}} + assert result == { + "foo": { + "team": "python-team", + "uploaders": "John ", + "vcs_git": "https://example.invalid/foo", + } + } class TestSaveCache: """Tests for save_cache function.""" - def test_saves_cache_file(self, tmp_path, monkeypatch): - import os - + def test_saves_cache_to_sqlite(self, tmp_path, monkeypatch): source_file = tmp_path / "Sources" source_file.write_text("dummy") - cache_file = tmp_path / "cache.json" - monkeypatch.setattr(debian_todo, "CACHE_PATH", cache_file) + db_path = tmp_path / "cache.sqlite3" + monkeypatch.setattr(debian_todo, "CACHE_PATH", db_path) - vcs_by_source = {"foo": {"vcs_git": "python-team", "uploaders": ""}} + vcs_by_source = { + "foo": { + "team": "python-team", + "uploaders": "", + "vcs_git": "https://example.invalid/foo", + } + } debian_todo.save_cache([str(source_file)], vcs_by_source) - 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"] + 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", + } def test_handles_missing_source_file(self, tmp_path, monkeypatch): - cache_file = tmp_path / "cache.json" - monkeypatch.setattr(debian_todo, "CACHE_PATH", cache_file) + db_path = tmp_path / "cache.sqlite3" + monkeypatch.setattr(debian_todo, "CACHE_PATH", db_path) debian_todo.save_cache(["/nonexistent/file"], {}) - # Should not create cache file if source file is missing - assert not cache_file.exists() + 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) diff --git a/tests/test_update_pkg.py b/tests/test_update_pkg.py index 6cea520..59760cf 100644 --- a/tests/test_update_pkg.py +++ b/tests/test_update_pkg.py @@ -296,10 +296,20 @@ class TestValidatePackageInfo: monkeypatch.setattr( debian_todo, "load_source_info_map", - lambda: {"mypkg": {"vcs_git": "python-team", "uploaders": ""}}, + lambda: { + "mypkg": { + "team": "python-team", + "vcs_git": "https://salsa.debian.org/python-team/deps/mypkg.git", + "uploaders": "", + } + }, ) result = debian_todo.validate_package_info("mypkg") - assert result == {"vcs_git": "python-team", "uploaders": ""} + assert result == { + "team": "python-team", + "vcs_git": "https://salsa.debian.org/python-team/deps/mypkg.git", + "uploaders": "", + } def test_package_not_found(self, monkeypatch): """Test package not in source info map.""" @@ -313,7 +323,7 @@ class TestValidatePackageInfo: monkeypatch.setattr( debian_todo, "load_source_info_map", - lambda: {"mypkg": {"vcs_git": "", "uploaders": ""}}, + lambda: {"mypkg": {"team": "", "vcs_git": "", "uploaders": ""}}, ) with pytest.raises(debian_todo.PackageNotFoundError) as exc_info: debian_todo.validate_package_info("mypkg") @@ -326,7 +336,11 @@ 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 = {"vcs_git": "python-team", "uploaders": ""} + source_info = { + "team": "python-team", + "vcs_git": "https://salsa.debian.org/python-team/deps/mypkg.git", + "uploaders": "", + } team_dir, pkg_dir, repo_dir = debian_todo.resolve_package_directories( "mypkg", source_info ) @@ -337,7 +351,11 @@ 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 = {"vcs_git": "homeassistant-team", "uploaders": ""} + source_info = { + "team": "homeassistant-team", + "vcs_git": "https://salsa.debian.org/homeassistant-team/deps/ha-pkg.git", + "uploaders": "", + } team_dir, pkg_dir, repo_dir = debian_todo.resolve_package_directories( "ha-pkg", source_info ) @@ -352,7 +370,11 @@ class TestEnsurePackageCheckout: def test_successful_checkout(self, tmp_path): """Test successful package checkout.""" pkg_dir = tmp_path / "python" / "mypkg" - source_info = {"vcs_git": "python-team", "uploaders": ""} + source_info = { + "team": "python-team", + "vcs_git": "https://salsa.debian.org/python-team/deps/mypkg.git", + "uploaders": "", + } with patch("debian_todo.run_command") as mock_run: debian_todo.ensure_package_checkout("mypkg", source_info, pkg_dir) @@ -367,7 +389,11 @@ class TestEnsurePackageCheckout: """Test checkout when parent directory already exists.""" pkg_dir = tmp_path / "python" / "mypkg" pkg_dir.mkdir(parents=True) - source_info = {"vcs_git": "python-team", "uploaders": ""} + source_info = { + "team": "python-team", + "vcs_git": "https://salsa.debian.org/python-team/deps/mypkg.git", + "uploaders": "", + } with patch("debian_todo.run_command") as mock_run: debian_todo.ensure_package_checkout("mypkg", source_info, pkg_dir) @@ -399,7 +425,10 @@ 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 == 2 + assert mock_run.call_count == 3 + mock_run.assert_any_call( + ["gbp", "pq", "drop"], tmp_path, "Import patch queue" + ) mock_run.assert_any_call( ["gbp", "pq", "import"], tmp_path, "Import patch queue" ) @@ -446,16 +475,12 @@ class TestRunPackageUpdates: """Tests for run_package_updates function.""" def test_runs_all_updates(self, 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) + """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 + ) class TestUpdateDebianControl: @@ -491,14 +516,14 @@ class TestUpdateDebianControl: # Verify Standards-Version was updated content = control_path.read_text() - assert "Standards-Version: 4.7.3" in content + assert "Standards-Version: 4.7.4" 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.3\n") + control_path.write_text("Source: mypkg\nStandards-Version: 4.7.4\n") with patch("debian_todo.run_command") as mock_run: debian_todo.update_debian_control(tmp_path) @@ -736,7 +761,13 @@ class TestUpdatePackage: monkeypatch.setattr( debian_todo, "load_source_info_map", - lambda: {"mypkg": {"vcs_git": "python-team", "uploaders": ""}}, + lambda: { + "mypkg": { + "team": "python-team", + "vcs_git": "https://salsa.debian.org/python-team/deps/mypkg.git", + "uploaders": "", + } + }, ) monkeypatch.setattr(debian_todo, "DEBIAN_SRC_BASE", tmp_path) @@ -761,27 +792,36 @@ 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"): + with patch("debian_todo.run_package_updates"), patch( + "debian_todo.save_watch_snapshot" + ): 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", "import"] - assert subprocess_calls[2][0] == ["gbp", "pq", "switch"] + assert subprocess_calls[1][0] == ["gbp", "pq", "drop"] + assert subprocess_calls[2][0] == ["gbp", "pq", "import"] + assert subprocess_calls[3][0] == ["gbp", "pq", "switch"] # Check gbp import-orig was called - assert subprocess_calls[3][0] == [ + assert subprocess_calls[4][0] == [ "gbp", "import-orig", "--uscan", "--pristine-tar", "--no-interactive" ] # Check dch was called - assert subprocess_calls[4][0][0] == "dch" + assert subprocess_calls[5][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": {"vcs_git": "homeassistant-team", "uploaders": ""}}, + lambda: { + "mypkg": { + "team": "homeassistant-team", + "vcs_git": "https://salsa.debian.org/homeassistant-team/deps/mypkg.git", + "uploaders": "", + } + }, ) monkeypatch.setattr(debian_todo, "DEBIAN_SRC_BASE", tmp_path) @@ -806,7 +846,9 @@ 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"): + with patch("debian_todo.run_package_updates"), patch( + "debian_todo.save_watch_snapshot" + ): debian_todo.update_package("mypkg") # Should not call salsa checkout @@ -815,17 +857,24 @@ 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", "import"] - assert subprocess_calls[2][0] == ["gbp", "pq", "switch"] - assert subprocess_calls[3][0][0] == "gbp" - assert subprocess_calls[4][0][0] == "dch" + 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" 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": {"vcs_git": "python-team", "uploaders": ""}}, + lambda: { + "mypkg": { + "team": "python-team", + "vcs_git": "https://salsa.debian.org/python-team/deps/mypkg.git", + "uploaders": "", + } + }, ) monkeypatch.setattr(debian_todo, "DEBIAN_SRC_BASE", tmp_path) @@ -851,7 +900,13 @@ class TestUpdatePackage: monkeypatch.setattr( debian_todo, "load_source_info_map", - lambda: {"mypkg": {"vcs_git": "python-team", "uploaders": ""}}, + lambda: { + "mypkg": { + "team": "python-team", + "vcs_git": "https://salsa.debian.org/python-team/deps/mypkg.git", + "uploaders": "", + } + }, ) monkeypatch.setattr(debian_todo, "DEBIAN_SRC_BASE", tmp_path) @@ -873,7 +928,13 @@ class TestUpdatePackage: monkeypatch.setattr( debian_todo, "load_source_info_map", - lambda: {"mypkg": {"vcs_git": "python-team", "uploaders": ""}}, + lambda: { + "mypkg": { + "team": "python-team", + "vcs_git": "https://salsa.debian.org/python-team/deps/mypkg.git", + "uploaders": "", + } + }, ) monkeypatch.setattr(debian_todo, "DEBIAN_SRC_BASE", tmp_path) @@ -906,7 +967,13 @@ class TestUpdatePackage: monkeypatch.setattr( debian_todo, "load_source_info_map", - lambda: {"mypkg": {"vcs_git": "python-team", "uploaders": ""}}, + lambda: { + "mypkg": { + "team": "python-team", + "vcs_git": "https://salsa.debian.org/python-team/deps/mypkg.git", + "uploaders": "", + } + }, ) monkeypatch.setattr(debian_todo, "DEBIAN_SRC_BASE", tmp_path)