"""
What a Windows node says about itself when something is wrong.

A 0.16 upgrade in service mode left the previous version's node running, and
every message on the way out pointed elsewhere: `meshbay-node service start`
printed "started" for a node that had not, the app said "the node started but
could not link to your hub account" about one that had stopped answering, the
Node page said "No operator paired" about one whose pairing was intact, and the
daemon itself had logged nowhere at all -- Task Scheduler discards its stderr.
"""

import argparse
import json
import logging
import os
import threading
import time
from http.server import BaseHTTPRequestHandler, HTTPServer

import pytest
from meshbay_node import __version__
from meshbay_node import platform as plat
from meshbay_node.cli import dispatch, lifecycle

# ── the daemon's log file ──────────────────────────────────────────────────


@pytest.fixture
def root_logger():
    """add_log_file changes the root logger: pin it and put it back."""
    root = logging.getLogger()
    handlers, level, disabled = list(root.handlers), root.level, root.disabled
    root.setLevel(logging.INFO)
    root.disabled = False
    yield root
    for h in root.handlers:
        if h not in handlers:
            h.close()
    root.handlers[:] = handlers
    root.setLevel(level)
    root.disabled = disabled


def test_the_daemon_logs_to_a_file_where_there_is_no_console(tmp_path, monkeypatch, root_logger):
    log_path = tmp_path / "state" / "node.log"
    monkeypatch.setattr(plat, "log_file", lambda: log_path)

    dispatch.add_log_file()
    logging.getLogger("meshbay_node.daemon").warning("hub login failed: probe")
    for h in root_logger.handlers:
        h.flush()

    assert log_path.exists(), "the log file's directory must be created, not assumed"
    assert "hub login failed: probe" in log_path.read_text(encoding="utf-8")


def test_no_log_file_where_journald_has_the_output(monkeypatch, root_logger):
    monkeypatch.setattr(plat, "log_file", lambda: None)
    before = list(root_logger.handlers)
    dispatch.add_log_file()
    assert root_logger.handlers == before


def test_the_log_file_is_under_localappdata_on_windows(monkeypatch, tmp_path):
    monkeypatch.undo()   # the conftest keeps log_file() away from the real profile
    monkeypatch.setattr(plat.sys, "platform", "win32")
    monkeypatch.setenv("LOCALAPPDATA", str(tmp_path))
    assert plat.log_file() == tmp_path / "meshbay" / "state" / "node.log"
    monkeypatch.setattr(plat.sys, "platform", "linux")
    assert plat.log_file() is None


def test_only_the_daemon_run_opens_the_log_file(monkeypatch):
    """CLI verbs print reports; appending them to the daemon's log would bury
    the daemon's own lines."""
    calls = []
    monkeypatch.setattr(dispatch, "add_log_file", lambda: calls.append(1))
    monkeypatch.setattr(plat, "load_node_env", lambda _d: 0)
    for argv, expected in ((["meshbay-node", "status"], 0), (["meshbay-node"], 1)):
        calls.clear()
        monkeypatch.setattr("sys.argv", argv)
        dispatch.start()
        assert len(calls) == expected, argv


# ── `service start` says what happened ──────────────────────────────────────


class _Status(BaseHTTPRequestHandler):
    body = {}

    def do_GET(self):  # noqa: N802
        data = json.dumps(self.body).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def log_message(self, *_a):
        pass


@pytest.fixture
def fake_daemon(tmp_path):
    server = HTTPServer(("127.0.0.1", 0), _Status)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    data_dir = tmp_path / "data"
    data_dir.mkdir()
    cfg = argparse.Namespace(
        data_dir=data_dir,
        node=argparse.Namespace(ui_port=server.server_address[1]))
    yield cfg
    server.shutdown()


def test_a_token_from_the_stopped_instance_is_not_taken_for_the_new_one(fake_daemon):
    _Status.body = {"version": "0.16.0", "status": "running"}
    token = fake_daemon.data_dir / "ui-token"
    token.write_text("t", encoding="utf-8")
    old = time.time() - 60
    os.utime(token, (old, old))

    assert lifecycle._await_daemon(fake_daemon, since=time.time() - 1, timeout=1.5) is None


