diff --git a/src/ezmsg/core/backend.py b/src/ezmsg/core/backend.py index 8b150798..fa644d5e 100644 --- a/src/ezmsg/core/backend.py +++ b/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, ) diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 00000000..1068cddb --- /dev/null +++ b/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" diff --git a/tests/test_attach.py b/tests/test_attach.py index 95ae7425..125dc208 100644 --- a/tests/test_attach.py +++ b/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) diff --git a/tests/test_hermetic_conftest.py b/tests/test_hermetic_conftest.py new file mode 100644 index 00000000..3496aa4b --- /dev/null +++ b/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) diff --git a/tests/test_run.py b/tests/test_run.py index 323bc4ba..35e69a72 100644 --- a/tests/test_run.py +++ b/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):