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

Compare commits

base: edward:0d9148e336ec00490bb37fd0909ffe68893b4ed3
Branches Tags
edward:main
..
compare: edward:3b7a07e1d9a2e87813c181bbfcab35b8021a5ebb
Branches Tags
edward:main

No commits in common. "0d9148e336ec00490bb37fd0909ffe68893b4ed3" and "3b7a07e1d9a2e87813c181bbfcab35b8021a5ebb" have entirely different histories.
0d9148e336 ... 3b7a07e1d9

5 changed files with 209 additions and 1050 deletions
Download patch file Download diff file Expand all files Collapse all files

13
README.md
View file

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

584
debian_todo/__init__.py
View file

@ -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<new>.+?)\s*\(currently in unstable:\s*(?P<current>.+?)\)\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 <edward@4angle.com>"
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,107 +182,25 @@ 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
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)
@ -306,69 +209,47 @@ def load_cache(source_paths: list[str]) -> Optional[dict[str, SourceInfo]]:
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
try:
conn = get_db_connection()
except sqlite3.Error:
return
data = {
"cache_version": CACHE_VERSION,
"sources_mtimes": sources_mtimes,
"vcs_by_source": vcs_by_source,
}
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:
with CACHE_PATH.open("w", encoding="utf-8") as handle:
json.dump(data, handle, sort_keys=True)
except OSError:
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:

14
notes
View file

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

474
tests/test_cache.py
View file

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

156
tests/test_update_pkg.py
View file

@ -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"<p><b>Current system load (26.1) is too high. "
b"Please retry later!</b></p>\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:
"""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_run.assert_called_once_with(
["update-debian-package"], check=True, text=True, cwd=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)