def test_a_daemon_that_started_is_seen(fake_daemon):
    _Status.body = {"version": "0.16.0", "status": "waiting_for_account"}
    since = time.time() - 1
    (fake_daemon.data_dir / "ui-token").write_text("t", encoding="utf-8")

    got = lifecycle._await_daemon(fake_daemon, since=since, timeout=5)
    assert got == _Status.body


def _service_start(monkeypatch, awaited, capsys):
    monkeypatch.setattr(plat, "service_status", lambda: {"installed": True, "state": "Ready"})
    monkeypatch.setattr(plat, "service_run", lambda: None)
    monkeypatch.setattr(plat, "service_end", lambda: None)
    monkeypatch.setattr(plat, "log_file", lambda: "C:/x/meshbay/state/node.log")
    monkeypatch.setattr(lifecycle, "load_config", lambda _p: object())
    monkeypatch.setattr(lifecycle, "_running_status", lambda _cfg: None)
    monkeypatch.setattr(lifecycle, "_await_daemon", lambda _cfg, _since, **_kw: awaited)
    monkeypatch.setattr(lifecycle.sys, "platform", "win32")
    lifecycle.service(argparse.Namespace(subcommand="start", config=None))
    return capsys.readouterr().out


def test_service_start_does_not_report_a_node_that_never_answered(monkeypatch, capsys):
    with pytest.raises(SystemExit) as exc:
        _service_start(monkeypatch, None, capsys)
    assert exc.value.code == 1
    out = capsys.readouterr().out
    assert not out.startswith("started"), "it must not claim the node started"
    assert "no node answered" in out
    assert "node.log" in out, "and must say where to look"


def test_service_start_names_the_version_that_answered(monkeypatch, capsys):
    out = _service_start(monkeypatch, {"version": __version__, "status": "running"}, capsys)
    assert f"node {__version__}" in out and "warning" not in out


def test_service_start_warns_when_the_service_runs_an_older_node(monkeypatch, capsys):
    out = _service_start(monkeypatch, {"version": "0.15.0", "status": "waiting_for_account"},
                         capsys)
    assert "0.15.0" in out and "warning" in out and "not replaced" in out


# ── a graceful stop, through the control API ───────────────────────────────


def test_the_control_api_asks_the_daemon_to_stop():
    from fastapi.testclient import TestClient
    from meshbay_node.ui import create_ui_app

    asked = []
    app = create_ui_app({"ui_token": "tok", "request_shutdown": lambda: asked.append(1)})
    client = TestClient(app)
    assert client.post("/api/shutdown").status_code in (401, 403), "behind the token"
    assert asked == []
    r = client.post("/api/shutdown?t=tok")
    assert r.status_code == 200 and r.json() == {"stopping": True}
    assert asked == [1]


def test_a_stop_request_ends_the_wait_for_the_hub():
    """A node waiting for its account is the one people stop and restart."""
    import asyncio

    from meshbay_node.daemon import NodeDaemon

    async def go():
        daemon = NodeDaemon.__new__(NodeDaemon)
        daemon._stop_event = asyncio.Event()
        asyncio.get_running_loop().call_later(0.1, daemon._stop_event.set)
        t0 = time.monotonic()
        stopped = await daemon._pause(30)
        return stopped, time.monotonic() - t0

    stopped, took = asyncio.run(go())
    assert stopped and took < 5


