"""The host agent: stdlib-only, read-only by construction, and wrong on no box.

Three things are asserted here:
  * parsing of /proc and /sys files exactly as Linux writes them (fake roots, so the
    weird cases are testable on any machine);
  * the collector produces a complete, non-crashing snapshot against the *real* /proc of
    whatever runs the tests, because that is the code that ships;
  * the HTTP surface answers GETs, needs a token when it is exposed, and has no write path.
"""
from __future__ import annotations

import importlib.util
import json
import re
import sys
import threading
import time
import urllib.error
import urllib.request
from http.server import ThreadingHTTPServer
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[1]
AGENT_PATH = ROOT / "agent" / "host_agent.py"

spec = importlib.util.spec_from_file_location("host_agent", AGENT_PATH)
assert spec and spec.loader
host_agent = importlib.util.module_from_spec(spec)
sys.modules["host_agent"] = host_agent
spec.loader.exec_module(host_agent)

linux_only = pytest.mark.skipif(not sys.platform.startswith("linux"),
                                reason="reads /proc and /sys")


# ---- source-level helpers ------------------------------------------------- #

@pytest.mark.parametrize("lspci,expected", [
    ("Advanced Micro Devices, Inc. [AMD/ATI] Navi 48 [Radeon AI PRO R9700]",
     "Radeon AI PRO R9700"),
    ("Advanced Micro Devices, Inc. [AMD/ATI] Strix Halo [Radeon Graphics / Radeon 8050S "
     "Graphics / Radeon 8060S Graphics]", "Radeon 8060S Graphics"),
    ("Advanced Micro Devices, Inc. [AMD/ATI] Phoenix1", "Phoenix1"),
    ("00:02.0 VGA compatible controller: Intel Corporation UHD Graphics 620",
     "UHD Graphics 620"),
    ("NVIDIA GB200", "NVIDIA GB200"),
    ("Vendor 1af1 Device 0001", "Vendor 1af1 Device 0001"),
    (None, None),
    ("", None),
])
def test_gpu_model_is_shortened_for_legends_without_losing_identity(lspci, expected) -> None:
    """`ATI` on a legend is worse than no name: it is true of every AMD GPU on the box."""
    assert host_agent.short_gpu_model(lspci) == expected


def test_gpu_model_is_capped_so_a_legend_cannot_blow_up() -> None:
    long_name = "X" * 200
    assert len(host_agent.short_gpu_model(long_name)) <= 41  # 40 + ellipsis


# ---- /proc and /sys parsing (fake roots) ---------------------------------- #

def write_file(root: Path, relative: str, text: str) -> None:
    path = root / relative
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(text)


def test_read_number_takes_the_first_field(tmp_path: Path) -> None:
    """/proc/uptime is `210036.18 9613750.86`; float() of the whole line is None."""
    write_file(tmp_path, "proc/uptime", "210036.18 9613750.86\n")
    collector = host_agent.Collector(str(tmp_path))
    assert collector.read_number("proc/uptime") == 210036.18


def test_read_number_handles_missing_and_junk_files(tmp_path: Path) -> None:
    collector = host_agent.Collector(str(tmp_path))
    assert collector.read_number("proc/nope") is None
    write_file(tmp_path, "proc/junk", "\n")
    assert collector.read_number("proc/junk") is None


