from __future__ import annotations

import importlib.util
import json
import socket
import threading
from pathlib import Path

import pytest

MODULE = Path(__file__).resolve().parents[1] / "lib" / "tailnet_edge.py"
spec = importlib.util.spec_from_file_location("subrouter_tailnet_edge", MODULE)
edge = importlib.util.module_from_spec(spec)
assert spec.loader is not None
spec.loader.exec_module(edge)


def status_fixture(*, debian1_online: bool = True):
    return {
        "BackendState": "Running",
        "Self": {"TailscaleIPs": ["100.126.128.50", "fd7a:115c:a1e0::1"]},
        "Peer": {
            "a": {"HostName": "debian1", "Online": debian1_online, "TailscaleIPs": ["100.106.253.50"]},
            "b": {"HostName": "debian2", "Online": True, "TailscaleIPs": ["100.79.69.43"]},
        },
    }


def test_resolve_tailnet_binds_self_and_only_explicit_peer():
    listen, allowed = edge.resolve_tailnet(status_fixture(), ["debian1"])
    assert listen == "100.126.128.50"
    assert allowed == {"100.106.253.50"}


@pytest.mark.parametrize(
    "mutate, message",
    [
        (lambda s: s.update(BackendState="Stopped"), "not running"),
        (lambda s: s["Peer"].pop("a"), "not unique/present"),
        (lambda s: s["Peer"]["a"].update(Online=False), "offline"),
        (lambda s: s["Self"].update(TailscaleIPs=["192.0.2.1"]), "exactly one tailnet"),
    ],
)
def test_resolve_tailnet_fails_closed(mutate, message):
    status = status_fixture()
    mutate(status)
    with pytest.raises(ValueError, match=message):
        edge.resolve_tailnet(status, ["debian1"])


def test_load_config_is_exact_and_loopback_only(tmp_path):
    cfg = tmp_path / "edge.json"
    cfg.write_text(json.dumps({
        "schema": edge.SCHEMA,
        "listen_port": 31416,
        "upstream_host": "127.0.0.1",
        "upstream_port": 31415,
        "allowed_hosts": ["debian1"],
    }))
    loaded = edge.load_config(cfg)
    assert loaded["allowed_hosts"] == ["debian1"]
    body = json.loads(cfg.read_text())
    body["upstream_host"] = "0.0.0.0"
    cfg.write_text(json.dumps(body))
    with pytest.raises(ValueError, match="upstream must be loopback"):
        edge.load_config(cfg)


def test_unapproved_peer_gets_403_without_upstream_connection():
    client, edge_side = socket.socketpair()
    edge.handle_client(edge_side, "100.79.69.43", {"100.106.253.50"}, ("127.0.0.1", 9))
    response = client.recv(4096)
    client.close()
    assert response.startswith(b"HTTP/1.1 403 Forbidden\r\n")
    assert b"Authorization" not in response


def test_approved_peer_is_raw_proxied_without_header_or_url_logging():
    upstream = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    upstream.bind(("127.0.0.1", 0))
    upstream.listen(1)
    port = upstream.getsockname()[1]
    seen = []

    def server():
        conn, _ = upstream.accept()
        try:
            seen.append(conn.recv(4096))
            conn.sendall(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nOK")
        finally:
            conn.close(); upstream.close()

    threading.Thread(target=server, daemon=True).start()
    client, edge_side = socket.socketpair()
    worker = threading.Thread(
        target=edge.handle_client,
        args=(edge_side, "100.106.253.50", {"100.106.253.50"}, ("127.0.0.1", port)),
        daemon=True,
    )
    worker.start()
    request = b"GET /r/opaque/_subrouter/status HTTP/1.1\r\nAuthorization: Bearer fixture-secret\r\nHost: edge\r\n\r\n"
    client.sendall(request)
    response = client.recv(4096)
    client.close(); worker.join(timeout=2)
    assert response.endswith(b"OK")
    assert seen == [request]