"""pacman's alpm hooks.

These decide what a plan says will happen around a transaction, so getting them
wrong means telling someone their kernel will not be rebuilt when it will, or
the reverse. Parsed from the real files, never executed.
"""

from __future__ import annotations

import sys
import tempfile
from pathlib import Path

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

from updater import hooks  # noqa: E402

SIMPLE = """\
[Trigger]
Operation = Upgrade
Type = Package
Target = linux

[Action]
Description = Updating linux initcpios...
When = PostTransaction
Exec = /usr/bin/mkinitcpio -P
"""

SNAP_PAC = """\
[Trigger]
Operation = Upgrade
Operation = Install
Operation = Remove
Type = Package
Target = *

[Action]
Description = Performing snapper pre snapshots for the following configurations...
Depends = snap-pac
When = PreTransaction
Exec = /usr/share/libalpm/scripts/snap-pac pre
NeedsTargets
AbortOnFail
"""

PATH_TRIGGER = """\
[Trigger]
Type = Path
Operation = Install
Operation = Upgrade
Target = usr/lib/systemd/system/*

[Action]
Description = Reloading system manager configuration...
When = PostTransaction
Exec = /usr/share/libalpm/scripts/systemd-hook daemon-reload
"""

MULTI_TRIGGER = """\
[Trigger]
Operation = Install
Operation = Upgrade
Target = glibc
Type = Package

[Trigger]
Operation = Upgrade
Target = etc/ld.so.conf
Type = Path

[Action]
Description = Configuring dynamic linker run-time bindings...
Exec = /usr/bin/ldconfig -r .
When = PostTransaction
"""


def write(directory: Path, name: str, text: str) -> Path:
    path = directory / name
    path.write_text(text)
    return path


def parse_one(text: str):
    with tempfile.TemporaryDirectory() as td:
        path = write(Path(td), "x.hook", text)
        return hooks._parse(path)


# --------------------------------------------------------------------------
# the things that were wrong
# --------------------------------------------------------------------------


def test_a_bare_flag_does_not_discard_the_whole_hook():
    """`NeedsTargets` has no `=`, and a strict INI reader drops the file."""
    hook = parse_one(SNAP_PAC)
    assert hook is not None, "a hook with a bare flag was dropped entirely"
    assert hook.needs_targets and hook.abort_on_fail
    assert hook.exec_.endswith("snap-pac pre")


def test_repeated_operation_keys_are_all_kept():
    """Three `Operation =` lines, not the last one only."""
    hook = parse_one(SNAP_PAC)
    assert hook is not None
    ops = set(hook.triggers[0].operations)
    assert ops == {"Upgrade", "Install", "Remove"}, ops


def test_a_hook_can_declare_several_triggers():
    hook = parse_one(MULTI_TRIGGER)
    assert hook is not None
    assert len(hook.triggers) == 2, [t.kind for t in hook.triggers]
    assert {t.kind for t in hook.triggers} == {"Package", "Path"}


def test_a_wildcard_trigger_fires_for_any_package():
    hook = parse_one(SNAP_PAC)
    assert hook.triggers[0].matches_name("anything-at-all")


def test_operations_actually_gate_the_trigger():
    """A Remove-only trigger must not fire for an upgrade."""
    hook = parse_one("""\
[Trigger]
Operation = Remove
Type = Package
Target = snap-pac

[Action]
Exec = /bin/true
""")
    trigger = hook.triggers[0]
    assert trigger.matches_name("snap-pac", "Remove")
    assert not trigger.matches_name("snap-pac", "Upgrade")


def test_comments_and_blank_lines_are_ignored():
    hook = parse_one("""\
# a leading comment
[Trigger]
# inside
Operation = Upgrade
Type = Package
Target = x

[Action]
# trailing
Exec = /bin/true
""")
    assert hook is not None and hook.triggers[0].targets == ("x",)


# --------------------------------------------------------------------------
# matching
# --------------------------------------------------------------------------


def test_a_package_trigger_matches_globs():
    hook = parse_one(SIMPLE)
    assert hook.triggers[0].matches_name("linux")
    assert not hook.triggers[0].matches_name("linux-headers")


