"""Reading build state out of /proc.

The observation layer replaces parsing paru's human output, which is brittle by
construction. That trade only pays if this side is right, and it is subtle:
`/proc/<pid>/stat` packs the command name inside parentheses *inside* a
space-separated field list, and the command name may itself contain spaces and
parentheses. Get it wrong and every field shifts, so the start time becomes
something else and a per-package timer reports nonsense.

These tests spawn real processes. Reading a real kernel is the only way to test
a /proc parser.
"""

from __future__ import annotations

import os
import subprocess
import sys
import time
from pathlib import Path

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

from updater import procwatch as pw  # noqa: E402


def spawn_with_comm(comm: str) -> subprocess.Popen:
    """A live process whose kernel command name is exactly `comm`.

    The trailing `sleep &` matters: a bare command at the end of `bash -c` is
    exec'd and replaces the shell, so the name we set would be thrown away.
    """
    return subprocess.Popen(
        ["bash", "-c", f'printf "%s" {comm!r} > /proc/self/comm; sleep 30 & wait']
    )


def kernel_comm(pid: int) -> str:
    raw = Path(f"/proc/{pid}/stat").read_text()
    return raw[raw.find("(") + 1 : raw.rfind(")")]


# --------------------------------------------------------------------------
# parsing /proc/<pid>/stat
# --------------------------------------------------------------------------


def test_the_command_name_may_contain_parentheses():
    """The reason the parser reads from the last `)`: the name is inside a field
    list, and can itself contain the delimiter."""
    for comm in ("ends)", ")", "a)b)c", "((", "))((", "weird (name)"):
        proc = spawn_with_comm(comm)
        try:
            time.sleep(0.25)
            got = pw.snapshot().get(proc.pid)
            assert got is not None, f"{comm!r}: process was not readable"
            assert got.comm == kernel_comm(proc.pid), (comm, got.comm)
        finally:
            proc.kill()
            proc.wait()


def test_the_fields_after_the_name_still_line_up():
    """A name full of parens must not shift ppid or the start time."""
    proc = spawn_with_comm("))((")
    try:
        time.sleep(0.4)
        got = pw.snapshot()[proc.pid]
        raw = Path(f"/proc/{proc.pid}/stat").read_text()
        fields = raw[raw.rfind(")") + 2 :].split()
        assert fields[0] in "RSDZTtWXxKWP", "field 0 should be the state letter"
        assert got.ppid == int(fields[1])
        uptime = float(Path("/proc/uptime").read_text().split()[0])
        age = uptime - (got.started - pw.boot_time())
        assert 0 <= age < 5, f"start time is off by {age:.1f}s"
    finally:
        proc.kill()
        proc.wait()


def test_a_dead_or_missing_process_is_not_an_error():
    """Processes die while being read; that is the normal case, not a failure."""
    proc = subprocess.Popen(["true"])
    proc.wait()
    btime, clk = pw.boot_time(), pw._clk_tck()
    for pid in (proc.pid, 0, 999999):
        assert pw.read_proc(pid, btime, clk) is None


def test_a_snapshot_survives_processes_dying_underneath_it():
    spawn = [subprocess.Popen(["true"]) for _ in range(40)]
    snap = pw.snapshot()
    assert os.getpid() in snap, "the reading process should always be present"
    for proc in spawn:
        proc.wait()


def test_started_time_agrees_with_ps():
    import datetime

    proc = subprocess.Popen(["sleep", "30"])
    try:
        time.sleep(0.3)
        got = pw.snapshot()[proc.pid]
        out = subprocess.run(
            ["ps", "-o", "lstart=", "-p", str(proc.pid)],
            capture_output=True,
            text=True,
            check=False,
        ).stdout.strip()
        if out:
            when = datetime.datetime.strptime(out, "%a %b %d %H:%M:%S %Y")
            assert abs(got.started - when.timestamp()) < 2
    finally:
        proc.kill()
        proc.wait()


# --------------------------------------------------------------------------
# walking the tree
# --------------------------------------------------------------------------


def test_descendants_walks_a_real_tree():
    proc = subprocess.Popen(["bash", "-c", "sleep 30 & sleep 30 & wait"])
    try:
        time.sleep(0.4)
        snap = pw.snapshot()
        kids = pw.descendants(proc.pid, snap)
        assert len(kids) == 2, [k.argv0 for k in kids]
        assert all(k.argv0 == "sleep" for k in kids)

        with_root = pw.descendants(proc.pid, snap, include_root=True)
        assert len(with_root) == 3
        assert proc.pid in {k.pid for k in with_root}
    finally:
        proc.kill()
        proc.wait()


def test_descendants_of_an_unknown_pid_is_empty():
    assert pw.descendants(999999, pw.snapshot()) == []


def test_descendants_is_breadth_first():
    """Ordering matters for a stable display: parents before children."""
    proc = subprocess.Popen(["bash", "-c", "bash -c 'sleep 30' & wait"])
    try:
        time.sleep(0.4)
        snap = pw.snapshot()
        kids = pw.descendants(proc.pid, snap)
        depths = []
        for kid in kids:
            depth, node = 0, kid
            while node.ppid != proc.pid and node.ppid in snap:
                node = snap[node.ppid]
                depth += 1
                if depth > 10:
                    break
            depths.append(depth)
        assert depths == sorted(depths), depths
    finally:
        proc.kill()
        proc.wait()


