edward/debian-todo
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Track upstream discovery state

Browse source
This commit is contained in:
Edward Betts 2026-07-19 11:42:42 +01:00
parent 928c7841c3
commit 0d9148e336
5 changed files with 945 additions and 165 deletions
Download patch file Download diff file Expand all files Collapse all files

13
README.md
View file

@ -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 <source-package>
```
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: `<source> <note>`)
- `.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

475
debian_todo/__init__.py
View file

@ -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<new>.+?)\s*\(currently in unstable:\s*(?P<current>.+?)\)\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 <edward@4angle.com>"
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)

9
notes
View file

@ -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

474
tests/test_cache.py
View file

@ -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 <j@e.com>"}
},
}))
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 <j@e.com>", "https://example.invalid/foo"),
)
conn.close()
result = debian_todo.load_cache([str(source_file)])
assert result == {"foo": {"vcs_git": "python-team", "uploaders": "John <j@e.com>"}}
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 <j@e.com>",
"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)

139
tests/test_update_pkg.py
View file

@ -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)