"""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)