"""repo.py tests.

The valuable ones assert agreement with pacman itself. Every number here was
originally checked by hand against `pacman -Q*`; this file keeps it that way,
because a repo reader that quietly disagrees with pacman is worse than no reader
at all -- it would report a plan that the transaction then contradicts.
"""

from __future__ import annotations

import shutil
import subprocess
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from updater import repo  # noqa: E402

PYALPM = True
PACMAN = shutil.which("pacman") is not None


def _code(fn) -> str:
    """A function's source with all whitespace removed.

    These assertions are about what the code does, and the exact line breaks
    are the formatter's business -- pinning them made every one of these tests
    fail the first time the tree was formatted.
    """

    return "".join(_code(fn).split())


def pacman_lines(*args: str) -> list[str]:
    proc = subprocess.run(["pacman", *args], capture_output=True, text=True, check=False)
    return [line for line in proc.stdout.splitlines() if line.strip()]


# --------------------------------------------------------------------------
# configuration
# --------------------------------------------------------------------------


def _code(fn) -> str:
    """A function's source with all whitespace removed.

    These assertions are about what the code does; where the line breaks fall
    is the formatter's business. Stripping whitespace from both the source and
    the snippet means a reformat cannot make them fail.
    """
    import inspect

    return _strip(inspect.getsource(fn))


def _strip(s: str) -> str:
    return "".join(s.split())


def test_read_conf_finds_the_real_repositories():
    config = repo.read_conf()
    assert config.repos, "no repositories parsed from pacman.conf"
    assert "options" not in config.repos, "'[options]' was mistaken for a repo"
    for name in ("core", "extra"):
        assert name in config.repos, name


def test_read_conf_missing_file_falls_back_instead_of_raising():
    config = repo.read_conf("/definitely/not/here/pacman.conf")
    assert config.repos
    assert config.ignore_pkg == frozenset()


def test_read_conf_parses_ignore_lists(tmp: Path | None = None):
    import tempfile

    with tempfile.TemporaryDirectory() as td:
        path = Path(td) / "pacman.conf"
        path.write_text(
            "[options]\n"
            "IgnorePkg = linux nvidia\n"
            "IgnoreGroup = modified\n"
            "\n"
            "[core]\n"
            "Include = /etc/pacman.d/mirrorlist\n"
            "\n"
            "[extra]\n"
        )
        config = repo.read_conf(path)
        assert set(config.ignore_pkg) == {"linux", "nvidia"}
        assert set(config.ignore_group) == {"modified"}
        assert config.repos == ("core", "extra")


# --------------------------------------------------------------------------
# the bug that made everything look up to date
# --------------------------------------------------------------------------


def test_connect_registers_every_configured_repo():
    """Without explicit registration pyalpm reports zero sync dbs.

    Which means zero updates, which is indistinguishable from a system that is
    fully up to date.
    """
    if not PACMAN:
        return
    config = repo.read_conf()
    with repo.connect() as handle:
        registered = {db.name for db in handle.get_syncdbs()}
    assert registered == set(config.repos), f"registered {registered}, expected {set(config.repos)}"


# --------------------------------------------------------------------------
# agreement with pacman
# --------------------------------------------------------------------------


def test_plan_agrees_with_pacman_on_the_number_of_upgrades():
    if not PACMAN:
        return
    with repo.connect() as handle:
        mine = len(repo.plan(handle))
    assert mine == len(pacman_lines("-Qu")), "plan disagrees with pacman -Qu"


def test_orphans_match_both_of_pacmans_strictness_levels():
    """A package merely *suggested* by another is not an orphan to `-Qtd`."""
    if not PACMAN:
        return
    with repo.connect() as handle:
        strict = repo.orphans(handle)
        loose = repo.orphans(handle, include_optional=True)
    assert len(strict) == len(pacman_lines("-Qtdq"))
    assert len(loose) == len(pacman_lines("-Qttdq"))
    # and the loose set is a superset, since it counts fewer things as required
    assert set(strict) <= set(loose)


def test_foreign_explicit_and_installed_agree_with_pacman():
    if not PACMAN:
        return
    with repo.connect() as handle:
        foreign = repo.foreign(handle)
        explicit = repo.explicit(handle)
        installed = repo.installed(handle)
    assert len(foreign) == len(pacman_lines("-Qqm"))
    assert len(explicit) == len(pacman_lines("-Qqe"))
    assert len(installed) == len(pacman_lines("-Qq"))
    # every foreign package is installed, by definition
    assert set(foreign) <= set(installed)
    assert set(explicit) <= set(installed)


