"""OTP 26 integration tests via Docker.

Reproduces the production failure mode we hit on portal@unsandbox.com:
erldistpy 0.1.7 handshake "succeeds" against OTP 26 but the peer
silently closes the link on the first REG_SEND, surfacing as
``ChannelError: peer closed after 0/4 bytes`` on Node.call().

The pure-Python tests and the existing live tests all run against the
system Erlang, which on most dev boxes is OTP 24 (Ubuntu 22.04 default).
OTP 24 didn't enforce DFLAG_MANDATORY_25_DIGEST or the OTP-25+
mandatory flag set, so flag-negotiation bugs pass silently. These
tests force the modern protocol by running a peer node out of the
official ``erlang:26`` Docker image.

Tests are skipped automatically if Docker isn't available — they're
intentionally NOT part of the smoke that PRs gate on, because they
require docker pull + ~7s of node boot per module.
"""

from __future__ import annotations

import shutil
import socket
import subprocess
import time

import pytest

from pathlib import Path

from erldistpy.etf import Atom
from erldistpy.node import Node
from erldistpy.tls import make_dist_tls_context


SNAME_OTP26 = "erldistpy_otp26"
COOKIE_OTP26 = "ERLDISTPY_OTP26_COOKIE"

# Same gen_target shape as tests/test_node.py — implements the
# gen_server-style {'$gen_call', {From, Ref}, Request} -> From ! {Ref, Reply}
# protocol by hand. Two operations:
#   {ping, X}        -> {pong, X}
#   {add, A, B}      -> {ok, A + B}
# Anything else surfaces as {error, {bad_request, _}}.
#
# We also register a heartbeat after 200ms so tests can poll a known
# pid presence rather than depending on a /tmp marker file (which would
# live inside the container, invisible to the host).
ERL_GEN_BOOT = (
    "Handle = fun "
    "  ({ping, X}) -> {pong, X}; "
    "  ({add, A, B}) -> {ok, A + B}; "
    "  (Other) -> {error, {bad_request, Other}} "
    "end, "
    "Loop = fun(F) -> "
    "  receive "
    "    {'$gen_call', {From, Ref}, Request} -> "
    "      From ! {Ref, Handle(Request)}, F(F); "
    "    Other -> "
    "      io:format(\"unexpected: ~p~n\", [Other]), F(F) "
    "  end "
    "end, "
    "Pid = spawn(fun() -> Loop(Loop) end), "
    "register(gen_target, Pid), "
    "timer:sleep(infinity)."
)


def _docker_available() -> bool:
    return shutil.which("docker") is not None


def _epmd_node_present(sname: str) -> bool:
    """Returns True if ``sname`` is registered in the host's epmd."""
    try:
        result = subprocess.run(
            ["epmd", "-names"],
            capture_output=True,
            text=True,
            timeout=2,
        )
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return False
    return f"name {sname} at port" in result.stdout


def _socket_open(host: str, port: int) -> bool:
    try:
        with socket.create_connection((host, port), timeout=0.5):
            return True
    except OSError:
        return False


@pytest.fixture(scope="module")
def otp26_gen_peer():
    if not _docker_available():
        pytest.skip("docker not installed")
    if not _socket_open("127.0.0.1", 4369):
        pytest.skip("EPMD not running on 127.0.0.1:4369 (host)")

    container_name = f"erldistpy-otp26-{int(time.time())}"

    # --network host shares the host's network namespace (Linux only),
    # so the OTP 26 node registers with the host's epmd and listens on
    # a host-routable interface. epmd lookups from erldistpy on the
    # host find the docker node like any other local Erlang node.
    proc = subprocess.Popen(
        [
            "docker",
            "run",
            "--rm",
            "--name",
            container_name,
            "--network",
            "host",
            "erlang:25",
            "erl",
            "-sname",
            SNAME_OTP26,
            "-setcookie",
            COOKIE_OTP26,
            "-noshell",
            "-eval",
            ERL_GEN_BOOT,
        ],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )

    deadline = time.monotonic() + 15.0
    ready = False
    while time.monotonic() < deadline:
        time.sleep(0.2)
        if _epmd_node_present(SNAME_OTP26):
            # Allow another 200ms for `register(gen_target, ...)` to land.
            time.sleep(0.2)
            ready = True
            break
    if not ready:
        subprocess.run(["docker", "rm", "-f", container_name], capture_output=True)
        proc.terminate()
        proc.wait(timeout=2)
        pytest.skip(
            f"OTP 26 docker node {SNAME_OTP26} did not register with host epmd"
        )

    yield SNAME_OTP26, COOKIE_OTP26

    subprocess.run(["docker", "rm", "-f", container_name], capture_output=True)
    proc.terminate()
    try:
        proc.wait(timeout=2)
    except subprocess.TimeoutExpired:
        proc.kill()


