diff --git a/plugboard/connector/connector.py b/plugboard/connector/connector.py index ceeb35c3..04aaca7e 100644 --- a/plugboard/connector/connector.py +++ b/plugboard/connector/connector.py @@ -17,6 +17,10 @@ class Connector(ABC, ExportMixin): def __init__(self, spec: ConnectorSpec, *args: _t.Any, **kwargs: _t.Any) -> None: self.spec: ConnectorSpec = spec + async def init(self) -> None: + """Acquire resources required by this connector.""" + pass + @abstractmethod async def connect_send(self) -> Channel: """Returns a `Channel` for sending messages.""" diff --git a/plugboard/connector/ray_channel.py b/plugboard/connector/ray_channel.py index 63a653be..2db7d533 100644 --- a/plugboard/connector/ray_channel.py +++ b/plugboard/connector/ray_channel.py @@ -1,5 +1,6 @@ """Provides `RayChannel` for use in cluster compute environments.""" +import asyncio import typing as _t from plugboard.connector.asyncio_channel import AsyncioChannel @@ -26,7 +27,7 @@ def __init__( # noqa: D417 actor_options: _t.Optional[dict[str, _t.Any]] = None, **kwargs: _t.Any, ) -> None: - """Instantiates `RayChannel`. + """Instantiates `RayChannel` and creates its remote actor. Args: actor_options: Optional; Options to pass to the Ray actor. Defaults to {"num_cpus": 0}. @@ -71,12 +72,21 @@ def __init__(self, *args: _t.Any, **kwargs: _t.Any) -> None: super().__init__(*args, **kwargs) if self.spec.mode != ConnectorMode.PIPELINE: raise ValueError("RayConnector only supports `PIPELINE` type connections.") - self._channel = RayChannel() + self._channel: _t.Optional[RayChannel] = None + self._init_lock = asyncio.Lock() + + async def init(self) -> None: + """Create the channel and its remote actor when execution starts.""" + async with self._init_lock: + if self._channel is None: + self._channel = RayChannel() async def connect_send(self) -> RayChannel: """Returns a `RayChannel` for sending messages.""" - return self._channel + await self.init() + return _t.cast(RayChannel, self._channel) async def connect_recv(self) -> RayChannel: """Returns a `RayChannel` for receiving messages.""" - return self._channel + await self.init() + return _t.cast(RayChannel, self._channel) diff --git a/plugboard/connector/zmq_channel.py b/plugboard/connector/zmq_channel.py index 4e773214..9d1734fc 100644 --- a/plugboard/connector/zmq_channel.py +++ b/plugboard/connector/zmq_channel.py @@ -105,6 +105,7 @@ def __init__( super().__init__(*args, **kwargs) self._zmq_address = zmq_address self._maxsize = maxsize + self._init_lock = asyncio.Lock() @abstractmethod async def connect_send(self) -> ZMQChannel: @@ -124,23 +125,31 @@ def __init__(self, *args: _t.Any, **kwargs: _t.Any) -> None: super().__init__(*args, **kwargs) self._send_channel: _t.Optional[ZMQChannel] = None self._recv_channel: _t.Optional[ZMQChannel] = None - - # Socket to receive sender address from sender - self._sender_rep_socket = create_socket(zmq.REP, []) - self._sender_rep_socket_port = self._sender_rep_socket.bind_to_random_port("tcp://*") - self._sender_rep_socket_addr = f"{self._zmq_address}:{self._sender_rep_socket_port}" - self._sender_req_lock = asyncio.Lock() + self._sender_rep_socket: _t.Optional[zmq_asyncio.Socket] = None + self._sender_rep_socket_addr: _t.Optional[str] = None + self._sender_req_lock: _t.Optional[asyncio.Lock] = None self._sender_addr: _t.Optional[str] = None - - # Socket to send sender address to receiver - self._receiver_rep_socket = create_socket(zmq.REP, []) - self._receiver_rep_socket_port = self._receiver_rep_socket.bind_to_random_port("tcp://*") - self._receiver_rep_socket_addr = f"{self._zmq_address}:{self._receiver_rep_socket_port}" - self._receiver_req_lock = asyncio.Lock() - - self._exchange_addr_task = asyncio.create_task(self._exchange_address()) - _zmq_exchange_addr_tasks.add(self._exchange_addr_task) - self._exchange_addr_task.add_done_callback(_zmq_exchange_addr_tasks.discard) + self._receiver_rep_socket: _t.Optional[zmq_asyncio.Socket] = None + self._receiver_rep_socket_addr: _t.Optional[str] = None + self._receiver_req_lock: _t.Optional[asyncio.Lock] = None + self._exchange_addr_task: _t.Optional[asyncio.Task[None]] = None + + async def init(self) -> None: + """Allocate address exchange sockets when execution starts.""" + async with self._init_lock: + if self._sender_rep_socket_addr is not None: + return + self._sender_rep_socket = create_socket(zmq.REP, []) + sender_port = self._sender_rep_socket.bind_to_random_port("tcp://*") + self._sender_rep_socket_addr = f"{self._zmq_address}:{sender_port}" + self._sender_req_lock = asyncio.Lock() + self._receiver_rep_socket = create_socket(zmq.REP, []) + receiver_port = self._receiver_rep_socket.bind_to_random_port("tcp://*") + self._receiver_rep_socket_addr = f"{self._zmq_address}:{receiver_port}" + self._receiver_req_lock = asyncio.Lock() + self._exchange_addr_task = asyncio.create_task(self._exchange_address()) + _zmq_exchange_addr_tasks.add(self._exchange_addr_task) + self._exchange_addr_task.add_done_callback(_zmq_exchange_addr_tasks.discard) def __getstate__(self) -> dict: state = self.__dict__.copy() @@ -152,6 +161,7 @@ def __getstate__(self) -> dict: "_exchange_addr_task", "_send_channel", "_recv_channel", + "_init_lock", ): if attr in state: del state[attr] @@ -161,28 +171,41 @@ def __setstate__(self, state: dict) -> None: self.__dict__.update(state) self._send_channel = None self._recv_channel = None + self._init_lock = asyncio.Lock() async def _exchange_address(self) -> None: + if ( + self._sender_req_lock is None + or self._sender_rep_socket is None + or self._receiver_req_lock is None + or self._receiver_rep_socket is None + ): + raise ChannelSetupError("ZMQ connector is not initialized") + sender_req_lock = self._sender_req_lock + sender_rep_socket = self._sender_rep_socket + receiver_req_lock = self._receiver_req_lock + receiver_rep_socket = self._receiver_rep_socket + async def _handle_sender_requests() -> None: - async with self._sender_req_lock: - sender_request = await self._sender_rep_socket.recv_json() + async with sender_req_lock: + sender_request = await sender_rep_socket.recv_json() if (sender_addr := sender_request.get("sender_address")) is None: - await self._sender_rep_socket.send_json({"success": False}) + await sender_rep_socket.send_json({"success": False}) else: self._sender_addr = sender_addr - await self._sender_rep_socket.send_json({"success": True}) + await sender_rep_socket.send_json({"success": True}) while True: - await self._sender_rep_socket.recv_json() - await self._sender_rep_socket.send_json({"success": False}) + await sender_rep_socket.recv_json() + await sender_rep_socket.send_json({"success": False}) async def _handle_receiver_requests() -> None: while self._sender_addr is None: await asyncio.sleep(0.5) while True: - async with self._receiver_req_lock: - await self._receiver_rep_socket.recv() - await self._receiver_rep_socket.send(self._sender_addr.encode()) + async with receiver_req_lock: + await receiver_rep_socket.recv() + await receiver_rep_socket.send(self._sender_addr.encode()) async with asyncio.TaskGroup() as tg: tg.create_task(_handle_sender_requests()) @@ -190,8 +213,11 @@ async def _handle_receiver_requests() -> None: async def connect_send(self) -> ZMQChannel: """Returns a `ZMQChannel` for sending messages.""" + await self.init() if self._send_channel is not None: return self._send_channel + if self._sender_rep_socket_addr is None: + raise ChannelSetupError("ZMQ connector is not initialized") send_socket = create_socket(zmq.PUSH, [(zmq.SNDHWM, self._maxsize)]) send_port = send_socket.bind_to_random_port("tcp://*") send_addr = f"{self._zmq_address}:{send_port}" @@ -210,8 +236,11 @@ async def connect_send(self) -> ZMQChannel: async def connect_recv(self) -> ZMQChannel: """Returns a `ZMQChannel` for receiving messages.""" + await self.init() if self._recv_channel is not None: return self._recv_channel + if self._receiver_rep_socket_addr is None: + raise ChannelSetupError("ZMQ connector is not initialized") recv_socket = create_socket(zmq.PULL, [(zmq.RCVHWM, self._maxsize)]) receiver_req_socket = create_socket(zmq.REQ, []) @@ -232,16 +261,28 @@ class _ZMQPubsubConnector(_ZMQConnector): def __init__(self, *args: _t.Any, **kwargs: _t.Any) -> None: super().__init__(*args, **kwargs) self._topic = str(self.spec.source) - self._xsub_socket = create_socket(zmq.XSUB, [(zmq.RCVHWM, self._maxsize)]) - self._xsub_port = self._xsub_socket.bind_to_random_port("tcp://*") - self._xpub_socket = create_socket(zmq.XPUB, [(zmq.SNDHWM, self._maxsize)]) - self._xpub_port = self._xpub_socket.bind_to_random_port("tcp://*") - self._poller = zmq_asyncio.Poller() - self._poller.register(self._xsub_socket, zmq.POLLIN) - self._poller.register(self._xpub_socket, zmq.POLLIN) - self._poll_task = asyncio.create_task(self._poll()) - _zmq_proxy_tasks.add(self._poll_task) - self._poll_task.add_done_callback(_zmq_proxy_tasks.discard) + self._xsub_port: _t.Optional[int] = None + self._xpub_port: _t.Optional[int] = None + self._poller: _t.Optional[zmq_asyncio.Poller] = None + self._poll_task: _t.Optional[asyncio.Task[None]] = None + self._xsub_socket: _t.Optional[zmq_asyncio.Socket] = None + self._xpub_socket: _t.Optional[zmq_asyncio.Socket] = None + + async def init(self) -> None: + """Allocate proxy sockets when execution starts.""" + async with self._init_lock: + if self._xsub_port is not None: + return + self._xsub_socket = create_socket(zmq.XSUB, [(zmq.RCVHWM, self._maxsize)]) + self._xsub_port = self._xsub_socket.bind_to_random_port("tcp://*") + self._xpub_socket = create_socket(zmq.XPUB, [(zmq.SNDHWM, self._maxsize)]) + self._xpub_port = self._xpub_socket.bind_to_random_port("tcp://*") + self._poller = zmq_asyncio.Poller() + self._poller.register(self._xsub_socket, zmq.POLLIN) + self._poller.register(self._xpub_socket, zmq.POLLIN) + self._poll_task = asyncio.create_task(self._poll()) + _zmq_proxy_tasks.add(self._poll_task) + self._poll_task.add_done_callback(_zmq_proxy_tasks.discard) def __getstate__(self) -> dict: state = self.__dict__.copy() @@ -252,6 +293,8 @@ def __getstate__(self) -> dict: return state async def _poll(self) -> None: + if self._poller is None or self._xpub_socket is None or self._xsub_socket is None: + raise ChannelSetupError("ZMQ connector is not initialized") poll_fn, xps, xss = self._poller.poll, self._xpub_socket, self._xsub_socket try: while True: @@ -266,6 +309,9 @@ async def _poll(self) -> None: async def connect_send(self) -> ZMQChannel: """Returns a `ZMQChannel` for sending pubsub messages.""" + await self.init() + if self._xsub_port is None: + raise ChannelSetupError("ZMQ connector is not initialized") send_socket = create_socket(zmq.PUB, [(zmq.SNDHWM, self._maxsize)]) send_socket.connect(f"{self._zmq_address}:{self._xsub_port}") await asyncio.sleep(0.1) # Ensure connections established before first send. Better way? @@ -273,6 +319,9 @@ async def connect_send(self) -> ZMQChannel: async def connect_recv(self) -> ZMQChannel: """Returns a `ZMQChannel` for receiving pubsub messages.""" + await self.init() + if self._xpub_port is None: + raise ChannelSetupError("ZMQ connector is not initialized") socket_opts: zmq_sockopts_t = [ (zmq.RCVHWM, self._maxsize), (zmq.SUBSCRIBE, self._topic.encode("utf8")), @@ -286,17 +335,24 @@ async def connect_recv(self) -> ZMQChannel: class _ZMQPubsubConnectorProxy(_ZMQConnector): """`_ZMQPubsubConnectorProxy` acts is a python asyncio based proxy for `ZMQChannel` messages.""" - @inject - def __init__( - self, *args: _t.Any, zmq_proxy: ZMQProxy = Provide[DI.zmq_proxy], **kwargs: _t.Any - ) -> None: + def __init__(self, *args: _t.Any, **kwargs: _t.Any) -> None: super().__init__(*args, **kwargs) self._topic = str(self.spec.source) - self._zmq_proxy = zmq_proxy + self._zmq_proxy: _t.Optional[ZMQProxy] = None self._send_channel: _t.Optional[ZMQChannel] = None self._recv_channel: _t.Optional[ZMQChannel] = None + @inject + async def _resolve_proxy(self, zmq_proxy: ZMQProxy = Provide[DI.zmq_proxy]) -> ZMQProxy: + return zmq_proxy + + async def init(self) -> None: + """Resolve the shared proxy when execution starts.""" + async with self._init_lock: + if self._zmq_proxy is None: + self._zmq_proxy = await self._resolve_proxy() + def __getstate__(self) -> dict: state = self.__dict__.copy() for attr in ("_send_channel", "_recv_channel"): @@ -311,8 +367,11 @@ def __setstate__(self, state: dict) -> None: async def connect_send(self) -> ZMQChannel: """Returns a `ZMQChannel` for sending pubsub messages.""" + await self.init() if self._send_channel is not None: return self._send_channel + if self._zmq_proxy is None: + raise ChannelSetupError("ZMQ connector is not initialized") send_socket = create_socket(zmq.PUB, [(zmq.SNDHWM, self._maxsize)]) send_socket.connect(self._zmq_proxy.xsub_addr) self._send_channel = ZMQChannel( @@ -323,6 +382,9 @@ async def connect_send(self) -> ZMQChannel: async def connect_recv(self) -> ZMQChannel: """Returns a `ZMQChannel` for receiving pubsub messages.""" + await self.init() + if self._zmq_proxy is None: + raise ChannelSetupError("ZMQ connector is not initialized") socket_opts: zmq_sockopts_t = [ (zmq.RCVHWM, self._maxsize), (zmq.SUBSCRIBE, self._topic.encode("utf8")), @@ -347,8 +409,11 @@ def __init__(self, *args: _t.Any, **kwargs: _t.Any) -> None: async def connect_recv(self) -> ZMQChannel: """Returns a `ZMQChannel` for receiving messages.""" + await self.init() if self._recv_channel is not None: return self._recv_channel + if self._zmq_proxy is None: + raise ChannelSetupError("ZMQ connector is not initialized") self._push_address = await self._zmq_proxy.add_push_socket( self._topic, maxsize=self._maxsize ) @@ -384,6 +449,10 @@ def __init__( raise ValueError(f"Unsupported connector mode: {self.spec.mode}") self._zmq_conn_impl: _ZMQConnector = zmq_conn_cls(*args, **kwargs) + async def init(self) -> None: + """Allocate resources for the selected ZMQ implementation.""" + await self._zmq_conn_impl.init() + @property def zmq_address(self) -> str: """The ZMQ address used for communication.""" diff --git a/plugboard/library/file_io.py b/plugboard/library/file_io.py index 6d313279..61990a34 100644 --- a/plugboard/library/file_io.py +++ b/plugboard/library/file_io.py @@ -1,5 +1,6 @@ """Provides `FileReader` and `FileWriter` components to access files from Plugboard models.""" +import asyncio from collections import deque from pathlib import Path import typing as _t @@ -106,7 +107,12 @@ def __init__( raise ValueError("Only CSV files support chunked writing.") self._storage_options = storage_options or {} self._header_written = False - self._check_file() + + async def init(self) -> None: + """Open and truncate the destination when execution starts.""" + await asyncio.to_thread(self._check_file) + self._header_written = False + await super().init() def _check_file(self) -> None: with fsspec.open(self._file_path, mode="w", **self._storage_options): diff --git a/plugboard/process/local_process.py b/plugboard/process/local_process.py index c673eb44..f636f585 100644 --- a/plugboard/process/local_process.py +++ b/plugboard/process/local_process.py @@ -57,6 +57,10 @@ async def _connect_state(self) -> None: async def init(self) -> None: """Performs component initialisation actions.""" + self.validate() + async with asyncio.TaskGroup() as tg: + for connector in self.connectors.values(): + tg.create_task(connector.init()) async with asyncio.TaskGroup() as tg: await self.connect_state() await self._connect_components() diff --git a/plugboard/process/process.py b/plugboard/process/process.py index 939ae94a..0f570358 100644 --- a/plugboard/process/process.py +++ b/plugboard/process/process.py @@ -109,12 +109,21 @@ async def _set_status(self, status: Status, publish: bool = True) -> None: @abstractmethod async def init(self) -> None: """Performs component initialisation actions.""" + self.validate() + self._is_initialised = True + await self._set_status(Status.INIT) + + def validate(self) -> None: + """Validate the process topology without acquiring external resources.""" + for component in self.components.values(): + if not hasattr(component, "_state_is_connected"): + raise ValidationError( + "Component invalid: did you forget to call super().__init__ in the constructor?" + ) errors = validate_process(self.dict()) if errors: msg = "Process validation failed:\n" + "\n".join(errors) raise ValidationError(msg) - self._is_initialised = True - await self._set_status(Status.INIT) @abstractmethod async def step(self) -> None: diff --git a/plugboard/process/ray_process.py b/plugboard/process/ray_process.py index 81a3e28f..a607b009 100644 --- a/plugboard/process/ray_process.py +++ b/plugboard/process/ray_process.py @@ -43,11 +43,7 @@ def __init__( """ # TODO: Replace with a namespace based on the job ID or similar self._namespace = f"plugboard-{gen_rand_str(16)}" - self._component_actors = { - # Recreate components on remote actors - c.id: self._create_component_actor(c) - for c in components - } + self._component_actors: dict[str, _t.Any] = {} self._tasks: dict[str, ray.ObjectRef] = {} super().__init__( @@ -113,6 +109,13 @@ async def _connect_state(self) -> None: async def init(self) -> None: """Performs component initialisation actions.""" + self.validate() + self._component_actors = { + # Recreate components on remote actors only when execution starts. + component.id: self._create_component_actor(component) + for component in self.components.values() + } + await asyncio.gather(*(connector.init() for connector in self.connectors.values())) await self.connect_state() await self._connect_components() coros = [component.init.remote() for component in self._component_actors.values()] diff --git a/plugboard/state/ray_state_backend.py b/plugboard/state/ray_state_backend.py index 010ec02a..07390f7b 100644 --- a/plugboard/state/ray_state_backend.py +++ b/plugboard/state/ray_state_backend.py @@ -67,8 +67,14 @@ def __init__( super().__init__(*args, **kwargs) default_options = {"num_cpus": 0} actor_options = actor_options or {} - actor_options = {**default_options, **actor_options} - self._actor = ray.remote(**actor_options)(_DictionaryActor).remote() + self._actor_options = {**default_options, **actor_options} + self._actor: _t.Any = None + + async def init(self) -> None: + """Create the state actor when process execution starts.""" + if self._actor is None: + self._actor = ray.remote(**self._actor_options)(_DictionaryActor).remote() + await super().init() @property def _state(self) -> dict[str, _t.Any]: diff --git a/tests/integration/test_process_validation.py b/tests/integration/test_process_validation.py index 8eaf9d05..8e3e3af0 100644 --- a/tests/integration/test_process_validation.py +++ b/tests/integration/test_process_validation.py @@ -77,7 +77,5 @@ async def step(self) -> None: ], ) - with pytest.raises(ExceptionGroup) as exc_info: + with pytest.raises(exceptions.ValidationError, match="forget to call super"): await process.init() - - assert exc_info.group_contains(exceptions.ValidationError), "No ValidationError raised" diff --git a/tests/unit/test_channel.py b/tests/unit/test_channel.py index 810a20be..408ad91c 100644 --- a/tests/unit/test_channel.py +++ b/tests/unit/test_channel.py @@ -50,6 +50,7 @@ async def test_channel(connector_cls: type[Connector], ray_ctx: None, job_id_ctx """Tests the various Channel implementations.""" spec = ConnectorSpec(mode=ConnectorMode.PIPELINE, source="test.send", target="test.recv") connector = ConnectorBuilder(connector_cls=connector_cls).build(spec) + await connector.init() send_channel, recv_channel = await asyncio.gather( connector.connect_send(), connector.connect_recv() @@ -94,6 +95,7 @@ async def test_multiprocessing_channel( """Tests the various Channel implementations in a multiprocess environment.""" spec = ConnectorSpec(mode=ConnectorMode.PIPELINE, source="test.send", target="test.recv") connector = ConnectorBuilder(connector_cls=connector_cls_mp).build(spec) + await connector.init() container_ctx = container_context( DI, global_context={"job_id": job_id_ctx}, scope=ContextScopes.APP diff --git a/tests/unit/test_ray_channel.py b/tests/unit/test_ray_channel.py new file mode 100644 index 00000000..b9b6668c --- /dev/null +++ b/tests/unit/test_ray_channel.py @@ -0,0 +1,66 @@ +"""Regression tests for Ray channel and connector initialization.""" + +import asyncio +from unittest.mock import AsyncMock, patch + +import pytest + +from plugboard.connector import RayChannel, RayConnector +from plugboard.schemas import ConnectorSpec + + +@pytest.mark.asyncio +async def test_direct_channel_operations_need_no_initialization() -> None: + """A directly constructed channel can send, receive, and close immediately.""" + with patch("plugboard.connector.ray_channel.ray.remote") as remote: + actor_class = remote.return_value.return_value + actor = actor_class.remote.return_value + actor.send.remote = AsyncMock() + actor.recv.remote = AsyncMock(return_value="message") + actor.close.remote = AsyncMock() + + channel = RayChannel(actor_options={"num_cpus": 1}, maxsize=3) + await channel.send("message") + assert await channel.recv() == "message" + await channel.close() + + remote.assert_called_once_with(num_cpus=1) + actor_class.remote.assert_called_once_with(maxsize=3) + actor.send.remote.assert_awaited_once_with("message") + actor.recv.remote.assert_awaited_once_with() + actor.close.remote.assert_awaited_once_with() + + +@pytest.mark.parametrize("property_name", ["maxsize", "is_closed"]) +def test_direct_channel_properties_need_no_initialization(property_name: str) -> None: + """Property access preserves the remote reference returned by the actor.""" + with patch("plugboard.connector.ray_channel.ray.remote") as remote: + actor = remote.return_value.return_value.remote.return_value + channel = RayChannel() + + assert getattr(channel, property_name) is actor.getattr.remote.return_value + + remote.assert_called_once_with(num_cpus=0) + actor.getattr.remote.assert_called_once_with(property_name) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("initialize_first", [False, True]) +async def test_connector_defers_and_shares_channel(initialize_first: bool) -> None: + """Explicit init and concurrent connections create one shared actor on demand.""" + with patch("plugboard.connector.ray_channel.ray.remote") as remote: + connector = RayConnector(spec=ConnectorSpec(source="source.value", target="sink.value")) + remote.assert_not_called() + + if initialize_first: + await connector.init() + + sender, receiver = await asyncio.gather(connector.connect_send(), connector.connect_recv()) + await connector.init() + + assert isinstance(sender, RayChannel) + assert sender is receiver + assert await connector.connect_send() is sender + assert await connector.connect_recv() is receiver + remote.assert_called_once_with(num_cpus=0) + remote.return_value.return_value.remote.assert_called_once_with() diff --git a/tests/unit/test_side_effect_free_construction.py b/tests/unit/test_side_effect_free_construction.py new file mode 100644 index 00000000..3573dae3 --- /dev/null +++ b/tests/unit/test_side_effect_free_construction.py @@ -0,0 +1,173 @@ +"""Tests that process inspection does not acquire external resources.""" + +from pathlib import Path +from unittest.mock import AsyncMock, patch + +import msgspec +import pytest +from typer.testing import CliRunner + +from plugboard.cli import app +from plugboard.component import Component, IOController as IO +from plugboard.connector import AsyncioConnector, RayConnector, ZMQConnector +from plugboard.diagram import MermaidDiagram +from plugboard.exceptions import ValidationError +from plugboard.library import FileWriter +from plugboard.process import LocalProcess, RayProcess +from plugboard.schemas import ConfigSpec, ConnectorMode, ConnectorSpec +from plugboard.state import RayStateBackend +from plugboard.utils import DI, Settings + + +class Source(Component): + """A source used to describe a valid topology.""" + + io = IO(outputs=["value"]) + + async def step(self) -> None: + """Produce no values; only topology metadata is used by these tests.""" + pass + + +class Sink(Component): + """A sink used to describe a valid topology.""" + + io = IO(inputs=["value"]) + + async def step(self) -> None: + """Consume no values; only topology metadata is used by these tests.""" + pass + + +def test_file_writer_inspection_does_not_touch_destination(tmp_path: Path) -> None: + """Construction, validation, diagramming, and export leave output files untouched.""" + output_path = tmp_path / "output.csv" + output_path.write_text("existing data\n") + writer = FileWriter(name="writer", path=str(output_path), field_names=["value"]) + process = LocalProcess( + components=[Source(name="source"), writer], + connectors=[ + AsyncioConnector(spec=ConnectorSpec(source="source.value", target="writer.value")) + ], + ) + + process.validate() + MermaidDiagram.from_process(process) + process.export() + config_path = tmp_path / "process.yaml" + process.dump(config_path) + + runner = CliRunner() + for command in ("validate", "diagram"): + result = runner.invoke(app, ["process", command, str(config_path)]) + assert result.exit_code == 0 + assert output_path.read_text() == "existing data\n" + + assert output_path.read_text() == "existing data\n" + + +@pytest.mark.asyncio +async def test_invalid_process_fails_before_initializing_writer_or_connector( + tmp_path: Path, +) -> None: + """Invalid topology cannot truncate files or initialize connectors.""" + output_path = tmp_path / "output.csv" + output_path.write_text("existing data\n") + writer = FileWriter(name="writer", path=str(output_path), field_names=["value"]) + connector = AsyncioConnector(spec=ConnectorSpec(source="source.value", target="sink.value")) + connector.init = AsyncMock() + process = LocalProcess( + components=[Source(name="source"), Sink(name="sink"), writer], + connectors=[connector], + ) + + with pytest.raises(ValidationError): + await process.init() + + connector.init.assert_not_awaited() + assert output_path.read_text() == "existing data\n" + + +def test_ray_process_inspection_does_not_create_actors(tmp_path: Path) -> None: + """Ray process metadata can be inspected without creating any actors.""" + with ( + patch("plugboard.process.ray_process.ray.remote") as component_remote, + patch("plugboard.connector.ray_channel.ray.remote") as channel_remote, + patch("plugboard.state.ray_state_backend.ray.remote") as state_remote, + ): + process = RayProcess( + components=[Source(name="source"), Sink(name="sink")], + connectors=[ + RayConnector(spec=ConnectorSpec(source="source.value", target="sink.value")) + ], + state=RayStateBackend(), + ) + + process.validate() + MermaidDiagram.from_process(process) + exported = process.export() + exported["connector_builder"] = {"type": "plugboard.connector.RayConnector"} + config_path = tmp_path / "ray-process.yaml" + config = ConfigSpec.model_validate({"plugboard": {"process": exported}}) + config_path.write_bytes(msgspec.yaml.encode(config.model_dump())) + + runner = CliRunner() + for command in ("validate", "diagram"): + result = runner.invoke(app, ["process", command, str(config_path)]) + assert result.exit_code == 0 + + component_remote.assert_not_called() + channel_remote.assert_not_called() + state_remote.assert_not_called() + + +@pytest.mark.parametrize("use_proxy", [False, True]) +def test_zmq_process_inspection_does_not_allocate_resources(use_proxy: bool) -> None: + """ZMQ sockets and proxy processes are deferred until initialization.""" + settings = Settings.model_validate({"flags": {"zmq_pubsub_proxy": use_proxy}}) + with ( + DI.override_providers_sync({"settings": settings}), + patch("plugboard.connector.zmq_channel.create_socket") as create_socket, + patch("plugboard.utils.di.ZMQProxy") as proxy, + ): + process = LocalProcess( + components=[Source(name="source"), Sink(name="sink")], + connectors=[ + ZMQConnector( + spec=ConnectorSpec( + source="source.value", + target="sink.value", + mode=ConnectorMode.PUBSUB, + ) + ) + ], + ) + + process.validate() + MermaidDiagram.from_process(process) + process.export() + + create_socket.assert_not_called() + proxy.assert_not_called() + + +@pytest.mark.asyncio +async def test_invalid_ray_process_fails_before_creating_actors() -> None: + """Invalid topology is rejected before process, channel, or state actors start.""" + with ( + patch("plugboard.process.ray_process.ray.remote") as component_remote, + patch("plugboard.connector.ray_channel.ray.remote") as channel_remote, + patch("plugboard.state.ray_state_backend.ray.remote") as state_remote, + ): + process = RayProcess( + components=[Sink(name="sink")], + connectors=[], + state=RayStateBackend(), + ) + + with pytest.raises(ValidationError, match="unconnected inputs"): + await process.init() + + component_remote.assert_not_called() + channel_remote.assert_not_called() + state_remote.assert_not_called()