erldistpy/tests/test_node.py

"""High-level Node tests.

Pure tests cover ref/pid generation. Live tests spawn an erl node with
a process that implements the gen_server reply protocol by hand
(no module compilation needed, just a receive on $gen_call) and
exercise Node.call() end-to-end.
"""

from __future__ import annotations

import os
import shutil
import socket
import subprocess
import time

import pytest

from erldistpy.etf import Atom, Reference
from erldistpy.node import CallProtocolError, CallTimeout, Node, NodeError

# Boot script registers `gen_target` which mimics gen_server's reply
# protocol. Two operations supported:
#   {ping, X}        -> {pong, X}
#   {add, A, B}      -> {ok, A + B}
#   slow             -> sleeps 2s before replying — used to force timeout
#
# The Y-combinator trick lets the receive loop be expressed inline.
READY_FLAG = "/tmp/erldistpy_test_node_ready"
ERL_GEN_BOOT = (
    "Handle = fun "
    "  ({ping, X}) -> {pong, X}; "
    "  ({add, A, B}) -> {ok, A + B}; "
    "  (slow) -> timer:sleep(2000), {ok, slow}; "
    "  (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), "
    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 gen_peer():
    if not shutil.which("erl"):
        pytest.skip("erl not installed")
    if not _epmd_running():
        pytest.skip("EPMD not running on localhost")

    if os.path.exists(READY_FLAG):
        os.remove(READY_FLAG)

    sname = "erldistpy_gen"
    cookie = "ERLDISTPY_GEN_COOKIE"
    proc = subprocess.Popen(
        ["erl", "-sname", sname, "-setcookie", cookie, "-noshell", "-eval", ERL_GEN_BOOT],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )

    deadline = time.monotonic() + 6.0
    from erldistpy.epmd import lookup
    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 gen_target did not become ready")

    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)


# --------------------------------------------------------------------------
# Pure validation
# --------------------------------------------------------------------------


def test_node_requires_qualified_our_name():
    with pytest.raises(ValueError, match="name@host"):
        Node(our_name="bare", peer_name="x", cookie="X")


def test_node_unknown_peer_raises():
    with pytest.raises(NodeError, match="not registered"):
        Node(
            our_name="erldistpy_test@localhost",
            peer_name="definitely_not_a_real_node_qzxw",
            cookie="X",
            connect_timeout=1.0,
        )


# --------------------------------------------------------------------------
# Live call() against a process implementing $gen_call
# --------------------------------------------------------------------------


def _client_name() -> str:
    return f"erldistpy_node@{socket.gethostname()}"


def test_call_ping(gen_peer):
    sname, cookie = gen_peer
    with Node(our_name=_client_name(), peer_name=sname, cookie=cookie) as node:
        reply = node.call("gen_target", (Atom("ping"), 42))
        assert reply == (Atom("pong"), 42)


def test_call_add(gen_peer):
    sname, cookie = gen_peer
    with Node(our_name=_client_name(), peer_name=sname, cookie=cookie) as node:
        reply = node.call("gen_target", (Atom("add"), 7, 35))
        assert reply == (Atom("ok"), 42)


def test_call_multiple_in_sequence(gen_peer):
    """Each call must increment Ref and not crosstalk with prior calls."""
    sname, cookie = gen_peer
    with Node(our_name=_client_name(), peer_name=sname, cookie=cookie) as node:
        for i in range(5):
            reply = node.call("gen_target", (Atom("add"), i, 1))
            assert reply == (Atom("ok"), i + 1)


def test_call_bad_request_surfaces_server_error(gen_peer):
    sname, cookie = gen_peer
    with Node(our_name=_client_name(), peer_name=sname, cookie=cookie) as node:
        reply = node.call("gen_target", Atom("nonsense"))
        # Server returns {error, {bad_request, Other}} per our handle fun
        assert isinstance(reply, tuple)
        assert reply[0] == Atom("error")


def test_call_timeout(gen_peer):
    """Server sleeps 2s; our timeout is 0.3s — must raise CallTimeout."""
    sname, cookie = gen_peer
    with (
        Node(our_name=_client_name(), peer_name=sname, cookie=cookie) as node,
        pytest.raises(CallTimeout),
    ):
        node.call("gen_target", Atom("slow"), timeout=0.3)


def test_call_unknown_registered_name_times_out(gen_peer):
    """Sending to a non-existent registered name is silently dropped by
    Erlang — we must surface this as a timeout, not hang forever."""
    sname, cookie = gen_peer
    with (
        Node(our_name=_client_name(), peer_name=sname, cookie=cookie) as node,
        pytest.raises(CallTimeout),
    ):
        node.call("does_not_exist", (Atom("ping"), 1), timeout=0.4)


def test_ref_counter_is_per_node_lifetime():
    """Synthesized references must be unique within a Node instance."""
    # We don't need a live peer; just exercise _next_ref via a dummy bypass.
    # Construct a Node without going through __init__ to avoid the network.
    node = Node.__new__(Node)
    node._our_atom = Atom("x@y")
    node.our_creation = 1
    node._ref_counter = 0
    refs = [node._next_ref() for _ in range(100)]
    assert len(set((r.creation, r.ids) for r in refs)) == 100
    assert all(isinstance(r, Reference) for r in refs)


def test_protocol_error_path_uses_callprotocolerror():
    """CallProtocolError exists as a distinct exception so callers can
    distinguish 'server misbehaved' from 'server timed out'."""
    assert issubclass(CallTimeout, NodeError)
    assert issubclass(CallProtocolError, NodeError)