def test_core_counts_dedupe_socket_and_core(tmp_path: Path) -> None:
    """One /proc/cpuinfo block per *thread*: counting 'core id' lines said '1 core'."""
    blocks = []
    for thread in range(8):  # 2 sockets x 2 cores x 2 threads
        socket, core = thread % 2, (thread // 2) % 2
        blocks.append(f"processor\t: {thread}\nphysical id\t: {socket}\ncore id\t\t: {core}\n")
    write_file(tmp_path, "proc/cpuinfo", "\n".join(blocks))
    collector = host_agent.Collector(str(tmp_path))
    assert collector.core_counts() == (8, 4)


def test_core_counts_falls_back_to_smt_width_on_sysfs(tmp_path: Path) -> None:
    """Some kernels/vms give no physical id/core id; cpu0's sibling list still says SMT=2."""
    write_file(tmp_path, "proc/cpuinfo", "processor\t: 0\nprocessor\t: 1\n")
    write_file(tmp_path, "sys/devices/system/cpu/cpu0/topology/thread_siblings_list", "0,1\n")
    collector = host_agent.Collector(str(tmp_path))
    logical, physical = collector.core_counts()
    assert (logical, physical) == (2, 1)


def test_core_counts_survive_a_kernel_with_no_hints(tmp_path: Path) -> None:
    collector = host_agent.Collector(str(tmp_path))
    assert collector.core_counts()[1] is None, "unknown must be None, never 1"


def test_cpu_modes_total_excludes_guest_ticks(tmp_path: Path) -> None:
    """/proc/stat's idle/iowait are already inside 'total'; guest is inside user on Linux."""
    write_file(tmp_path, "proc/stat",
               "cpu  100 10 20 500 5 3 2 1 0 0\nintr 1\nctxt 2\nprocesses 3\n")
    collector = host_agent.Collector(str(tmp_path))
    sample = collector.cpu()["modes"]
    assert sample["user"] == 100 and sample["idle"] == 500
    assert sample["total"] == pytest.approx(100 + 10 + 20 + 500 + 5 + 3 + 2 + 1)


# ---- the real collector --------------------------------------------------- #

@linux_only
def test_snapshot_against_the_live_system_is_complete() -> None:
    """Real /proc and /sys, but only of the box running pytest — no network, so this is not
    `live` (see tests/test_live_fleet.py): it belongs in the default run."""
    collector = host_agent.Collector("/")
    snap = collector.snapshot()
    for key in ("schema", "ts", "host", "cpu", "memory", "filesystems", "disk_io",
                "network", "gpus", "temperatures", "processes", "errors"):
        assert key in snap, f"{key} missing from a real snapshot"
    assert snap["host"]["hostname"]
    assert snap["cpu"]["modes"]["total"] > 0
    assert len(snap["cpu"]["cores"]) >= 1
    assert snap["memory"]["total_bytes"] > 0
    assert snap["filesystems"], "the box it runs on has at least one mounted filesystem"
    assert snap["errors"] == [], f"degraded sections must be empty on a healthy box: {snap['errors']}"


def test_captured_fixture_has_the_shape_the_dashboard_depends_on(agent_snapshot) -> None:
    """Ground truth from driveripper: if the agent's schema drifts, this fails first."""
    assert agent_snapshot["schema"] == host_agent.SCHEMA_VERSION
    host = agent_snapshot["host"]
    assert host["hostname"] == "driveripper"
    assert (host["cpu_cores_physical"], host["cpu_cores_logical"]) == (24, 48)
    assert host["uptime_seconds"] > 0
    gpu = agent_snapshot["gpus"][0]
    assert gpu["model"] == "Radeon AI PRO R9700"
    assert "model_full" in gpu and "Navi 48" in gpu["model_full"]
    assert gpu["vram_total_bytes"] > 0
    # the dashboard renders n/a for anything a driver omits — it must never invent a 0
    assert gpu["busy_pct"] is None or 0.0 <= gpu["busy_pct"] <= 100.0


@linux_only
def test_uptime_and_boot_ts_are_reported() -> None:
    """Both are needed: uptime for the table, boot_ts to detect a counter-reset reboot."""
    info = host_agent.Collector("/").host_info()
    assert info["uptime_seconds"] and info["uptime_seconds"] > 0
    assert info["boot_ts"] and info["boot_ts"] <= time.time()
    assert abs((time.time() - info["boot_ts"]) - info["uptime_seconds"]) < 60
    assert info["cpu_cores_logical"] and info["cpu_cores_physical"]


@linux_only
def test_snapshot_is_repeatable_and_does_not_deadlock() -> None:
    """snapshot() reads processes() internally; a non-reentrant lock used to hang it."""
    collector = host_agent.Collector("/")
    first = collector.snapshot()
    second = collector.snapshot(top=0)
    assert first["processes"]["total"] > 0
    assert second["processes"]["top_cpu"] == [], "top=0 must skip the expensive scan"


@linux_only
def test_disk_io_comes_from_sysfs_not_proc_diskstats() -> None:
    """/proc/diskstats field order has moved between kernels; /sys/block/<dev>/stat has not."""
    devices = host_agent.Collector("/").disk_io()["devices"]
    assert devices
    for device in devices:
        assert device["reads_total"] is None or device["reads_total"] >= 0
        assert device["size_bytes"] is None or device["size_bytes"] > 0


# ---- HTTP surface --------------------------------------------------------- #

@pytest.fixture
def agent_server():
    """Run the real handler on an ephemeral loopback port; yield (get, port)."""
    servers: list[ThreadingHTTPServer] = []

    def start(token: str | None = None) -> tuple[str, int]:
        host_agent.Handler.collector = host_agent.Collector("/")
        host_agent.Handler.token = token
        httpd = ThreadingHTTPServer(("127.0.0.1", 0), host_agent.Handler)
        servers.append(httpd)
        threading.Thread(target=httpd.serve_forever, daemon=True).start()
        return f"http://127.0.0.1:{httpd.server_address[1]}", httpd.server_address[1]

    yield start
    for httpd in servers:
        httpd.shutdown()
        httpd.server_close()


def _get(url: str, headers: dict[str, str] | None = None) -> tuple[int, dict]:
    request = urllib.request.Request(url, headers=headers or {})
    try:
        with urllib.request.urlopen(request, timeout=5) as response:
            return response.status, json.loads(response.read())
    except urllib.error.HTTPError as error:
        body = error.read()
        try:
            return error.code, json.loads(body)
        except json.JSONDecodeError:  # pragma: no cover - handler always sends JSON
            return error.code, {"raw": body.decode(errors="replace")}


def test_health_info_schema_snapshot_all_answer(agent_server) -> None:
    base, _port = agent_server()
    for path in ("/health", "/info", "/schema", "/snapshot"):
        status, body = _get(base + path)
        assert status == 200, path
        assert body["ok"] is True, path


def test_snapshot_query_param_controls_the_process_scan(agent_server) -> None:
    base, _port = agent_server()
    _status, body = _get(base + "/snapshot?top=0")
    assert body["processes"]["top_cpu"] == []
    _status, full = _get(base + "/snapshot?top=3")
    assert len(full["processes"]["top_cpu"]) <= 3


def test_unknown_path_is_404_not_a_snapshot(agent_server) -> None:
    base, _port = agent_server()
    status, body = _get(base + "/etc/shadow")
    assert status == 404
    assert body["ok"] is False


def test_token_is_required_when_configured_but_health_stays_open(agent_server) -> None:
    base, _port = agent_server(token="s3cret")
    assert _get(base + "/snapshot")[0] == 401
    assert _get(base + "/snapshot", {"Authorization": "Bearer wrong"})[0] == 401
    assert _get(base + "/snapshot", {"Authorization": "Bearer s3cret"})[0] == 200
    # /health stays open so a monitoring probe needs no secret
    assert _get(base + "/health")[0] == 200


def test_agent_has_no_write_path(agent_server) -> None:
    """No do_POST/do_PUT exists: read-only is structural, not a policy toggle."""
    base, _port = agent_server()
    request = urllib.request.Request(base + "/snapshot", data=b"{}", method="POST")
    with pytest.raises(urllib.error.HTTPError) as excinfo:
        urllib.request.urlopen(request, timeout=5)
    assert excinfo.value.code in (405, 501)


def test_refuses_to_bind_off_loopback_without_a_token() -> None:
    """0.0.0.0 with no token would hand host inventory to the whole LAN."""
    assert host_agent.main(["--host", "0.0.0.0", "--port", "0"]) == 2


def test_once_mode_prints_json_and_exits(capsys) -> None:
    assert host_agent.main(["--once"]) == 0
    printed = capsys.readouterr().out
    assert json.loads(printed)["ok"] is True


def test_agent_is_dependency_free() -> None:
    """The agent must run on a stock distro python: stdlib imports, and few of them.

    Two checks, because they fail for different reasons: anything outside the stdlib means a
    `pip install` on every monitored host (deployment pain), while a growing footprint means
    the agent can break on an older interpreter than the dashboard's own Python.
    """
    text = AGENT_PATH.read_text()
    imports = re.findall(r"^(?:import|from)\s+([a-zA-Z0-9_.]+)", text, re.M)
    # indented (function-local) imports count too — subprocess is loaded lazily on purpose
    imports += re.findall(r"^\s+(?:import|from)\s+([a-zA-Z0-9_.]+)", text, re.M)
    top = {name.split(".")[0] for name in imports}
    third_party = top - set(sys.stdlib_module_names) - {"host_agent"}
    assert not third_party, f"agent grew a non-stdlib import: {third_party}"
    assert top <= {"__future__", "argparse", "collections", "dataclasses", "http", "json",
                   "os", "pathlib", "re", "socket", "subprocess", "threading", "time",
                   "typing"}, f"agent import footprint grew: {sorted(top)}"


# ---- how the token is supplied ------------------------------------------- #

def test_token_file_wins_over_the_command_line(tmp_path) -> None:
    path = tmp_path / "token"
    path.write_text("from-file\n", encoding="utf-8")
    token, error = host_agent.resolve_token("from-flag", str(path))
    assert error is None
    assert token == "from-file", "argv is world-readable in ps; the file has to win"


def test_a_missing_token_file_is_fatal_and_says_which_one(tmp_path) -> None:
    missing = tmp_path / "nope"
    token, error = host_agent.resolve_token("", str(missing))
    assert token is None
    assert error and str(missing) in error, "a silent fallback to no-token would expose the host"


def test_an_empty_token_file_is_fatal(tmp_path) -> None:
    path = tmp_path / "token"
    path.write_text("   \n", encoding="utf-8")
    token, error = host_agent.resolve_token("ignored", str(path))
    assert token is None and error is not None


def test_no_token_at_all_is_allowed_on_loopback(tmp_path) -> None:
    token, error = host_agent.resolve_token("", "")
    assert (token, error) == (None, None)


def test_main_refuses_to_bind_off_loopback_without_a_token(tmp_path) -> None:
    code = host_agent.main(["--host", "0.0.0.0", "--token-file", str(tmp_path / "missing")])
    assert code == 2, "the agent must not start exposed and unauthenticated"