def test_installed_reports_real_versions():
    with repo.connect() as handle:
        installed = repo.installed(handle)
    assert installed, "no packages reported as installed"
    assert "pacman" in installed
    assert installed["pacman"][0].isdigit(), installed["pacman"]


def test_plan_never_reports_a_foreign_package_as_a_removal():
    """AUR packages have no sync entry and `pacman -Su` leaves them alone."""
    with repo.connect() as handle:
        plan = repo.plan(handle)
        foreign = set(repo.foreign(handle))
    assert not hasattr(plan, "removals"), "the removals field came back"
    assert not (foreign & {u.name for u in plan.updates})


def test_plan_totals_are_sums_of_their_parts():
    with repo.connect() as handle:
        plan = repo.plan(handle)
    assert plan.download_total == sum(u.download for u in plan.updates)
    assert plan.installed_delta == sum(u.delta for u in plan.updates)
    assert len(plan) == len(plan.updates)


# --------------------------------------------------------------------------
# refresh machinery
# --------------------------------------------------------------------------


def test_sync_argv_uses_fakeroot_when_available():
    """pacman's sandbox refuses files it does not own, so fakeroot matters."""
    argv = repo._sync_argv("/tmp/example")
    assert "--dbpath" in argv and "/tmp/example" in argv
    if shutil.which("fakeroot"):
        assert argv[0] == "fakeroot"
        assert "--disable-sandbox-filesystem" in argv
    # and it must never be able to touch the real database
    assert "/var/lib/pacman" not in " ".join(argv)


def test_sync_db_age_is_a_number_or_none():
    age = repo.sync_db_age()
    assert age is None or age >= 0


def test_sync_db_age_of_a_missing_path_is_none():
    assert repo.sync_db_age("/definitely/not/here") is None


def test_preview_always_yields_a_usable_handle_even_when_sync_fails():
    """Offline, or a mirror outage, must degrade to the databases on disk."""
    with repo.preview(sync=False) as (handle, refreshed, note):
        assert refreshed is False
        assert note, "an unrefreshed preview must say so"
        # and it still answers questions rather than collapsing
        assert repo.installed(handle)
        assert isinstance(repo.plan(handle).updates, list)


def test_preview_reports_a_reason_when_it_cannot_sync():
    with repo.preview(sync=False) as (_handle, _refreshed, note):
        assert "skip" in note.lower() or "refresh" in note.lower()


def test_a_preview_almost_matching_pacman_proves_the_sync_path_reads_the_same_format():
    """With sync disabled the preview must equal a plain connect()."""
    with repo.preview(sync=False) as (preview_handle, _r, _n):
        preview_plan = repo.plan(preview_handle)
    with repo.connect() as handle:
        direct_plan = repo.plan(handle)
    assert len(preview_plan.updates) == len(direct_plan.updates)


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


# --------------------------------------------------------------------------
# sizes, against pacman's own numbers
# --------------------------------------------------------------------------


def _pacman_size(args: list[str], needle: str) -> dict[str, int]:
    """Parse `pacman -Qi/-Si` human sizes back into bytes, for comparison."""
    import re
    import subprocess

    out = subprocess.run(["pacman", *args], capture_output=True, text=True, check=False).stdout
    sizes: dict[str, int] = {}
    name = None
    units = {"B": 1, "KiB": 1024, "MiB": 1024**2, "GiB": 1024**3}
    for line in out.splitlines():
        if line.startswith("Name "):
            name = line.split(":", 1)[1].strip()
        elif needle in line and name:
            m = re.search(r"([\d.]+)\s*(\w+)", line.split(":", 1)[1])
            if m:
                sizes[name] = int(float(m.group(1)) * units.get(m.group(2), 1))
    return sizes


def test_installed_size_matches_pacman():
    """pyalpm's `isize` must agree with `pacman -Qi`.

    `size` is not a substitute: it is 0 on a local package and the download size
    on a sync one, which made every on-disk delta look like a full-size increase.
    """
    with repo.connect() as handle:
        installed = repo.installed(handle)
        names = sorted(installed)[:40]
        if not names:
            return
        truth = _pacman_size(["-Qi", *names], "Installed Size")
        local = {p.name: p for p in handle.get_localdb().pkgcache}

        checked = 0
        for name, expected in truth.items():
            pkg = local.get(name)
            if pkg is None or expected == 0:
                continue
            got = getattr(pkg, "isize", 0)
            # pacman rounds to two decimals, so allow a little slack
            assert abs(got - expected) <= max(2048, expected * 0.01), (
                f"{name}: isize={got:,} but pacman says {expected:,}"
            )
            checked += 1
        assert checked >= 5, f"only compared {checked} packages"