def test_call_ping_otp26(otp26_gen_peer):
    """The minimal repro of the production failure.

    erldistpy 0.1.6 silently dropped this call against OTP 26 portal.
    0.1.7 added DFLAG_MANDATORY_25_DIGEST; whether THAT was enough is
    exactly what this test answers.
    """
    sname, cookie = otp26_gen_peer
    with Node(our_name="erldistpy_test_otp26@localhost", peer_name=sname, cookie=cookie) as n:
        reply = n.call("gen_target", (Atom("ping"), 42), timeout=5.0)
        assert reply == (Atom("pong"), 42)


def test_call_add_otp26(otp26_gen_peer):
    sname, cookie = otp26_gen_peer
    with Node(our_name="erldistpy_test_otp26_add@localhost", peer_name=sname, cookie=cookie) as n:
        reply = n.call("gen_target", (Atom("add"), 5, 7), timeout=5.0)
        assert reply == (Atom("ok"), 12)


def test_call_sequence_otp26(otp26_gen_peer):
    """Multiple calls on the same dist link survive net_tick window."""
    sname, cookie = otp26_gen_peer
    with Node(our_name="erldistpy_test_otp26_seq@localhost", peer_name=sname, cookie=cookie) as n:
        assert n.call("gen_target", (Atom("ping"), Atom("a")), timeout=5.0) == (Atom("pong"), Atom("a"))
        assert n.call("gen_target", (Atom("ping"), Atom("b")), timeout=5.0) == (Atom("pong"), Atom("b"))
        assert n.call("gen_target", (Atom("add"), 1, 2), timeout=5.0) == (Atom("ok"), 3)


# ---------------------------------------------------------------------------
# OTP 26 + TLS dist — the production combo (mTLS over Erlang dist)
# ---------------------------------------------------------------------------


TLS_SNAME = "erldistpy_otp26_tls"
TLS_COOKIE = "ERLDISTPY_OTP26_TLS_COOKIE"


def _gen_certs(workdir: Path) -> dict[str, str]:
    """Generate a CA + leaf cert with a SAN matching localhost. Same shape
    as tests/test_tls.py — duplicated locally to keep this file self-
    contained (the docker container needs to read from a known path)."""
    import os
    hostname = socket.gethostname()
    ca_cnf = workdir / "ca.cnf"
    ca_cnf.write_text(
        "[req]\n"
        "distinguished_name = dn\n"
        "x509_extensions = v3_ca\n"
        "prompt = no\n"
        "[dn]\n"
        "CN = erldistpy-otp26-ca\n"
        "[v3_ca]\n"
        "basicConstraints = critical,CA:TRUE\n"
        "keyUsage = critical,keyCertSign,cRLSign\n"
        "subjectKeyIdentifier = hash\n"
    )
    subprocess.run(
        ["openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes",
         "-keyout", str(workdir / "ca.key"),
         "-out", str(workdir / "ca.pem"),
         "-days", "1",
         "-config", str(ca_cnf), "-extensions", "v3_ca"],
        check=True, capture_output=True,
    )
    leaf_cnf = workdir / "leaf.cnf"
    leaf_cnf.write_text(
        "[v3]\n"
        "basicConstraints = CA:FALSE\n"
        "keyUsage = digitalSignature,keyEncipherment\n"
        "extendedKeyUsage = serverAuth,clientAuth\n"
        "subjectAltName = @alt\n"
        "[alt]\n"
        f"DNS.1 = {hostname}\n"
        "DNS.2 = localhost\n"
    )
    for name in ("server", "client"):
        subprocess.run(
            ["openssl", "req", "-newkey", "rsa:2048", "-nodes",
             "-keyout", str(workdir / f"{name}.key"),
             "-out", str(workdir / f"{name}.csr"),
             "-subj", f"/CN={hostname}"],
            check=True, capture_output=True,
        )
        subprocess.run(
            ["openssl", "x509", "-req",
             "-in", str(workdir / f"{name}.csr"),
             "-CA", str(workdir / "ca.pem"),
             "-CAkey", str(workdir / "ca.key"),
             "-CAcreateserial",
             "-out", str(workdir / f"{name}.pem"),
             "-days", "1",
             "-extfile", str(leaf_cnf), "-extensions", "v3"],
            check=True, capture_output=True,
        )
        os.chmod(workdir / f"{name}.key", 0o644)  # readable by container's erl uid
    return {
        "ca": str(workdir / "ca.pem"),
        "server_cert": str(workdir / "server.pem"),
        "server_key": str(workdir / "server.key"),
        "client_cert": str(workdir / "client.pem"),
        "client_key": str(workdir / "client.key"),
    }


