Isolate tests from external GraphServers

src/ezmsg/core/backend.py

@@ -698,16 +698,17 @@ async def create_graph_context() -> GraphContext:
             self._graph_context = graph_context
             self._graph_server_spawned = graph_context._graph_server is not None
 
+            address = graph_context.graph_address
+            if address is None:
+                address = GraphService.default_address()
+
             if graph_context._graph_server is None:
-                address = graph_context.graph_address
-                if address is None:
-                    address = GraphService.default_address()
                 logger.info(f"Connected to GraphServer @ {address}")
             else:
-                logger.info(f"Spawned GraphServer @ {graph_context.graph_address}")
+                logger.info(f"Spawned GraphServer @ {address}")
 
             self._execution_context.create_processes(
-                graph_address=graph_context.graph_address,
+                graph_address=address,
                 backend_process=self._backend_process,
             )
 

tests/conftest.py

@@ -0,0 +1,40 @@
+"""Run every test against a fresh, suite-owned GraphServer.
+
+The server binds directly to an OS-assigned loopback port. Before the test
+runs, both the environment inherited by child processes and ezmsg's imported
+address constants are pointed at that server. Tests that pass an explicit
+server address continue to use their own server.
+"""
+
+import pytest
+
+from ezmsg.core import channelmanager, netprotocol
+from ezmsg.core.graphserver import GraphService
+
+assert netprotocol.GRAPHSERVER_ADDR_ENV == "EZMSG_GRAPHSERVER_ADDR", (
+    "The env var this fixture sets no longer matches ezmsg's; update both together."
+)
+
+
+@pytest.fixture(autouse=True)
+def hermetic_graph_server(monkeypatch: pytest.MonkeyPatch):
+    """Provide a fresh GraphServer on a fresh loopback port for each test."""
+    for _ in range(10):
+        service = GraphService(address=("127.0.0.1", 0))
+        server = service.create_server()
+        if server.address.port != netprotocol.GRAPHSERVER_PORT_DEFAULT:
+            break
+        server.stop()
+    else:
+        raise RuntimeError("Could not allocate a non-default GraphServer port")
+
+    address = str(server.address)
+    monkeypatch.setenv(netprotocol.GRAPHSERVER_ADDR_ENV, address)
+    monkeypatch.setattr(netprotocol, "GRAPHSERVER_ADDR", address)
+    monkeypatch.setattr(channelmanager, "GRAPHSERVER_ADDR", address)
+
+    try:
+        yield server
+    finally:
+        server.stop()
+        assert not server.is_alive(), "Hermetic GraphServer failed to stop"

tests/test_attach.py

@@ -1,12 +1,11 @@
-import pytest
 import asyncio
-import ezmsg.core as ez
-
-from ezmsg.core.graphserver import GraphService
-
+from collections.abc import AsyncGenerator
 from multiprocessing import Process
+from pathlib import Path
 
-from collections.abc import AsyncGenerator
+import pytest
+
+import ezmsg.core as ez
 
 
 class TransmitReceiveSettings(ez.Settings):
@@ -75,6 +74,8 @@ def __init__(self, settings: TransmitReceiveSettings) -> None:
 TX_TOPIC = "TX"
 RX_TOPIC = "RX"
 ACK_TOPIC = "ACK"
+PROCESS_TIMEOUT = 30.0
+PROCESS_CLEANUP_TIMEOUT = 5.0
 
 
 class TransmitReceiveProcess(AttachTestProcess):
@@ -104,31 +105,72 @@ def run(self) -> None:
             )
 
 
-@pytest.mark.asyncio
-@pytest.mark.skip(reason="canonical port isn't always available")
-async def test_attach():
-    graph_service = GraphService(address=GraphService.default_address())
-    graph_server = graph_service.create_server()
+async def wait_for_processes(processes: list[Process]) -> None:
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + PROCESS_TIMEOUT
+    remaining = list(processes)
 
-    async with ez.GraphContext(graph_service):
-        settings = TransmitReceiveSettings()
+    while remaining:
+        for process in list(remaining):
+            if process.is_alive():
+                continue
+            process.join()
+            assert process.exitcode == 0, (
+                f"{process.name} exited with status {process.exitcode}"
+            )
+            remaining.remove(process)
+
+        if not remaining:
+            return
+        if loop.time() >= deadline:
+            names = ", ".join(process.name for process in remaining)
+            raise AssertionError(
+                f"Processes did not exit within {PROCESS_TIMEOUT}s: {names}"
+            )
 
-        txrx_process = TransmitReceiveProcess(settings)
-        txrx_process.start()
+        await asyncio.sleep(0.05)
 
-        echo_process = AttachEchoProcess(settings)
-        echo_process.start()
 
-        echo_process.join()
-        txrx_process.join()
+async def close_process(process: Process) -> None:
+    if process.is_alive():
+        process.terminate()
+        await asyncio.to_thread(process.join, PROCESS_CLEANUP_TIMEOUT)
+    if process.is_alive():
+        process.kill()
+        await asyncio.to_thread(process.join, PROCESS_CLEANUP_TIMEOUT)
+
+    assert not process.is_alive(), f"Could not stop {process.name}"
+    process.close()
 
-    graph_server.stop()
 
