"""Distribution channel tests.

Unit tests cover the encode/decode of pass-through messages.
Live tests spawn an Erlang node with a registered echo process,
complete the full EPMD lookup + handshake + REG_SEND + recv cycle,
and verify the round-tripped payload matches.
"""

from __future__ import annotations

import shutil
import socket
import struct
import subprocess
import time

import pytest

from erldistpy.channel import (
    OP_REG_SEND,
    OP_SEND,
    OP_SEND_SENDER,
    PASS_THROUGH,
    Channel,
    ChannelError,
    IncomingMessage,
    _decode_message,
    _encode_message,
)
from erldistpy.epmd import lookup
from erldistpy.etf import Atom, Pid, encode
from erldistpy.handshake import handshake

# --------------------------------------------------------------------------
# Pure encode/decode
# --------------------------------------------------------------------------


def test_encode_message_control_only():
    """LINK/UNLINK/etc. have no payload — just a control tuple."""
    control = (1, Pid(Atom("a@b"), 0, 0, 1), Pid(Atom("c@d"), 0, 0, 1))
    body = _encode_message(control, None)
    assert body[0] == PASS_THROUGH
    # Decode round-trip
    msg = _decode_message(body)
    assert msg.control == control
    assert msg.payload is None


def test_encode_message_with_payload():
    control = (OP_REG_SEND, Pid(Atom("client@h"), 1, 0, 5), Atom(""), Atom("echo"))
    payload = (Atom("hello"), b"world")
    body = _encode_message(control, payload)
    msg = _decode_message(body)
    assert msg.control == control
    assert msg.payload == payload


def test_decode_rejects_wrong_first_byte():
    with pytest.raises(ChannelError, match="pass-through .0x70. or dist header"):
        _decode_message(b"\x00" + encode((1,)))


def test_decode_rejects_non_tuple_control():
    body = bytes([PASS_THROUGH]) + encode(42)
    with pytest.raises(ChannelError, match="not a tuple"):
        _decode_message(body)


def test_decode_rejects_trailing_garbage():
    body = bytes([PASS_THROUGH]) + encode((1,)) + b"\xFF"
    with pytest.raises(ChannelError):
        _decode_message(body)


def test_incoming_message_op_property():
    msg = IncomingMessage(control=(OP_SEND, Atom(""), Pid(Atom("x@y"), 0, 0, 1)), payload=42)
    assert msg.op == OP_SEND


# --------------------------------------------------------------------------
# Raw socket framing (loopback with a paired socket)
# --------------------------------------------------------------------------


def _socket_pair() -> tuple[socket.socket, socket.socket]:
    return socket.socketpair()


def test_send_raw_recv_raw_round_trip():
    a, b = _socket_pair()
    ch_a = Channel(a)
    ch_b = Channel(b)
    ch_a.send_raw(b"hello")
    assert ch_b.recv_raw() == b"hello"


def test_tick_round_trip():
    a, b = _socket_pair()
    ch_a = Channel(a)
    ch_b = Channel(b)
    ch_a.send_tick()
    assert ch_b.recv_raw() == b""


def test_send_control_recv_message():
    a, b = _socket_pair()
    ch_a = Channel(a)
    ch_b = Channel(b)
    control = (OP_REG_SEND, Pid(Atom("a@h"), 1, 0, 7), Atom(""), Atom("echo"))
    payload = (Atom("greet"), b"hi")
    ch_a.send_control(control, payload)
    msg = ch_b.recv_message()
    assert msg is not None
    assert msg.control == control
    assert msg.payload == payload


def test_recv_message_skips_ticks():
    a, b = _socket_pair()
    ch_a = Channel(a)
    ch_b = Channel(b)
    ch_a.send_tick()
    ch_a.send_tick()
    control = (OP_SEND, Atom(""), Pid(Atom("x@y"), 0, 0, 1))
    ch_a.send_control(control, b"payload")
    msg = ch_b.recv_message()
    assert msg is not None
    assert msg.control == control
    assert msg.payload == b"payload"


def test_send_reg_send_helper():
    a, b = _socket_pair()
    ch_a = Channel(a)
    ch_b = Channel(b)
    from_pid = Pid(Atom("client@h"), 42, 0, 9)
    ch_a.send_reg_send(from_pid, Atom("kernel"), (Atom("ping"), 1))
    msg = ch_b.recv_message()
    assert msg is not None
    assert msg.control == (OP_REG_SEND, from_pid, Atom(""), Atom("kernel"))
    assert msg.payload == (Atom("ping"), 1)


def test_recv_after_peer_closes_raises():
    a, b = _socket_pair()
    ch_a = Channel(a)
    Channel(b).close()
    with pytest.raises(ChannelError, match="peer closed"):
        # The 4-byte length header will fail because peer closed
        ch_a.recv_raw()


