"""AUR client tests. Network is never required: `fetch` takes an injectable opener, so the tests drive it with a fake transport and stay fast and deterministic. """ from __future__ import annotations import io import json import sys import tempfile from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from updater import aur # noqa: E402 class FakeResponse(io.BytesIO): def __enter__(self): return self def __exit__(self, *exc): return False def fake_transport(payloads: dict[str, dict], seen: list[str] | None = None, fail: bool = False): """An opener that answers RPC queries from a name -> raw-result mapping.""" def opener(url: str, timeout: float): if seen is not None: seen.append(url) if fail: raise OSError("network down") query = url.split("?", 1)[1] names = [ v for k, v in (p.split("=", 1) for p in query.split("&")) if k == "arg%5B%5D" or k == "arg[]" ] import urllib.parse names = [urllib.parse.unquote(n) for n in names] results = [payloads[n] for n in names if n in payloads] return FakeResponse(json.dumps({"resultcount": len(results), "results": results}).encode()) return opener RAW = { "demo": { "Name": "demo", "Version": "1.2.3-1", "Description": "a demo", "Maintainer": "alice", "OutOfDate": None, "NumVotes": 12, "Popularity": 1.5, "FirstSubmitted": 1000, "LastModified": 2000, "URL": "https://demo.invalid", "PackageBase": "demo", "Depends": ["glibc"], "MakeDepends": ["cmake"], "CoMaintainers": ["bob"], }, "orphan": { "Name": "orphan", "Version": "0.1-1", "Maintainer": None, "OutOfDate": 1700000000, "NumVotes": 0, "Popularity": 0.0, "FirstSubmitted": 500, "LastModified": 900, "PackageBase": "orphan", }, } # -------------------------------------------------------------------------- def test_parse_rpc_payload(): pkgs = aur.fetch(["demo"], opener=fake_transport(RAW)) p = pkgs["demo"] assert p.version == "1.2.3-1" assert p.maintainer == "alice" assert p.comaintainers == ("bob",) assert p.votes == 12 assert p.depends == ("glibc",) assert p.make_depends == ("cmake",) assert not p.is_orphaned and not p.is_flagged def test_orphan_and_flagged_flags(): p = aur.fetch(["orphan"], opener=fake_transport(RAW))["orphan"] assert p.is_orphaned assert p.is_flagged assert p.is_orphaned and p.out_of_date == 1700000000 def test_unknown_names_are_omitted_not_fatal(): """A package deleted from the AUR simply does not come back.""" pkgs = aur.fetch(["demo", "gone"], opener=fake_transport(RAW)) assert set(pkgs) == {"demo"} assert aur.deleted(["demo", "gone"], pkgs) == ["gone"] def test_generated_debug_packages_are_not_deletions(): """makepkg splits -debug out of a base package; it was never an AUR entry.""" pkgs = aur.fetch(["demo"], opener=fake_transport(RAW)) assert aur.deleted(["demo", "demo-debug"], pkgs) == [] # but a -debug package with no installed base really is unknown assert aur.deleted(["demo", "nothing-debug"], pkgs) == ["nothing-debug"] def test_debug_suffix_accepted_when_the_base_is_only_installed_locally(): pkgs: dict = {} assert aur.deleted(["thing-debug"], pkgs, known_bases={"thing"}) == [] assert aur.deleted(["thing-debug"], pkgs) == ["thing-debug"] def test_batching_splits_large_requests(): seen: list[str] = [] names = [f"pkg{i}" for i in range(aur.BATCH + 10)] aur.fetch(names, opener=fake_transport({}, seen=seen)) assert len(seen) == 2, f"expected 2 requests, got {len(seen)}" assert "arg%5B%5D=pkg0" in seen[0] or "arg%5B%5D=pkg0" in seen[0] def test_network_failure_raises_a_typed_error(): try: aur.fetch(["demo"], opener=fake_transport(RAW, fail=True)) except aur.AurError: pass else: raise AssertionError("expected AurError when the transport fails") def test_missing_fields_do_not_crash(): pkgs = aur.fetch(["bare"], opener=fake_transport({"bare": {"Name": "bare"}})) p = pkgs["bare"] assert p.version == "" and p.maintainer is None and p.depends == () assert p.age_days == 0.0 # -------------------------------------------------------------------------- # state: this is what makes "adopted by a new maintainer" possible # -------------------------------------------------------------------------- def test_state_round_trips_and_reports_maintainer_change(): with tempfile.TemporaryDirectory() as td: path = Path(td) / "aur-state.json" state = aur.State.load(path) assert state.previous_maintainer("demo") is None state.record("demo", maintainer="alice", commit="abc123", version="1.0-1") state.save() reloaded = aur.State.load(path) assert reloaded.previous_maintainer("demo") == "alice" assert reloaded.previous_commit("demo") == "abc123" # the package changes hands reloaded.record("demo", maintainer="carol", commit="def456") assert reloaded.previous_maintainer("demo") == "carol" def test_state_survives_a_corrupt_file(): with tempfile.TemporaryDirectory() as td: path = Path(td) / "aur-state.json" path.write_text("{ this is not json") state = aur.State.load(path) assert state.entries == {} state.record("x", maintainer="y") state.save() assert aur.State.load(path).previous_maintainer("x") == "y" def test_state_save_is_atomic_leaving_no_temp_file(): with tempfile.TemporaryDirectory() as td: path = Path(td) / "aur-state.json" state = aur.State.load(path) state.record("demo", maintainer="alice") state.save() assert path.is_file() assert not list(Path(td).glob("*.tmp")) def test_state_ignores_unknown_fields_from_a_newer_version(): with tempfile.TemporaryDirectory() as td: path = Path(td) / "aur-state.json" path.write_text( json.dumps( { "version": 99, "packages": { "demo": {"maintainer": "alice", "commit": "c", "something_new": True} }, } ) ) state = aur.State.load(path) assert state.previous_maintainer("demo") == "alice" def main() -> int: tests = [(n, f) for n, f in sorted(globals().items()) if n.startswith("test_") and callable(f)] failed = 0 for name, fn in tests: try: fn() except Exception as exc: # noqa: BLE001 failed += 1 print(f"FAIL {name}\n {type(exc).__name__}: {exc}") else: print(f"ok {name}") print(f"\n{len(tests) - failed}/{len(tests)} passed") return 1 if failed else 0 if __name__ == "__main__": raise SystemExit(main())