"""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())