def _write_ssl_config(workdir: Path, certs_in_container: dict[str, str]) -> str:
    cfg = workdir / "ssl_dist.config"
    body = (
        "[{server, "
        f'[{{certfile, "{certs_in_container["server_cert"]}"}}, '
        f'{{keyfile, "{certs_in_container["server_key"]}"}}, '
        f'{{cacertfile, "{certs_in_container["ca"]}"}}, '
        "{verify, verify_peer}, "
        "{fail_if_no_peer_cert, true}]}, "
        "{client, "
        f'[{{certfile, "{certs_in_container["server_cert"]}"}}, '
        f'{{keyfile, "{certs_in_container["server_key"]}"}}, '
        f'{{cacertfile, "{certs_in_container["ca"]}"}}, '
        "{verify, verify_peer}]}]."
    )
    cfg.write_text(body)
    return str(cfg)


@pytest.fixture(scope="module")
def otp26_tls_peer(tmp_path_factory):
    """OTP 26 + TLS dist peer in Docker — the production combo.

    This is the variant that reproduces (or should reproduce) the
    portal@unsandbox.com failure: handshake "succeeds" then the peer
    silently closes the link on the first REG_SEND.
    """
    if not _docker_available():
        pytest.skip("docker not installed")
    if not shutil.which("openssl"):
        pytest.skip("openssl not installed")
    if not _socket_open("127.0.0.1", 4369):
        pytest.skip("EPMD not running on host")

    # Snap-confined Docker (canonical's snap package) can't bind-mount
    # /tmp — use a path under $HOME which the confinement allows. Caller
    # could override via ERLDISTPY_TLS_OTP26_WORKDIR.
    import os as _os
    import tempfile as _tempfile

    home_base = _os.environ.get(
        "ERLDISTPY_TLS_OTP26_WORKDIR", str(Path.home() / ".erldistpy-test")
    )
    Path(home_base).mkdir(parents=True, exist_ok=True)
    workdir = Path(_tempfile.mkdtemp(prefix="tls_otp26_", dir=home_base))
    _os.chmod(workdir, 0o755)
    host_certs = _gen_certs(workdir)
    # Inside the container the certs land at the same path because we
    # mount workdir → /certs:ro and we generated under workdir.
    container_certs = {
        k: v.replace(str(workdir), "/certs") for k, v in host_certs.items()
    }
    ssl_config_host = _write_ssl_config(workdir, container_certs)
    ssl_config_container = ssl_config_host.replace(str(workdir), "/certs")

    container_name = f"erldistpy-otp26-tls-{int(time.time())}"

    proc = subprocess.Popen(
        [
            "docker", "run", "--rm",
            "--name", container_name,
            "--network", "host",
            "-v", f"{workdir}:/certs:ro",
            "erlang:25",
            "erl",
            "-sname", TLS_SNAME,
            "-setcookie", TLS_COOKIE,
            "-proto_dist", "inet_tls",
            "-ssl_dist_optfile", ssl_config_container,
            "-noshell",
            "-eval", ERL_GEN_BOOT,
        ],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )

    deadline = time.monotonic() + 20.0
    ready = False
    while time.monotonic() < deadline:
        time.sleep(0.3)
        if _epmd_node_present(TLS_SNAME):
            time.sleep(0.3)
            ready = True
            break
    if not ready:
        subprocess.run(["docker", "rm", "-f", container_name], capture_output=True)
        proc.terminate()
        proc.wait(timeout=3)
        pytest.skip(f"OTP 26 TLS docker node {TLS_SNAME} did not register")

    yield TLS_SNAME, TLS_COOKIE, host_certs

    subprocess.run(["docker", "rm", "-f", container_name], capture_output=True)
    proc.terminate()
    try:
        proc.wait(timeout=3)
    except subprocess.TimeoutExpired:
        proc.kill()


def test_call_over_tls_otp26(otp26_tls_peer):
    """The minimum repro of the production failure: TLS-dist on OTP 26.

    On erldistpy ≤0.1.7 against the prod portal this surfaces as
    ``ChannelError: peer closed after 0/4 bytes`` on the first call.
    """
    sname, cookie, certs = otp26_tls_peer
    ctx = make_dist_tls_context(
        cert=certs["client_cert"],
        key=certs["client_key"],
        ca=certs["ca"],
    )
    with Node(
        our_name="erldistpy_test_otp26_tls@localhost",
        peer_name=sname,
        cookie=cookie,
        tls_context=ctx,
    ) as n:
        assert n.tls is True
        reply = n.call("gen_target", (Atom("ping"), 7), timeout=5.0)
        assert reply == (Atom("pong"), 7)


# ---------------------------------------------------------------------------
# OTP 26 + TLS dist + Elixir GenServer — the exact production combo
# ---------------------------------------------------------------------------


ELIXIR_TLS_SNAME = "erldistpy_otp26_elixir_tls"
ELIXIR_TLS_COOKIE = "ERLDISTPY_ELIXIR_TLS"