# --------------------------------------------------------------------------
# classification
# --------------------------------------------------------------------------


def test_classify_picks_the_most_specific_label():
    """`cargo` is more useful than the `compile` its children would give."""
    proc = subprocess.Popen(["bash", "-c", "exec -a cargo sleep 30"])
    try:
        time.sleep(0.3)
        snap = pw.snapshot()
        assert pw.classify(pw.descendants(proc.pid, snap, include_root=True)) == "cargo"
    finally:
        proc.kill()
        proc.wait()


def test_classify_of_an_unrecognised_tree_is_empty():
    proc = subprocess.Popen(["sleep", "30"])
    try:
        time.sleep(0.2)
        snap = pw.snapshot()
        assert pw.classify(pw.descendants(proc.pid, snap, include_root=True)) == ""
    finally:
        proc.kill()
        proc.wait()


def test_classify_is_deterministic_regardless_of_order():
    """The label must not depend on which process happened to be listed first."""
    proc_a = subprocess.Popen(["bash", "-c", "exec -a rustc sleep 30"])
    proc_b = subprocess.Popen(["bash", "-c", "exec -a cargo sleep 30"])
    try:
        time.sleep(0.3)
        snap = pw.snapshot()
        tree = pw.descendants(proc_a.pid, snap, include_root=True) + pw.descendants(
            proc_b.pid, snap, include_root=True
        )
        assert pw.classify(tree) == pw.classify(list(reversed(tree))) == "cargo"
    finally:
        proc_a.kill()
        proc_b.kill()
        proc_a.wait()
        proc_b.wait()


# --------------------------------------------------------------------------
# mapping a working directory back to a package
# --------------------------------------------------------------------------


def test_package_of_reads_the_first_path_component():
    clone = Path("/tmp/clone")
    cases = {
        "/tmp/clone/foo": "foo",
        "/tmp/clone/foo/src/build": "foo",
        "/tmp/clone/foo/pkg/foo/usr": "foo",
        "/tmp/clone/foo-bar/sub": "foo-bar",
        "/tmp/clone": None,
        "/tmp/elsewhere/foo": None,
        None: None,
    }
    for path, want in cases.items():
        assert pw.package_of(path, clone) == want, (path, pw.package_of(path, clone))


def test_package_of_honours_an_explicit_known_set():
    clone = Path("/tmp/clone")
    assert pw.package_of("/tmp/clone/foo/src", clone, known={"foo"}) == "foo"
    assert pw.package_of("/tmp/clone/foo/src", clone, known={"bar"}) is None


def test_package_of_rejects_a_prefix_that_is_not_a_boundary():
    """`/tmp/clone-other/foo` is not inside `/tmp/clone`."""
    assert pw.package_of("/tmp/clone-other/foo", Path("/tmp/clone")) is None


# --------------------------------------------------------------------------
# durations
# --------------------------------------------------------------------------


def test_format_duration():
    cases = {
        0: "0s",
        59: "59s",
        60: "1m00s",
        3599: "59m59s",
        3600: "1h00m",
        3661: "1h01m",
        86400: "24h00m",
    }
    for seconds, want in cases.items():
        assert pw.format_duration(seconds) == want, seconds


def test_elapsed_is_never_negative():
    """A clock adjustment must not produce a negative timer on screen."""
    proc = pw.Proc(pid=1, ppid=0, comm="x", cmdline=(), cwd=None, started=time.time() + 3600)
    assert proc.elapsed == 0.0


def test_a_package_finishes_exactly_once():
    """A duplicate here is a duplicate row in the live feed.

    `finished` served as both the pending queue and the record of what had been
    reported, so clearing it to report made the next poll finish the same
    package again -- and the drain after the launcher exits always polls again.
    """
    import stat
    import tempfile

    directory = Path(tempfile.mkdtemp())
    clone = directory / "clone"
    (clone / "pkgA").mkdir(parents=True)
    # a launcher that spawns a child, as paru does; the root is excluded from
    # the tree by design, so the builder has to be a child
    tool = directory / "launcher"
    tool.write_text(f'#!/bin/bash\n( cd "{clone}/pkgA" && exec -a makepkg sleep 0.4 )\n')
    tool.chmod(tool.stat().st_mode | stat.S_IXUSR)

    watcher = pw.Watcher(clone, known={"pkgA"})
    proc = subprocess.Popen([str(tool)])
    finishes = []
    try:
        while proc.poll() is None:
            watcher.poll(proc.pid)
            finishes += watcher.new_finishes()
            time.sleep(0.03)
        # the drain, then one more poll than strictly needed
        watcher.poll(None)
        finishes += watcher.new_finishes()
        watcher.poll(None)
        finishes += watcher.new_finishes()
    finally:
        proc.kill()
        proc.wait()

    assert finishes == ["pkgA"], finishes


def test_new_finishes_drains():
    watcher = pw.Watcher(Path("/tmp/does-not-matter"))
    watcher.seen.add("pkg")
    watcher.poll(None)
    assert watcher.new_finishes() == ["pkg"]
    assert watcher.new_finishes() == []
    # and a later poll must not resurrect it
    watcher.poll(None)
    assert watcher.new_finishes() == []


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