+@pytest.mark.asyncio
+async def test_attach(monkeypatch: pytest.MonkeyPatch):
+    """Independent processes attach to one already-running default server.
+
+    Previously skipped as "canonical port isn't always available": the test
+    needed the shared default port, which anything on the machine could
+    occupy. The hermetic conftest pins the default to a session-private
+    address and runs a server there for every test, so attaching — from
+    this process and from the spawned children, which inherit the pinned
+    environment — is reliable. The conftest's server IS the attach target;
+    the test no longer creates its own.
+    """
+    # pytest's importlib mode does not put the repository root on sys.path.
+    # Spawned processes need it there to unpickle these test process classes.
+    monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[1]))
+
+    async with ez.GraphContext():
+        settings = TransmitReceiveSettings()
+        txrx_process = TransmitReceiveProcess(settings)
+        echo_process = AttachEchoProcess(settings)
+        started_processes: list[Process] = []
+
+        try:
+            for process in (txrx_process, echo_process):
+                process.start()
+                started_processes.append(process)
 
-if __name__ == "__main__":
-    loop = asyncio.new_event_loop()
-    try:
-        asyncio.set_event_loop(loop)
-        loop.run_until_complete(test_attach())
-    finally:
-        loop.close()
+            await wait_for_processes(started_processes)
+        finally:
+            for process in started_processes:
+                await close_process(process)

tests/test_hermetic_conftest.py

@@ -0,0 +1,64 @@
+"""The hermetic-conftest contract: private defaults, fresh server per test."""
+
+import os
+
+import pytest
+
+from ezmsg.core import channelmanager, netprotocol
+from ezmsg.core.graphserver import GraphService
+from ezmsg.core.netprotocol import close_stream_writer
+
+
+class TestHermeticDefaults:
+    def test_every_default_resolution_agrees_on_the_pinned_address(self):
+        pinned = os.environ[netprotocol.GRAPHSERVER_ADDR_ENV]
+        assert netprotocol.GRAPHSERVER_ADDR == pinned
+        assert channelmanager.GRAPHSERVER_ADDR == pinned
+        assert str(GraphService.default_address()) == pinned
+        # The whole point: the pinned port is NOT the shared default one a
+        # developer's live server may occupy.
+        pinned_port = int(pinned.rsplit(":", 1)[1])
+        assert pinned_port != netprotocol.GRAPHSERVER_PORT_DEFAULT
+
+    @pytest.mark.asyncio
+    async def test_default_clients_attach_to_the_per_test_server(self):
+        service = GraphService()  # no address: resolves the pinned default
+        started = await service.ensure()
+        # Attach, not start: the autouse server is already listening there.
+        assert started is None
+        _reader, writer = await service.open_connection()
+        await close_stream_writer(writer)
+
+    @pytest.mark.asyncio
+    async def test_implicit_auto_start_still_creates_a_private_server(
+        self, monkeypatch: pytest.MonkeyPatch
+    ):
+        # Exercise the real implicit-start decision without touching the
+        # canonical port: port 0 lets the OS choose the server's address.
+        monkeypatch.delenv(netprotocol.GRAPHSERVER_ADDR_ENV)
+        monkeypatch.setattr(GraphService, "PORT_DEFAULT", 0)
+        monkeypatch.setenv(netprotocol.SERVER_PORT_START_ENV, "0")
+
+        service = GraphService()
+        server = await service.ensure()
+        assert server is not None
+        try:
+            assert service.address.port != 0
+            _reader, writer = await service.open_connection()
+            await close_stream_writer(writer)
+        finally:
+            server.stop()
+            assert not server.is_alive()
+
+
+_SERVERS_SEEN: list[object] = []
+
+
+class TestPerTestFreshness:
+    # Strong references keep ids unique for the comparison below.
+    def test_server_is_fresh_per_test_first(self, hermetic_graph_server):
+        _SERVERS_SEEN.append(hermetic_graph_server)
+
+    def test_server_is_fresh_per_test_second(self, hermetic_graph_server):
+        _SERVERS_SEEN.append(hermetic_graph_server)
+        assert len({id(server) for server in _SERVERS_SEEN}) == len(_SERVERS_SEEN)

tests/test_run.py

@@ -3,6 +3,8 @@
 import pytest
 
 import ezmsg.core as ez
+from ezmsg.core.backend import ExecutionContext
+from ezmsg.core.graphserver import GraphService
 
 from ez_test_utils import (
     get_test_fn,
@@ -79,6 +81,23 @@ def test_local_system(toy_system_fixture, num_messages):
         assert len(results) == num_messages
 
 
+def test_default_graph_address_is_resolved_before_process_creation(monkeypatch):
+    captured_addresses: list[object] = []
+    create_processes = ExecutionContext.create_processes
+
+    def capture_graph_address(self, graph_address, backend_process):
+        captured_addresses.append(graph_address)
+        create_processes(self, graph_address, backend_process)
+
+    monkeypatch.setattr(ExecutionContext, "create_processes", capture_graph_address)
+
+    with get_test_fn() as test_filename:
+        system = ToySystem(ToySystemSettings(num_msgs=1, output_fn=str(test_filename)))
+        ez.run(SYSTEM=system, force_single_process=True)
+
+    assert captured_addresses == [GraphService.default_address()]
+
+
 @pytest.mark.parametrize("passthrough_settings", [False, True])
 @pytest.mark.parametrize("num_messages", [1, 5, 10])
 def test_run_comps_conns(passthrough_settings, num_messages):