# Inline Elixir script: spawn a GenServer registered as `gen_target` that
# answers the same {ping, X} / {add, A, B} calls. This is the SHAPE the
# real Wallet.Bridge uses — GenServer.handle_call/3 routing via the
# `:"$gen_call"` envelope, not a hand-coded receive.
ELIXIR_GENSERVER_BOOT = """
defmodule GenTarget do
  use GenServer

  def start_link, do: GenServer.start_link(__MODULE__, %{}, name: :gen_target)

  def init(state), do: {:ok, state}

  def handle_call({:ping, x}, _from, state), do: {:reply, {:pong, x}, state}
  def handle_call({:add, a, b}, _from, state), do: {:reply, {:ok, a + b}, state}
  def handle_call(other, _from, state), do: {:reply, {:error, {:bad_request, other}}, state}
end

{:ok, _} = GenTarget.start_link()
Process.sleep(:infinity)
"""


@pytest.fixture(scope="module")
def otp26_elixir_tls_peer(tmp_path_factory):
    """Elixir GenServer over TLS on OTP 26 — exactly what portal runs.

    Difference from otp26_tls_peer: the registered process is a real
    Elixir GenServer (uses `gen_server:reply/2` for replies, all the
    OTP machinery) instead of a hand-coded receive loop.
    """
    if not _docker_available():
        pytest.skip("docker not installed")
    if not shutil.which("openssl"):
        pytest.skip("openssl not installed")
    if not _socket_open("127.0.0.1", 4369):
        pytest.skip("EPMD not running on host")

    import os as _os
    import tempfile as _tempfile

    home_base = _os.environ.get(
        "ERLDISTPY_TLS_OTP26_WORKDIR", str(Path.home() / ".erldistpy-test")
    )
    Path(home_base).mkdir(parents=True, exist_ok=True)
    workdir = Path(_tempfile.mkdtemp(prefix="elixir_tls_otp26_", dir=home_base))
    _os.chmod(workdir, 0o755)
    host_certs = _gen_certs(workdir)
    container_certs = {
        k: v.replace(str(workdir), "/certs") for k, v in host_certs.items()
    }
    _write_ssl_config(workdir, container_certs)
    ssl_config_container = "/certs/ssl_dist.config"

    boot_script = workdir / "boot.exs"
    boot_script.write_text(ELIXIR_GENSERVER_BOOT)

    container_name = f"erldistpy-elixir-tls-{int(time.time())}"

    proc = subprocess.Popen(
        [
            "docker", "run", "--rm",
            "--name", container_name,
            "--network", "host",
            "-v", f"{workdir}:/certs:ro",
            "elixir:1.16-otp-26",
            "elixir",
            "--sname", ELIXIR_TLS_SNAME,
            "--cookie", ELIXIR_TLS_COOKIE,
            "--erl", f"-proto_dist inet_tls -ssl_dist_optfile {ssl_config_container}",
            "/certs/boot.exs",
        ],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )

    deadline = time.monotonic() + 25.0
    ready = False
    while time.monotonic() < deadline:
        time.sleep(0.3)
        if _epmd_node_present(ELIXIR_TLS_SNAME):
            time.sleep(0.5)  # Elixir GenServer needs an extra beat to register
            ready = True
            break
    if not ready:
        subprocess.run(["docker", "rm", "-f", container_name], capture_output=True)
        proc.terminate()
        proc.wait(timeout=3)
        pytest.skip(f"Elixir TLS docker node {ELIXIR_TLS_SNAME} did not register")

    yield ELIXIR_TLS_SNAME, ELIXIR_TLS_COOKIE, host_certs

    subprocess.run(["docker", "rm", "-f", container_name], capture_output=True)
    proc.terminate()
    try:
        proc.wait(timeout=3)
    except subprocess.TimeoutExpired:
        proc.kill()


def test_call_elixir_genserver_over_tls_otp26(otp26_elixir_tls_peer):
    """Repro of the prod failure: erldistpy calls an Elixir GenServer over TLS.

    Difference from test_call_over_tls_otp26: peer is a real
    GenServer, not a hand-coded receive. Replies go through
    :gen_server.reply/2 which sends via :erlang.send/2 — same path as
    Wallet.Bridge in production.
    """
    sname, cookie, certs = otp26_elixir_tls_peer
    ctx = make_dist_tls_context(
        cert=certs["client_cert"],
        key=certs["client_key"],
        ca=certs["ca"],
    )
    with Node(
        our_name="erldistpy_test_elixir_tls@localhost",
        peer_name=sname,
        cookie=cookie,
        tls_context=ctx,
    ) as n:
        assert n.tls is True
        reply = n.call("gen_target", (Atom("ping"), 7), timeout=5.0)
        assert reply == (Atom("pong"), 7)