class _RealNode:
    """A real daemon process in a throwaway profile, pointed at a hub that is
    not there (it waits for it, control API up)."""

    def __init__(self, tmp_path):
        import socket

        with socket.socket() as s:
            s.bind(("127.0.0.1", 0))
            self.port = s.getsockname()[1]
        self.tmp = tmp_path
        home = tmp_path / "home"
        self.conf = tmp_path / "node.toml"
        unlock = tmp_path / "unlock.key"
        unlock.write_text("x" * 43, encoding="utf-8")
        self.data = tmp_path / "data"
        self.conf.write_text(
            f'data_dir = "{self.data.as_posix()}"\n'
            '[hub]\nurl = "http://127.0.0.1:1"\nusername = "probe"\n'
            f'[node]\nquic_enabled = false\nui_port = {self.port}\n'
            f'[keystore]\npath = "{(tmp_path / "keystore.enc").as_posix()}"\n'
            f'unlock_file = "{unlock.as_posix()}"\n', encoding="utf-8")
        self.env = {**os.environ, "HOME": str(home), "LOCALAPPDATA": str(home),
                    "USERPROFILE": str(home)}
        self.procs = []

    def spawn(self, name: str):
        import subprocess
        import sys

        out = self.tmp / f"{name}.txt"
        with open(out, "w", encoding="utf-8") as f:
            proc = subprocess.Popen(
                [sys.executable, "-c", "from meshbay_node.daemon import main; main()",
                 "--config", str(self.conf)],
                env=self.env, stderr=f, stdout=f)
        proc.out = out
        self.procs.append(proc)
        return proc

    def wait_up(self, proc) -> None:
        deadline = time.monotonic() + 40
        while not (self.data / "ui-token").exists() and time.monotonic() < deadline:
            assert proc.poll() is None, proc.out.read_text(encoding="utf-8")
            time.sleep(0.2)
        time.sleep(1)

    def kill_all(self) -> None:
        for p in self.procs:
            if p.poll() is None:
                p.kill()


def test_a_real_daemon_shuts_down_properly_when_asked(tmp_path):
    """The whole path, on a real process: before this, nine stops of a Windows
    node in a row logged not one shutdown -- every one was a TerminateProcess."""
    node = _RealNode(tmp_path)
    proc = node.spawn("first")
    try:
        node.wait_up(proc)
        assert plat.request_graceful_stop(node.data, node.port, timeout=15)
        assert proc.wait(timeout=10) is not None
        err = proc.out.read_text(encoding="utf-8")
        assert "Node stopped" in err, err[-2000:]
    finally:
        node.kill_all()


def test_a_second_node_leaves_the_running_one_alone_and_says_why(tmp_path):
    """Found on a real install: the sign-in launcher run while a node was up
    started a second one, which wrote its token over the first's, failed to bind
    inside uvicorn's task and exited with nothing in the log. The node that kept
    running then refused every stop, status and restart -- the token file named
    a process that no longer existed."""
    node = _RealNode(tmp_path)
    first = node.spawn("first")
    try:
        node.wait_up(first)
        token = (node.data / "ui-token").read_text(encoding="utf-8")

        second = node.spawn("second")
        assert second.wait(timeout=40) == 2
        said = second.out.read_text(encoding="utf-8")
        assert f"127.0.0.1:{node.port} is taken" in said, said[-2000:]
        assert "already running" in said

        assert (node.data / "ui-token").read_text(encoding="utf-8") == token
        assert first.poll() is None
        assert plat.request_graceful_stop(node.data, node.port, timeout=15), (
            "the running node can no longer be asked to stop")
    finally:
        node.kill_all()


def test_the_control_port_is_exclusive():
    """On Windows SO_REUSEADDR would let a second socket share a port in use."""
    from meshbay_node.daemon import ControlPortTaken, bind_control_port

    first = bind_control_port(0)
    try:
        port = first.getsockname()[1]
        with pytest.raises(ControlPortTaken, match="already running"):
            bind_control_port(port)
    finally:
        first.close()


# ── "No operator paired" only when the roster says so ──────────────────────


async def test_a_node_that_has_not_read_its_roster_does_not_claim_there_is_no_operator():
    from meshbay_node import ops
    result = await ops.list_groups({"config": None, "groups_ctx": {}})
    assert result["operator_paired"] is None


async def test_a_node_with_a_roster_reports_the_operator(tmp_path):
    from meshbay_node import ops
    from meshbay_node.roster import Roster

    roster = Roster(db_path=tmp_path / "roster.db")
    await roster.open()
    try:
        result = await ops.list_groups({"config": None, "groups_ctx": {}, "roster": roster})
    finally:
        await roster.close()
    assert result["operator_paired"] is False