def test_sync_size_agrees_with_pacman_and_is_not_the_download_size():
    with repo.connect() as handle:
        best = repo._best_versions(handle)
        names = sorted(best)[:40]
        if not names:
            return
        installed_truth = _pacman_size(["-Si", *names], "Installed Size")
        download_truth = _pacman_size(["-Si", *names], "Download Size")

        compared = 0
        for name in names:
            sp = best.get(name)
            if sp is None:
                continue
            want_installed = installed_truth.get(name, 0)
            if want_installed == 0:
                continue
            got = getattr(sp, "isize", 0)
            assert abs(got - want_installed) <= max(2048, want_installed * 0.01), (
                f"{name}: isize={got:,} but pacman says {want_installed:,}"
            )
            # and it must not have been confused with the download size
            want_download = download_truth.get(name, 0)
            if want_download and want_installed != want_download:
                assert got != want_download, f"{name}: isize was populated with the download size"
            compared += 1
        assert compared >= 3, f"only compared {compared} packages"


def test_an_unchanged_package_has_a_zero_delta_not_a_full_size():
    """The regression: delta used to be the whole installed size."""
    with repo.connect() as handle:
        local = handle.get_localdb()
        best = repo._best_versions(handle)
        same = [
            (lp, best[lp.name])
            for lp in local.pkgcache
            if lp.name in best and best[lp.name].version == lp.version
        ]
        assert same, "no packages at their sync version to check"
        for lp, sp in same[:50]:
            delta = getattr(sp, "isize", 0) - getattr(lp, "isize", 0)
            assert delta == 0, f"{lp.name}: unchanged but delta is {delta:,}"
            assert delta != getattr(sp, "isize", 0) or getattr(sp, "isize", 0) == 0


def test_size_is_the_trap_and_isize_is_the_answer():
    """Documents the pyalpm semantics that caused the wrong-size bug.

    On a sync package `size` is the compressed download size, not the installed
    size; on a local package it is 0. `isize` is the installed size in both
    cases. A tiny package can legitimately download larger than it installs, so
    the only trustworthy check is against pacman, which the tests above do.
    """
    with repo.connect() as handle:
        best = repo._best_versions(handle)
        matched = 0
        for name, sp in list(best.items())[:300]:
            size = getattr(sp, "size", 0)
            download = getattr(sp, "download_size", 0)
            if size and download:
                assert size == download, f"{name}: sync `size` was expected to equal download_size"
                matched += 1
        assert matched >= 5, f"only checked {matched} sync packages"

        # and on a local package it carries nothing useful
        local = list(handle.get_localdb().pkgcache)[:200]
        assert local, "no local packages to check"
        zeroed = sum(1 for lp in local if getattr(lp, "size", 0) == 0)
        assert zeroed > len(local) // 2, (
            f"local `size` was 0 for only {zeroed}/{len(local)}; pyalpm changed"
        )


if __name__ == "__main__":
    raise SystemExit(main())


def test_ignore_pkg_is_a_glob_not_a_literal_name():
    """pacman.conf documents both lists as shell globs.

    Comparing for equality meant `IgnorePkg = linux*` matched nothing and the
    plan offered packages pacman would refuse to touch.
    """

    class P:
        def __init__(self, name, groups=()):
            self.name, self.groups = name, list(groups)

    class H:
        ignorepkgs = ["linux*", "nvidia"]
        ignoregrps = ["plasma*"]

    assert repo._is_ignored(H(), P("linux")) is True
    assert repo._is_ignored(H(), P("linux-lts")) is True
    assert repo._is_ignored(H(), P("nvidia")) is True
    assert repo._is_ignored(H(), P("mesa")) is False
    assert repo._is_ignored(H(), P("kwin", ["plasma-desktop"])) is True
    assert repo._is_ignored(H(), P("kwin", ["gnome"])) is False


def test_every_handle_registers_the_ignore_lists():
    """`connect()` did and `preview()` did not, and the plan uses preview.

    So `_is_ignored` always saw an empty list and offered every IgnorePkg
    package as a pending upgrade.
    """

    for fn in (repo.connect, repo.preview, repo._fallback):
        source = _code(fn)
        assert _strip("_register_ignores") in source, fn.__name__