def test_recv_handles_truncated_length_header():
    """If only 2 of 4 length bytes arrive then peer dies, we must error
    instead of hanging or returning garbage."""
    a, b = _socket_pair()
    a.sendall(b"\x00\x00")  # only 2 of 4 header bytes
    a.close()
    ch_b = Channel(b)
    with pytest.raises(ChannelError, match="peer closed"):
        ch_b.recv_raw()


# --------------------------------------------------------------------------
# Live end-to-end — EPMD + handshake + REG_SEND + recv reply
# --------------------------------------------------------------------------


# Spawn an erl node with a tiny echo process registered as `echo`.
# The Y-combinator trick lets us write a recursive fun inside -eval.
# A flag file is written after registration so the fixture can wait for
# the process to actually be reachable (EPMD registers the node *before*
# our -eval runs, so polling EPMD alone is racy).
READY_FLAG = "/tmp/erldistpy_test_channel_ready"
ERL_ECHO_BOOT = (
    "EchoLoop = fun(F) -> "
    "receive {From, Msg} -> From ! {echoed, Msg}, F(F) end "
    "end, "
    "Pid = spawn(fun() -> EchoLoop(EchoLoop) end), "
    "register(echo, Pid), "
    f'file:write_file("{READY_FLAG}", "1"), '
    "timer:sleep(infinity)."
)


def _epmd_running() -> bool:
    try:
        with socket.create_connection(("localhost", 4369), timeout=0.5):
            return True
    except OSError:
        return False


@pytest.fixture(scope="module")
def echo_node():
    if not shutil.which("erl"):
        pytest.skip("erl not installed")
    if not _epmd_running():
        pytest.skip("EPMD not running on localhost")

    import os
    # Clear stale flag from a previous run
    if os.path.exists(READY_FLAG):
        os.remove(READY_FLAG)

    sname = "erldistpy_echo"
    cookie = "ERLDISTPY_CH_COOKIE"
    proc = subprocess.Popen(
        ["erl", "-sname", sname, "-setcookie", cookie, "-noshell", "-eval", ERL_ECHO_BOOT],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    # Wait for BOTH: node registered with EPMD AND echo process up
    deadline = time.monotonic() + 6.0
    ready = False
    while time.monotonic() < deadline:
        time.sleep(0.1)
        if lookup(sname, timeout=0.5) is None:
            continue
        if os.path.exists(READY_FLAG):
            ready = True
            break
    if not ready:
        proc.terminate()
        proc.wait(timeout=2)
        pytest.skip("erl echo process did not become ready in time")

    yield sname, cookie

    proc.terminate()
    try:
        proc.wait(timeout=2)
    except subprocess.TimeoutExpired:
        proc.kill()
    if os.path.exists(READY_FLAG):
        os.remove(READY_FLAG)


def test_live_reg_send_echo(echo_node):
    sname, cookie = echo_node
    info = lookup(sname)
    assert info is not None
    sock = socket.create_connection(("localhost", info.port), timeout=2)
    try:
        client_name = f"erldistpy_ch@{socket.gethostname()}"
        result = handshake(sock, our_name=client_name, cookie=cookie)
        ch = Channel(sock, recv_timeout=3.0)
        from_pid = Pid(Atom(client_name), 1, 0, result.our_creation)
        message_body = b"hello over native dist"
        ch.send_reg_send(from_pid, Atom("echo"), (from_pid, message_body))
        msg = ch.recv_message()
        assert msg is not None
        # Peer may use OP_SEND (legacy) or OP_SEND_SENDER (if it offered DFLAG_SEND_SENDER)
        assert msg.op in (OP_SEND, OP_SEND_SENDER)
        # Payload echoes our tuple form
        assert msg.payload == (Atom("echoed"), message_body)
    finally:
        sock.close()


def test_live_reg_send_survives_tick(echo_node):
    """An inbound tick frame before the reply must not break recv_message()."""
    sname, cookie = echo_node
    info = lookup(sname)
    assert info is not None
    sock = socket.create_connection(("localhost", info.port), timeout=2)
    try:
        client_name = f"erldistpy_ch_tick@{socket.gethostname()}"
        result = handshake(sock, our_name=client_name, cookie=cookie)
        ch = Channel(sock, recv_timeout=3.0)
        ch.send_tick()  # provoke nothing in particular — just exercise the wire
        from_pid = Pid(Atom(client_name), 2, 0, result.our_creation)
        ch.send_reg_send(from_pid, Atom("echo"), (from_pid, b"after tick"))
        msg = ch.recv_message()
        assert msg is not None
        assert msg.payload == (Atom("echoed"), b"after tick")
    finally:
        sock.close()


# Mirror struct import only used implicitly by ETF; keep linters happy.
_ = struct