diff --git a/AGENTS.md b/AGENTS.md index a423735..bce5131 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -8,7 +8,9 @@ This is a Python CLI tool for Debian package maintainers. It tracks packages wit ## Architecture -- Single-file CLI application using Click for command handling +- `debian_todo/` - Python package with core logic +- `todo` - CLI entry point script +- `tests/` - pytest test suite - Rich library for terminal output (tables and formatting) - python-debian library for parsing APT Sources files - JSON-based caching for performance @@ -36,7 +38,12 @@ This is a Python CLI tool for Debian package maintainers. It tracks packages wit ## Testing -No test suite. Manual testing with: +Run tests with pytest: +```bash +PYTHONPATH=. pytest tests/ -v +``` + +Manual testing: ```bash ./todo list ./todo list --show-prerelease @@ -60,5 +67,6 @@ System packages (Debian): - python3-click - python3-debian - python3-rich +- python3-pytest (for testing) No pip/venv setup; uses system Python. diff --git a/README.md b/README.md index d551326..420a736 100644 --- a/README.md +++ b/README.md @@ -54,6 +54,15 @@ Each line starts with the source package name, followed by a note. The tool fetches data for a specific maintainer email configured in the `TODO_URL` constant. Modify this URL to track packages for a different maintainer. +## Testing + +Run tests with pytest: + +```bash +apt install python3-pytest +PYTHONPATH=. pytest tests/ -v +``` + ## Features - Filters new upstream notifications, hiding other TODO types diff --git a/debian_todo/__init__.py b/debian_todo/__init__.py new file mode 100644 index 0000000..530e654 --- /dev/null +++ b/debian_todo/__init__.py @@ -0,0 +1,424 @@ +"""CLI tool for tracking Debian packages with new upstream versions. + +Fetches TODO items from the Debian UDD (Ultimate Debian Database) and displays +packages where a new upstream version is available. Filters out pre-release +versions and shows team/uploader metadata. +""" + +import json +import re +import glob +import os +from pathlib import Path +from typing import Any, Optional, cast +from urllib.request import urlopen + +import click +from debian import deb822 +from rich.console import Console +from rich.table import Table + + +TODO_URL = "https://udd.debian.org/dmd/?email1=edward%404angle.com&format=json" +TODO_PATH = Path("todo.json") +NOTES_PATH = Path("notes") +TodoItem = dict[str, Any] +TodoList = list[TodoItem] + +PRERELEASE_RE = re.compile(r"(?i)(?:^|[0-9.\-])(?:alpha|beta|rc|a|b)\d*") +CURRENTLY_RE = re.compile( + r"^(?P.+?)\s*\(currently in unstable:\s*(?P.+?)\)\s*$" +) +CACHE_PATH = Path(".vcs_git_cache.json") +CACHE_VERSION = 4 +SourceInfo = dict[str, str] +HIDE_UPLOADER = "Edward Betts " + + +def parse_details(details: str) -> tuple[str, Optional[str]]: + """Parse version details string into new and current versions. + + Args: + details: String like "1.2.3 (currently in unstable: 1.2.2-1)" + + Returns: + Tuple of (new_version, current_version). current_version is None + if not present in the input. + """ + match = CURRENTLY_RE.match(details) + if match: + return match.group("new").strip(), match.group("current").strip() + return details.strip(), None + + +def is_prerelease_version(details: str) -> bool: + """Check if the new version in details is a pre-release. + + Detects versions containing alpha, beta, rc, a, or b suffixes. + """ + new_version, _ = parse_details(details) + return bool(PRERELEASE_RE.search(new_version)) + + +def vcs_git_to_team(vcs_git: Optional[str]) -> Optional[str]: + """Extract team name from a Vcs-Git URL. + + For salsa.debian.org URLs, extracts the group/team name. + Returns the full URL for non-Salsa repositories. + """ + if not vcs_git: + return None + match = re.search(r"salsa\.debian\.org/([^/]+)/", vcs_git) + if not match: + return vcs_git + return match.group(1) + + +def normalize_uploaders(uploaders: str) -> str: + """Clean and format uploaders string. + + Splits comma-separated uploaders, removes the configured HIDE_UPLOADER, + and joins with newlines. + """ + parts = [part.strip().strip(",") for part in uploaders.split(",")] + cleaned = [part for part in parts if part and part != HIDE_UPLOADER] + 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. + + Returns None if cache is missing, corrupted, or stale (based on + Sources file mtimes or cache version mismatch). + """ + 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): + 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)): + return None + + vcs_by_source = data.get("vcs_by_source") + if not isinstance(vcs_by_source, dict): + 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] = {"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 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 + data = { + "cache_version": CACHE_VERSION, + "sources_mtimes": sources_mtimes, + "vcs_by_source": vcs_by_source, + } + try: + with CACHE_PATH.open("w", encoding="utf-8") as handle: + json.dump(data, handle, sort_keys=True) + except OSError: + return + + +def load_source_info_map() -> dict[str, SourceInfo]: + """Load Vcs-Git and uploader info for all source packages. + + Parses APT Sources files from /var/lib/apt/lists/ and extracts + Vcs-Git URLs and Uploaders fields. Results are cached to disk. + """ + source_paths = sorted(glob.glob("/var/lib/apt/lists/*Sources")) + cached = load_cache(source_paths) + if cached is not None: + return cached + + vcs_by_source: dict[str, SourceInfo] = {} + for path in source_paths: + with Path(path).open("r", encoding="utf-8", errors="replace") as handle: + for entry in deb822.Deb822.iter_paragraphs(handle): + source = entry.get("Source") or entry.get("Package") + if not source or source in vcs_by_source: + continue + vcs_git = entry.get("Vcs-Git") + uploaders = entry.get("Uploaders") + team = None + if vcs_git: + team = vcs_git_to_team(vcs_git.strip()) + uploaders_text = "" + if uploaders: + uploaders_text = normalize_uploaders( + re.sub(r"\s+", " ", uploaders).strip() + ) + if team or uploaders_text: + vcs_by_source[source] = { + "vcs_git": team or "", + "uploaders": uploaders_text, + } + save_cache(source_paths, vcs_by_source) + return vcs_by_source + + +def fetch_todo_list() -> TodoList: + """Fetch the TODO list from UDD as JSON.""" + with urlopen(TODO_URL) as response: + payload = response.read().decode("utf-8") + return cast(TodoList, json.loads(payload)) + + +def save_todo_list(todo_list: TodoList) -> None: + """Save TODO list to local JSON file.""" + with TODO_PATH.open("w", encoding="utf-8") as handle: + json.dump(todo_list, handle, indent=2, ensure_ascii=True) + handle.write("\n") + + +def summarize_sources(todo_list: TodoList) -> set[str]: + """Extract set of source package names from TODO list.""" + sources: set[str] = set() + for item in todo_list: + source = item.get(":source") + if isinstance(source, str): + sources.add(source) + return sources + + +def load_notes() -> dict[str, str]: + """Load per-package notes from the notes file. + + Each line should be: + Multiple notes for the same package are joined with semicolons. + """ + if not NOTES_PATH.exists(): + return {} + notes_by_source: dict[str, list[str]] = {} + with NOTES_PATH.open("r", encoding="utf-8") as handle: + for line in handle: + stripped = line.strip() + if not stripped: + continue + parts = stripped.split(None, 1) + if not parts: + continue + source = parts[0] + note = parts[1] if len(parts) > 1 else "" + notes_by_source.setdefault(source, []).append(note) + return { + source: "; ".join(note for note in notes if note) + for source, notes in notes_by_source.items() + } + + +def filter_todo_list(todo_list: TodoList, include_prerelease: bool = False) -> TodoList: + """Filter TODO list to only new upstream version items. + + Removes non-upstream items, pre-releases (unless include_prerelease=True), + and items where normalized versions already match. + """ + filtered: TodoList = [] + for item in todo_list: + shortname = item.get(":shortname") + details = item.get(":details") + if not isinstance(shortname, str) or not isinstance(details, str): + continue + if not shortname.startswith("newupstream_"): + continue + if not include_prerelease and is_prerelease_version(details): + continue + new_version, current_version = parse_details(details) + if current_version: + normalized_new = normalize_upstream_version(new_version) + normalized_current = normalize_upstream_version(current_version) + if normalized_new == normalized_current: + continue + filtered.append(item) + return filtered + + +def normalize_upstream_version(version: str) -> str: + """Strip epoch and Debian revision from version string. + + "1:2.3.4-5" -> "2.3.4" + """ + if ":" in version: + version = version.split(":", 1)[1] + if "-" in version: + version = version.rsplit("-", 1)[0] + return version.strip() + + +def print_changes(old_list: TodoList, new_list: TodoList) -> None: + """Print added and removed packages between two TODO lists.""" + + def format_details(details: str) -> str: + new_version, current_version = parse_details(details) + display_new = new_version + if is_prerelease_version(details): + display_new = f"{new_version} (pre)" + return f"New: {display_new} | Current: {current_version or '-'}" + + def build_details(todo_list: TodoList) -> dict[str, str]: + details_by_source: dict[str, str] = {} + for item in todo_list: + source = item.get(":source") + details = item.get(":details") + if isinstance(source, str) and isinstance(details, str): + details_by_source[source] = format_details(details) + return details_by_source + + old_details = build_details(old_list) + new_details = build_details(new_list) + old_sources = set(old_details) + new_sources = set(new_details) + added = sorted(new_sources - old_sources) + removed = sorted(old_sources - new_sources) + if not added and not removed: + print("No changes in todo.json.") + return + if added: + print("New packages:") + for source in added: + print(f" {source} - {new_details.get(source, '-')}") + if removed: + print("Removed packages:") + for source in removed: + print(f" {source} - {old_details.get(source, '-')}") + + +def list_todos(include_prerelease: bool) -> None: + """Display filtered TODO items in a table. + + Downloads todo.json from UDD if not present locally. + """ + if not TODO_PATH.exists(): + print("Downloading todo.json...") + todo_list = fetch_todo_list() + save_todo_list(todo_list) + else: + with TODO_PATH.open("r", encoding="utf-8") as handle: + todo_list = cast(TodoList, json.load(handle)) + + source_info_map = load_source_info_map() + notes_by_source = load_notes() + console = Console() + filtered = filter_todo_list(todo_list, include_prerelease=include_prerelease) + is_narrow = console.width < 100 + if is_narrow: + console.print("Debian New Upstream TODOs") + else: + table = Table(title="Debian New Upstream TODOs") + 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("Team", justify="right") + table.add_column("Note/Uploaders", overflow="fold") + + for todo in filtered: + new_version, current_version = parse_details(todo[":details"]) + source_info = source_info_map.get(todo[":source"], {}) + vcs_git = source_info.get("vcs_git") + uploaders = source_info.get("uploaders", "") + source = todo[":source"] + note = notes_by_source.get(source, "") + display_new = new_version + if is_prerelease_version(todo[":details"]): + display_new = f"[yellow]{new_version} (pre)[/yellow]" + display_team = vcs_git or "-" + if display_team == "homeassistant-team": + display_team = "HA" + elif display_team.endswith("-team"): + display_team = display_team[:-5] + display_note = note or uploaders or "-" + + if is_narrow: + parts = [f"[bold]{source}[/bold]"] + if display_team != "-": + parts.append(f"[dim]{display_team}[/dim]") + parts.append(f"N: {display_new}") + parts.append(f"C: {current_version or '-'}") + console.print(" ".join(parts)) + if display_note != "-": + console.print(" " + display_note) + else: + table.add_row( + source, + display_new, + current_version or "-", + display_team, + display_note, + ) + if not is_narrow: + console.print(table) + console.print(f"Packages: {len(filtered)}") + + +def update_todos() -> None: + """Fetch latest TODO list from UDD and show changes.""" + old_list: TodoList = [] + if TODO_PATH.exists(): + with TODO_PATH.open("r", encoding="utf-8") as handle: + old_list = cast(TodoList, json.load(handle)) + todo_list = fetch_todo_list() + save_todo_list(todo_list) + print_changes(filter_todo_list(old_list), filter_todo_list(todo_list)) + + +@click.group(invoke_without_command=True) +@click.pass_context +def cli(context: click.Context) -> None: + """Track Debian packages with new upstream versions.""" + if context.invoked_subcommand is None: + list_todos(include_prerelease=False) + + +@cli.command("list", help="List filtered new upstream todo entries.") +@click.option( + "--show-prerelease", + is_flag=True, + help="Include prerelease versions in the list output.", +) +def list_command(show_prerelease: bool) -> None: + list_todos(include_prerelease=show_prerelease) + + +@cli.command("update", help="Fetch the latest todo list and show changes.") +def update_command() -> None: + update_todos() + + +def main() -> None: + """Entry point for the CLI.""" + cli() diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_cache.py b/tests/test_cache.py new file mode 100644 index 0000000..f36ad4a --- /dev/null +++ b/tests/test_cache.py @@ -0,0 +1,120 @@ +"""Tests for cache loading and saving.""" + +import json +import pytest +from pathlib import Path + +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) + + 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) + + result = debian_todo.load_cache([]) + assert result is None + + 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) + + result = debian_todo.load_cache([str(source_file)]) + assert result is None + + def test_returns_cache_when_valid(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": {"vcs_git": "python-team", "uploaders": "John "} + }, + })) + monkeypatch.setattr(debian_todo, "CACHE_PATH", cache_file) + + result = debian_todo.load_cache([str(source_file)]) + assert result == {"foo": {"vcs_git": "python-team", "uploaders": "John "}} + + def test_normalizes_legacy_string_format(self, tmp_path, monkeypatch): + import os + + source_file = tmp_path / "Sources" + source_file.write_text("dummy") + mtime = os.path.getmtime(str(source_file)) + + cache_file = tmp_path / "cache.json" + cache_file.write_text(json.dumps({ + "cache_version": debian_todo.CACHE_VERSION, + "sources_mtimes": {str(source_file): mtime}, + "vcs_by_source": {"foo": "python-team"}, # legacy string format + })) + monkeypatch.setattr(debian_todo, "CACHE_PATH", cache_file) + + result = debian_todo.load_cache([str(source_file)]) + assert result == {"foo": {"vcs_git": "python-team", "uploaders": ""}} + + +class TestSaveCache: + """Tests for save_cache function.""" + + def test_saves_cache_file(self, tmp_path, monkeypatch): + import os + + source_file = tmp_path / "Sources" + source_file.write_text("dummy") + cache_file = tmp_path / "cache.json" + monkeypatch.setattr(debian_todo, "CACHE_PATH", cache_file) + + vcs_by_source = {"foo": {"vcs_git": "python-team", "uploaders": ""}} + 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"] + + def test_handles_missing_source_file(self, tmp_path, monkeypatch): + cache_file = tmp_path / "cache.json" + monkeypatch.setattr(debian_todo, "CACHE_PATH", cache_file) + + debian_todo.save_cache(["/nonexistent/file"], {}) + + # Should not create cache file if source file is missing + assert not cache_file.exists() diff --git a/tests/test_filter.py b/tests/test_filter.py new file mode 100644 index 0000000..c2a7ab7 --- /dev/null +++ b/tests/test_filter.py @@ -0,0 +1,117 @@ +"""Tests for TODO list filtering functions.""" + +import pytest + +from debian_todo import filter_todo_list, summarize_sources + + +class TestFilterTodoList: + """Tests for filter_todo_list function.""" + + def test_filters_non_upstream_items(self): + todo_list = [ + {":shortname": "newupstream_foo", ":details": "1.0.0", ":source": "foo"}, + {":shortname": "rc_bug", ":details": "bug info", ":source": "bar"}, + {":shortname": "other_type", ":details": "other", ":source": "baz"}, + ] + result = filter_todo_list(todo_list) + assert len(result) == 1 + assert result[0][":source"] == "foo" + + def test_filters_prerelease_by_default(self): + todo_list = [ + {":shortname": "newupstream_foo", ":details": "1.0.0", ":source": "foo"}, + {":shortname": "newupstream_bar", ":details": "2.0.0rc1", ":source": "bar"}, + ] + result = filter_todo_list(todo_list) + assert len(result) == 1 + assert result[0][":source"] == "foo" + + def test_includes_prerelease_when_requested(self): + todo_list = [ + {":shortname": "newupstream_foo", ":details": "1.0.0", ":source": "foo"}, + {":shortname": "newupstream_bar", ":details": "2.0.0rc1", ":source": "bar"}, + ] + result = filter_todo_list(todo_list, include_prerelease=True) + assert len(result) == 2 + + def test_filters_matching_versions(self): + # When normalized versions match, the item should be filtered out + todo_list = [ + { + ":shortname": "newupstream_foo", + ":details": "1.0.0 (currently in unstable: 1.0.0-1)", + ":source": "foo", + }, + ] + result = filter_todo_list(todo_list) + assert len(result) == 0 + + def test_keeps_different_versions(self): + todo_list = [ + { + ":shortname": "newupstream_foo", + ":details": "1.1.0 (currently in unstable: 1.0.0-1)", + ":source": "foo", + }, + ] + result = filter_todo_list(todo_list) + assert len(result) == 1 + + def test_handles_epoch_in_current(self): + # Epoch should be stripped for comparison + todo_list = [ + { + ":shortname": "newupstream_foo", + ":details": "1.0.0 (currently in unstable: 1:1.0.0-1)", + ":source": "foo", + }, + ] + result = filter_todo_list(todo_list) + assert len(result) == 0 + + def test_skips_invalid_items(self): + todo_list = [ + {":shortname": "newupstream_foo", ":source": "foo"}, # missing :details + {":details": "1.0.0", ":source": "bar"}, # missing :shortname + {":shortname": 123, ":details": "1.0.0", ":source": "baz"}, # wrong type + ] + result = filter_todo_list(todo_list) + assert len(result) == 0 + + def test_empty_list(self): + assert filter_todo_list([]) == [] + + +class TestSummarizeSources: + """Tests for summarize_sources function.""" + + def test_extracts_sources(self): + todo_list = [ + {":source": "foo", ":shortname": "newupstream_foo"}, + {":source": "bar", ":shortname": "newupstream_bar"}, + {":source": "baz", ":shortname": "rc_bug"}, + ] + result = summarize_sources(todo_list) + assert result == {"foo", "bar", "baz"} + + def test_deduplicates_sources(self): + todo_list = [ + {":source": "foo", ":shortname": "type1"}, + {":source": "foo", ":shortname": "type2"}, + ] + result = summarize_sources(todo_list) + assert result == {"foo"} + + def test_skips_non_string_sources(self): + todo_list = [ + {":source": "foo"}, + {":source": 123}, + {":source": None}, + {}, + ] + result = summarize_sources(todo_list) + assert result == {"foo"} + + def test_empty_list(self): + assert summarize_sources([]) == set() diff --git a/tests/test_notes.py b/tests/test_notes.py new file mode 100644 index 0000000..b50357c --- /dev/null +++ b/tests/test_notes.py @@ -0,0 +1,57 @@ +"""Tests for notes loading.""" + +import pytest +from pathlib import Path + +import debian_todo + + +class TestLoadNotes: + """Tests for load_notes function.""" + + def test_loads_notes_from_file(self, tmp_path, monkeypatch): + notes_file = tmp_path / "notes" + notes_file.write_text("foo some note\nbar another note\n") + monkeypatch.setattr(debian_todo, "NOTES_PATH", notes_file) + + result = debian_todo.load_notes() + assert result == {"foo": "some note", "bar": "another note"} + + def test_returns_empty_when_file_missing(self, tmp_path, monkeypatch): + notes_file = tmp_path / "notes" + monkeypatch.setattr(debian_todo, "NOTES_PATH", notes_file) + + result = debian_todo.load_notes() + assert result == {} + + def test_handles_package_without_note(self, tmp_path, monkeypatch): + notes_file = tmp_path / "notes" + notes_file.write_text("foo\n") + monkeypatch.setattr(debian_todo, "NOTES_PATH", notes_file) + + result = debian_todo.load_notes() + assert result == {"foo": ""} + + def test_joins_multiple_notes(self, tmp_path, monkeypatch): + notes_file = tmp_path / "notes" + notes_file.write_text("foo first note\nfoo second note\n") + monkeypatch.setattr(debian_todo, "NOTES_PATH", notes_file) + + result = debian_todo.load_notes() + assert result == {"foo": "first note; second note"} + + def test_skips_blank_lines(self, tmp_path, monkeypatch): + notes_file = tmp_path / "notes" + notes_file.write_text("foo note one\n\nbar note two\n\n") + monkeypatch.setattr(debian_todo, "NOTES_PATH", notes_file) + + result = debian_todo.load_notes() + assert result == {"foo": "note one", "bar": "note two"} + + def test_handles_multiword_notes(self, tmp_path, monkeypatch): + notes_file = tmp_path / "notes" + notes_file.write_text("foo this is a long note with many words\n") + monkeypatch.setattr(debian_todo, "NOTES_PATH", notes_file) + + result = debian_todo.load_notes() + assert result == {"foo": "this is a long note with many words"} diff --git a/tests/test_vcs.py b/tests/test_vcs.py new file mode 100644 index 0000000..5d12c37 --- /dev/null +++ b/tests/test_vcs.py @@ -0,0 +1,71 @@ +"""Tests for VCS and uploader related functions.""" + +import pytest + +from debian_todo import vcs_git_to_team, normalize_uploaders, HIDE_UPLOADER + + +class TestVcsGitToTeam: + """Tests for vcs_git_to_team function.""" + + def test_salsa_python_team(self): + url = "https://salsa.debian.org/python-team/packages/python-foo.git" + assert vcs_git_to_team(url) == "python-team" + + def test_salsa_homeassistant_team(self): + url = "https://salsa.debian.org/homeassistant-team/python-bar.git" + assert vcs_git_to_team(url) == "homeassistant-team" + + def test_salsa_personal_repo(self): + url = "https://salsa.debian.org/username/package.git" + assert vcs_git_to_team(url) == "username" + + def test_non_salsa_url(self): + url = "https://github.com/user/repo.git" + assert vcs_git_to_team(url) == url + + def test_none_input(self): + assert vcs_git_to_team(None) is None + + def test_empty_string(self): + assert vcs_git_to_team("") is None + + def test_salsa_with_branch(self): + url = "https://salsa.debian.org/python-team/packages/foo.git -b debian/main" + assert vcs_git_to_team(url) == "python-team" + + +class TestNormalizeUploaders: + """Tests for normalize_uploaders function.""" + + def test_single_uploader(self): + uploaders = "John Doe " + assert normalize_uploaders(uploaders) == "John Doe " + + def test_multiple_uploaders(self): + uploaders = "John Doe , Jane Doe " + result = normalize_uploaders(uploaders) + assert result == "John Doe \nJane Doe " + + def test_hides_configured_uploader(self): + uploaders = f"John Doe , {HIDE_UPLOADER}" + result = normalize_uploaders(uploaders) + assert result == "John Doe " + assert HIDE_UPLOADER not in result + + def test_only_hidden_uploader(self): + uploaders = HIDE_UPLOADER + assert normalize_uploaders(uploaders) == "" + + def test_strips_whitespace(self): + uploaders = " John Doe , Jane Doe " + result = normalize_uploaders(uploaders) + assert result == "John Doe \nJane Doe " + + def test_empty_string(self): + assert normalize_uploaders("") == "" + + def test_extra_commas(self): + uploaders = "John Doe ,, Jane Doe ," + result = normalize_uploaders(uploaders) + assert result == "John Doe \nJane Doe " diff --git a/tests/test_version.py b/tests/test_version.py new file mode 100644 index 0000000..6c57861 --- /dev/null +++ b/tests/test_version.py @@ -0,0 +1,104 @@ +"""Tests for version parsing and normalization functions.""" + +import pytest + +from debian_todo import ( + parse_details, + is_prerelease_version, + normalize_upstream_version, +) + + +class TestParseDetails: + """Tests for parse_details function.""" + + def test_with_current_version(self): + details = "1.2.3 (currently in unstable: 1.2.2-1)" + new, current = parse_details(details) + assert new == "1.2.3" + assert current == "1.2.2-1" + + def test_without_current_version(self): + details = "1.2.3" + new, current = parse_details(details) + assert new == "1.2.3" + assert current is None + + def test_with_epoch_in_current(self): + details = "2.0.0 (currently in unstable: 1:1.9.0-2)" + new, current = parse_details(details) + assert new == "2.0.0" + assert current == "1:1.9.0-2" + + def test_strips_whitespace(self): + details = " 1.2.3 (currently in unstable: 1.2.2-1 ) " + new, current = parse_details(details) + assert new == "1.2.3" + assert current == "1.2.2-1" + + def test_empty_string(self): + new, current = parse_details("") + assert new == "" + assert current is None + + +class TestIsPrereleaseVersion: + """Tests for is_prerelease_version function.""" + + def test_alpha_version(self): + assert is_prerelease_version("1.0.0alpha1") is True + assert is_prerelease_version("1.0.0-alpha2") is True + assert is_prerelease_version("1.0.0.alpha") is True + + def test_beta_version(self): + assert is_prerelease_version("1.0.0beta1") is True + assert is_prerelease_version("1.0.0-beta2") is True + assert is_prerelease_version("2.0b1") is True + + def test_rc_version(self): + assert is_prerelease_version("1.0.0rc1") is True + assert is_prerelease_version("1.0.0-rc2") is True + assert is_prerelease_version("1.0.0.rc3") is True + + def test_short_prerelease(self): + assert is_prerelease_version("1.0a1") is True + assert is_prerelease_version("1.0b2") is True + + def test_stable_version(self): + assert is_prerelease_version("1.0.0") is False + assert is_prerelease_version("2.3.4") is False + assert is_prerelease_version("1.0.0-1") is False + + def test_version_with_current(self): + assert is_prerelease_version("1.0.0rc1 (currently in unstable: 0.9.0-1)") is True + assert is_prerelease_version("1.0.0 (currently in unstable: 0.9.0-1)") is False + + def test_case_insensitive(self): + assert is_prerelease_version("1.0.0Alpha1") is True + assert is_prerelease_version("1.0.0BETA1") is True + assert is_prerelease_version("1.0.0RC1") is True + + +class TestNormalizeUpstreamVersion: + """Tests for normalize_upstream_version function.""" + + def test_simple_version(self): + assert normalize_upstream_version("1.2.3") == "1.2.3" + + def test_strips_debian_revision(self): + assert normalize_upstream_version("1.2.3-1") == "1.2.3" + assert normalize_upstream_version("1.2.3-2ubuntu1") == "1.2.3" + + def test_strips_epoch(self): + assert normalize_upstream_version("1:1.2.3") == "1.2.3" + assert normalize_upstream_version("2:1.2.3") == "1.2.3" + + def test_strips_epoch_and_revision(self): + assert normalize_upstream_version("1:1.2.3-4") == "1.2.3" + + def test_strips_whitespace(self): + assert normalize_upstream_version(" 1.2.3 ") == "1.2.3" + + def test_version_with_hyphen_in_upstream(self): + # Only the last hyphen is treated as revision separator + assert normalize_upstream_version("1.2.3-beta-1") == "1.2.3-beta" diff --git a/todo b/todo index 9327ee7..64e6bb1 100755 --- a/todo +++ b/todo @@ -1,420 +1,7 @@ #!/usr/bin/python3 -"""CLI tool for tracking Debian packages with new upstream versions. - -Fetches TODO items from the Debian UDD (Ultimate Debian Database) and displays -packages where a new upstream version is available. Filters out pre-release -versions and shows team/uploader metadata. -""" - -import json -import re -import glob -import os -from pathlib import Path -from typing import Any, Optional, cast -from urllib.request import urlopen - -import click -from debian import deb822 -from rich.console import Console -from rich.table import Table - - -TODO_URL = "https://udd.debian.org/dmd/?email1=edward%404angle.com&format=json" -TODO_PATH = Path("todo.json") -NOTES_PATH = Path("notes") -TodoItem = dict[str, Any] -TodoList = list[TodoItem] - -PRERELEASE_RE = re.compile(r"(?i)(?:^|[0-9.\-])(?:alpha|beta|rc|a|b)\d*") -CURRENTLY_RE = re.compile( - r"^(?P.+?)\s*\(currently in unstable:\s*(?P.+?)\)\s*$" -) -CACHE_PATH = Path(".vcs_git_cache.json") -CACHE_VERSION = 4 -SourceInfo = dict[str, str] -HIDE_UPLOADER = "Edward Betts " - - -def parse_details(details: str) -> tuple[str, Optional[str]]: - """Parse version details string into new and current versions. - - Args: - details: String like "1.2.3 (currently in unstable: 1.2.2-1)" - - Returns: - Tuple of (new_version, current_version). current_version is None - if not present in the input. - """ - match = CURRENTLY_RE.match(details) - if match: - return match.group("new").strip(), match.group("current").strip() - return details.strip(), None - - -def is_prerelease_version(details: str) -> bool: - """Check if the new version in details is a pre-release. - - Detects versions containing alpha, beta, rc, a, or b suffixes. - """ - new_version, _ = parse_details(details) - return bool(PRERELEASE_RE.search(new_version)) - - -def vcs_git_to_team(vcs_git: Optional[str]) -> Optional[str]: - """Extract team name from a Vcs-Git URL. - - For salsa.debian.org URLs, extracts the group/team name. - Returns the full URL for non-Salsa repositories. - """ - if not vcs_git: - return None - match = re.search(r"salsa\.debian\.org/([^/]+)/", vcs_git) - if not match: - return vcs_git - return match.group(1) - - -def normalize_uploaders(uploaders: str) -> str: - """Clean and format uploaders string. - - Splits comma-separated uploaders, removes the configured HIDE_UPLOADER, - and joins with newlines. - """ - parts = [part.strip().strip(",") for part in uploaders.split(",")] - cleaned = [part for part in parts if part and part != HIDE_UPLOADER] - 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. - - Returns None if cache is missing, corrupted, or stale (based on - Sources file mtimes or cache version mismatch). - """ - 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): - 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)): - return None - - vcs_by_source = data.get("vcs_by_source") - if not isinstance(vcs_by_source, dict): - 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] = {"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 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 - data = { - "cache_version": CACHE_VERSION, - "sources_mtimes": sources_mtimes, - "vcs_by_source": vcs_by_source, - } - try: - with CACHE_PATH.open("w", encoding="utf-8") as handle: - json.dump(data, handle, sort_keys=True) - except OSError: - return - - -def load_source_info_map() -> dict[str, SourceInfo]: - """Load Vcs-Git and uploader info for all source packages. - - Parses APT Sources files from /var/lib/apt/lists/ and extracts - Vcs-Git URLs and Uploaders fields. Results are cached to disk. - """ - source_paths = sorted(glob.glob("/var/lib/apt/lists/*Sources")) - cached = load_cache(source_paths) - if cached is not None: - return cached - - vcs_by_source: dict[str, SourceInfo] = {} - for path in source_paths: - with Path(path).open("r", encoding="utf-8", errors="replace") as handle: - for entry in deb822.Deb822.iter_paragraphs(handle): - source = entry.get("Source") or entry.get("Package") - if not source or source in vcs_by_source: - continue - vcs_git = entry.get("Vcs-Git") - uploaders = entry.get("Uploaders") - team = None - if vcs_git: - team = vcs_git_to_team(vcs_git.strip()) - uploaders_text = "" - if uploaders: - uploaders_text = normalize_uploaders( - re.sub(r"\s+", " ", uploaders).strip() - ) - if team or uploaders_text: - vcs_by_source[source] = { - "vcs_git": team or "", - "uploaders": uploaders_text, - } - save_cache(source_paths, vcs_by_source) - return vcs_by_source - - -def fetch_todo_list() -> TodoList: - """Fetch the TODO list from UDD as JSON.""" - with urlopen(TODO_URL) as response: - payload = response.read().decode("utf-8") - return cast(TodoList, json.loads(payload)) - - -def save_todo_list(todo_list: TodoList) -> None: - """Save TODO list to local JSON file.""" - with TODO_PATH.open("w", encoding="utf-8") as handle: - json.dump(todo_list, handle, indent=2, ensure_ascii=True) - handle.write("\n") - - -def summarize_sources(todo_list: TodoList) -> set[str]: - """Extract set of source package names from TODO list.""" - sources: set[str] = set() - for item in todo_list: - source = item.get(":source") - if isinstance(source, str): - sources.add(source) - return sources - - -def load_notes() -> dict[str, str]: - """Load per-package notes from the notes file. - - Each line should be: - Multiple notes for the same package are joined with semicolons. - """ - if not NOTES_PATH.exists(): - return {} - notes_by_source: dict[str, list[str]] = {} - with NOTES_PATH.open("r", encoding="utf-8") as handle: - for line in handle: - stripped = line.strip() - if not stripped: - continue - parts = stripped.split(None, 1) - if not parts: - continue - source = parts[0] - note = parts[1] if len(parts) > 1 else "" - notes_by_source.setdefault(source, []).append(note) - return { - source: "; ".join(note for note in notes if note) - for source, notes in notes_by_source.items() - } - - -def filter_todo_list(todo_list: TodoList, include_prerelease: bool = False) -> TodoList: - """Filter TODO list to only new upstream version items. - - Removes non-upstream items, pre-releases (unless include_prerelease=True), - and items where normalized versions already match. - """ - filtered: TodoList = [] - for item in todo_list: - shortname = item.get(":shortname") - details = item.get(":details") - if not isinstance(shortname, str) or not isinstance(details, str): - continue - if not shortname.startswith("newupstream_"): - continue - if not include_prerelease and is_prerelease_version(details): - continue - new_version, current_version = parse_details(details) - if current_version: - normalized_new = normalize_upstream_version(new_version) - normalized_current = normalize_upstream_version(current_version) - if normalized_new == normalized_current: - continue - filtered.append(item) - return filtered - - -def normalize_upstream_version(version: str) -> str: - """Strip epoch and Debian revision from version string. - - "1:2.3.4-5" -> "2.3.4" - """ - if ":" in version: - version = version.split(":", 1)[1] - if "-" in version: - version = version.rsplit("-", 1)[0] - return version.strip() - - -def print_changes(old_list: TodoList, new_list: TodoList) -> None: - """Print added and removed packages between two TODO lists.""" - - def format_details(details: str) -> str: - new_version, current_version = parse_details(details) - display_new = new_version - if is_prerelease_version(details): - display_new = f"{new_version} (pre)" - return f"New: {display_new} | Current: {current_version or '-'}" - - def build_details(todo_list: TodoList) -> dict[str, str]: - details_by_source: dict[str, str] = {} - for item in todo_list: - source = item.get(":source") - details = item.get(":details") - if isinstance(source, str) and isinstance(details, str): - details_by_source[source] = format_details(details) - return details_by_source - - old_details = build_details(old_list) - new_details = build_details(new_list) - old_sources = set(old_details) - new_sources = set(new_details) - added = sorted(new_sources - old_sources) - removed = sorted(old_sources - new_sources) - if not added and not removed: - print("No changes in todo.json.") - return - if added: - print("New packages:") - for source in added: - print(f" {source} - {new_details.get(source, '-')}") - if removed: - print("Removed packages:") - for source in removed: - print(f" {source} - {old_details.get(source, '-')}") - - -def list_todos(include_prerelease: bool) -> None: - """Display filtered TODO items in a table. - - Downloads todo.json from UDD if not present locally. - """ - if not TODO_PATH.exists(): - print("Downloading todo.json...") - todo_list = fetch_todo_list() - save_todo_list(todo_list) - else: - with TODO_PATH.open("r", encoding="utf-8") as handle: - todo_list = cast(TodoList, json.load(handle)) - - source_info_map = load_source_info_map() - notes_by_source = load_notes() - console = Console() - filtered = filter_todo_list(todo_list, include_prerelease=include_prerelease) - is_narrow = console.width < 100 - if is_narrow: - console.print("Debian New Upstream TODOs") - else: - table = Table(title="Debian New Upstream TODOs") - 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("Team", justify="right") - table.add_column("Note/Uploaders", overflow="fold") - - for todo in filtered: - new_version, current_version = parse_details(todo[":details"]) - source_info = source_info_map.get(todo[":source"], {}) - vcs_git = source_info.get("vcs_git") - uploaders = source_info.get("uploaders", "") - source = todo[":source"] - note = notes_by_source.get(source, "") - display_new = new_version - if is_prerelease_version(todo[":details"]): - display_new = f"[yellow]{new_version} (pre)[/yellow]" - display_team = vcs_git or "-" - if display_team == "homeassistant-team": - display_team = "HA" - elif display_team.endswith("-team"): - display_team = display_team[:-5] - display_note = note or uploaders or "-" - - if is_narrow: - parts = [f"[bold]{source}[/bold]"] - if display_team != "-": - parts.append(f"[dim]{display_team}[/dim]") - parts.append(f"N: {display_new}") - parts.append(f"C: {current_version or '-'}") - console.print(" ".join(parts)) - if display_note != "-": - console.print(" " + display_note) - else: - table.add_row( - source, - display_new, - current_version or "-", - display_team, - display_note, - ) - if not is_narrow: - console.print(table) - console.print(f"Packages: {len(filtered)}") - -def update_todos() -> None: - """Fetch latest TODO list from UDD and show changes.""" - old_list: TodoList = [] - if TODO_PATH.exists(): - with TODO_PATH.open("r", encoding="utf-8") as handle: - old_list = cast(TodoList, json.load(handle)) - todo_list = fetch_todo_list() - save_todo_list(todo_list) - print_changes(filter_todo_list(old_list), filter_todo_list(todo_list)) - -@click.group(invoke_without_command=True) -@click.pass_context -def cli(context: click.Context) -> None: - """Track Debian packages with new upstream versions.""" - if context.invoked_subcommand is None: - list_todos(include_prerelease=False) - -@cli.command("list", help="List filtered new upstream todo entries.") -@click.option( - "--show-prerelease", - is_flag=True, - help="Include prerelease versions in the list output.", -) -def list_command(show_prerelease: bool) -> None: - list_todos(include_prerelease=show_prerelease) - -@cli.command("update", help="Fetch the latest todo list and show changes.") -def update_command() -> None: - update_todos() +"""CLI entry point for debian-todo.""" +from debian_todo import main if __name__ == "__main__": - cli() + main()