def test_a_path_trigger_matches_files_the_package_owns():
    hook = parse_one(PATH_TRIGGER)
    trigger = hook.triggers[0]
    assert trigger.matches_path("/usr/lib/systemd/system/foo.service")
    assert trigger.matches_path("usr/lib/systemd/system/foo.service")
    assert not trigger.matches_path("/etc/other")


def test_fires_for_consults_both_trigger_kinds():
    hook = parse_one(MULTI_TRIGGER)
    assert hook.fires_for({"glibc"}, {})
    assert hook.fires_for({"unrelated"}, {"unrelated": ["etc/ld.so.conf"]})
    assert not hook.fires_for({"unrelated"}, {"unrelated": ["etc/hosts"]})


def test_snapshot_hooks_are_recognised():
    assert parse_one(SNAP_PAC).is_snapshot
    assert not parse_one(SIMPLE).is_snapshot


def test_a_label_falls_back_to_the_command_then_the_filename():
    with_desc = parse_one(SIMPLE)
    assert with_desc.label == "Updating linux initcpios"
    no_desc = parse_one("""\
[Trigger]
Operation = Upgrade
Type = Package
Target = x

[Action]
Exec = /usr/bin/foo --bar
""")
    assert no_desc.label == "foo"


# --------------------------------------------------------------------------
# loading a directory
# --------------------------------------------------------------------------


def test_a_hook_without_a_trigger_is_not_loaded():
    assert parse_one("[Action]\nExec = /bin/true\n") is None


def test_a_hook_without_an_action_is_not_loaded():
    assert parse_one("[Trigger]\nType = Package\nTarget = x\n") is None


def test_an_unreadable_file_is_skipped_not_fatal():
    with tempfile.TemporaryDirectory() as td:
        directory = Path(td)
        write(directory, "good.hook", SIMPLE)
        (directory / "broken.hook").write_bytes(b"\xff\xfe\x00 not utf-8")
        loaded = hooks.load((directory,))
        assert [h.path.name for h in loaded] == ["good.hook"]


def test_a_later_directory_overrides_by_filename():
    """That is what pacman does: /etc replaces the packaged hook."""
    with tempfile.TemporaryDirectory() as a, tempfile.TemporaryDirectory() as b:
        write(Path(a), "same.hook", SIMPLE)
        write(Path(b), "same.hook", SIMPLE.replace("linux", "other"))
        loaded = hooks.load((Path(a), Path(b)))
        assert len(loaded) == 1, [h.path for h in loaded]
        assert loaded[0].triggers[0].targets == ("other",)


def test_scheduled_splits_before_and_after():
    loaded = [parse_one(SNAP_PAC), parse_one(SIMPLE)]
    before, after = hooks.scheduled(loaded, {"linux"}, {})
    # both are called x.hook, so distinguish them by when they run
    assert len(before) == 1 and before[0].when == "PreTransaction"
    assert len(after) == 1 and after[0].when == "PostTransaction"


def test_the_real_hooks_on_this_machine_load():
    """A smoke test against whatever is actually installed."""
    loaded = hooks.load()
    if not loaded:
        return  # nothing installed; nothing to assert
    for hook in loaded:
        assert hook.triggers, hook.path
        for trigger in hook.triggers:
            assert trigger.kind in ("Package", "Path"), trigger.kind
            assert trigger.targets


def test_a_linux_upgrade_rebuilds_the_initramfs():
    """The single most useful thing this can get right.

    The mkinitcpio hook triggers on the *path* usr/lib/modules/*/vmlinuz, so it
    only fires when the file list is supplied. Matching on package names alone
    would miss it, which is exactly why the file lists are loaded.
    """
    loaded = hooks.load()
    if not loaded:
        return
    paths = {
        "linux": [
            "usr/lib/modules/6.12.1-arch1-1/vmlinuz",
            "usr/lib/modules/6.12.1-arch1-1/modules.dep",
        ]
    }
    _, after = hooks.scheduled(loaded, {"linux"}, paths)
    labels = " ".join(h.label.lower() for h in after)
    assert "initcpio" in labels, f"no initramfs hook fired:\n{labels}"


def test_path_triggers_are_not_matched_on_names_alone():
    """Without file lists the initramfs hook must not be claimed."""
    loaded = hooks.load()
    if not loaded:
        return
    _, after = hooks.scheduled(loaded, {"linux"}, {})
    labels = " ".join(h.label.lower() for h in after)
    assert "initcpio" not in labels, labels


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