From e1251c26399b758b03f2ad0953fbdae577c56a7c Mon Sep 17 00:00:00 2001 From: Toby Coleman Date: Wed, 4 Nov 2026 21:43:35 +0000 Subject: [PATCH] feat: add optional types and validation for component ports --- docs/usage/key-concepts.md | 122 ++++++++++ plugboard/component/component.py | 59 ++++- plugboard/component/io_controller.py | 70 +++++- plugboard/component/port_types.py | 179 ++++++++++++++ plugboard/process/process.py | 25 ++ tests/integration/test_typed_ports.py | 181 ++++++++++++++ tests/unit/test_port_types.py | 330 ++++++++++++++++++++++++++ 7 files changed, 949 insertions(+), 17 deletions(-) create mode 100644 plugboard/component/port_types.py create mode 100644 tests/integration/test_typed_ports.py create mode 100644 tests/unit/test_port_types.py diff --git a/docs/usage/key-concepts.md b/docs/usage/key-concepts.md index 06980fdc..3730b88c 100644 --- a/docs/usage/key-concepts.md +++ b/docs/usage/key-concepts.md @@ -21,6 +21,128 @@ When implementing your own components, you will need to: * Optionally define an `init()` method to do any required preparatory steps before the model in run. * In the case of event based models, define custom [`Event`][plugboard.events.Event] subclasses and corresponding event handler methods decorated with `Event.handler`. +#### Optional input and output types + +Declare port types by passing a mapping of names to Python types to `IOController`: + +```python +from plugboard.component import Component, IOController as IO + + +class Double(Component): + io = IO(inputs={"value": int}, outputs={"result": int}) + + async def step(self) -> None: + self.result = self.value * 2 +``` + +Existing lists of names still work: `IO(inputs=["value"], outputs=["result"])` +declares both ports as `typing.Any`. You can mix a list on one side with a typed +mapping on the other, or use `Any` explicitly for individual ports. Attribute +annotations alone do not declare port types or directions. Mapping declarations +provide runtime metadata; add ordinary attribute annotations separately if you +want your editor or static type checker to check `self.value` and `self.result`. + +Supported types are `Any`, Python classes, unions such as `int | None`, and +`list[T]`, `dict[K, V]`, `set[T]`, `frozenset[T]`, fixed-length `tuple[T, U]`, and +variable-length `tuple[T, ...]`. These can be nested. `None` in a declaration +means `type(None)`. Other typing constructs, including string forward references, +`TypeVar`, `Literal`, `Annotated`, `TypedDict`, and protocols, raise `IOSetupError`. +Use an ordinary class for a nominal check of an array, dataframe, or other custom +payload; its internal contents, shape, and dtype are not inspected. + +Ports inherited from a base component retain their types. A name-only declaration +does not erase an inherited type. Adding a type to an inherited `Any` port is +allowed; conflicting explicit types raise `IOSetupError`, including conflicts +between multiple base classes. + +The same syntax works for ports chosen at instance construction. For example, +this component creates its output names from a configuration list: + +```python +from plugboard.schemas import ComponentArgsDict +from typing import Unpack + + +class DynamicSource(Component): + # A class declaration is still needed for the component lifecycle. + io = IO(outputs=["ready"]) + + def __init__(self, fields: list[str], **kwargs: Unpack[ComponentArgsDict]) -> None: + super().__init__(**kwargs) + self.io = IO( + outputs={"ready": bool, **{name: int for name in fields}}, + input_events=self.io.input_events, + output_events=self.io.output_events, + namespace=self.name, + component=self, + runtime_validation="deep", + validate_outputs=True, + ) + self._fields = fields + + async def step(self) -> None: + self.ready = True + for field in self._fields: + setattr(self, field, 42) +``` + +Keep constructor arguments serializable when using YAML configuration or JSON +state backends. The example accepts a list of names and constructs the Python +types inside the component. Type declarations in component code also work when +the component is selected through YAML; there is no YAML type-expression syntax. + +#### Type validation + +Process initialization checks declared output-to-input compatibility, even when +runtime checking is disabled. For example, `Dog` output can connect to `Animal` +input when `Dog` subclasses `Animal`; the reverse fails with `ValidationError`. +All possible members of an output union must be accepted by the input. Mutable +containers (`list`, `dict`, `set`) are invariant, so `list[Dog]` cannot connect to +`list[Animal]`; `tuple` and `frozenset` allow narrower element types. Each edge of +a fan-in or fan-out is checked. `Any` and unknown element types of bare containers +allow a connection without proving its compatibility. + +Runtime value checks are **off by default**. Enable them per controller: + +```python +io = IO( + inputs={"samples": list[float], "label": str | None}, + outputs={"average": float}, + runtime_validation="deep", + validate_outputs=True, +) +``` + +| Option | Behavior | +| --- | --- | +| `runtime_validation="off"` | No runtime value checks; initialization still checks declarations. | +| `runtime_validation="shallow"` | Check each supplied input's outer type, without checking container contents. | +| `runtime_validation="deep"` | Also check supported container contents and tuple lengths. | +| `validate_outputs=True` | Apply the selected runtime mode to outputs before buffering and sending. Defaults to `False`. | + +Runtime validation uses `isinstance` semantics: subclasses are accepted, including +`bool` for `int`. There is no numeric widening or coercion: an `int` is not a +`float`, and a string containing digits is not an `int`. Union members follow the +same rules. Values are checked in place, without converting or copying them. + +Input checks cover channel messages, initial values, and direct assignments +(including event handlers) when inputs are bound for a step. A missing input is +not checked, while an explicitly supplied `None` must match its declared type. +Output checks run once per port before fan-out. Failures raise `ValidationError` +with the component, port, expected and actual types, and a nested path where +available, and mark the component as failed. Union failures report the union at +the point where no alternative matches. Process execution may wrap failures in an +`ExceptionGroup` or a Ray task error. + +Validators are built when the controller is created and reused for every value; +untyped ports have no validators. Deep checks visit container contents on every +message, so their cost grows with payload size. Use shallow checks for large +payloads when checking only their outer class is sufficient. Validation policy +belongs to each `IOController` declaration: a subclass declaring a new controller +should specify its desired policy, and dynamically replaced controllers should +do the same. Class controllers pass their policy to component instances. + ### Connectors Data flows between components via **Connectors**. You will need to use these to tell Plugboard how data should flow between your components. Your overall model might be very simple, for example: diff --git a/plugboard/component/component.py b/plugboard/component/component.py index af51a2eb..3835dff0 100644 --- a/plugboard/component/component.py +++ b/plugboard/component/component.py @@ -90,14 +90,22 @@ def __init__( except IOSetupError: pass self.io = IO( - inputs=self.__class__.io.inputs, - outputs=self.__class__.io.outputs, + inputs={ + name: self.__class__.io.input_types.get(name, _t.Any) + for name in self.__class__.io.inputs + }, + outputs={ + name: self.__class__.io.output_types.get(name, _t.Any) + for name in self.__class__.io.outputs + }, initial_values=self._initial_values, input_events=self.__class__.io.input_events, output_events=self.__class__.io.output_events, event_field_coverage=self.__class__.io.event_field_coverage, namespace=self.name, component=self, + runtime_validation=self.__class__.io.runtime_validation, + validate_outputs=self.__class__.io.validate_outputs, ) self.resources = resources or Resource( cpu=self.__class__.resources.cpu, @@ -174,13 +182,36 @@ def _configure_io(cls) -> None: # Get all parent classes that are Component subclasses io_args, exports = cls._get_aggregated_io_args() event_field_coverage = cls._get_event_field_coverage() + port_types: dict[str, dict[str, _t.Any]] = {"inputs": {}, "outputs": {}} + for base in reversed(cls.__mro__): + base_io = base.__dict__.get("io") + if not isinstance(base_io, IO): + continue + for direction, types in port_types.items(): + for field, hint in getattr(base_io, f"{direction[:-1]}_types").items(): + previous = types.get(field, _t.Any) + if previous is not _t.Any and hint is not _t.Any and previous != hint: + raise IOSetupError( + f"Conflicting inherited types for {cls.__name__}.{field}: " + f"{previous!r} and {hint!r}" + ) + if hint is not _t.Any or field not in types: + types[field] = hint + runtime_validation = cls.io.runtime_validation + validate_outputs = cls.io.validate_outputs # Set io arguments for subclass cls.io = IO( - inputs=sorted(io_args["inputs"], key=str), - outputs=sorted(io_args["outputs"], key=str), + inputs={ + name: port_types["inputs"].get(name, _t.Any) for name in sorted(io_args["inputs"]) + }, + outputs={ + name: port_types["outputs"].get(name, _t.Any) for name in sorted(io_args["outputs"]) + }, input_events=_t.cast(list[_t.Type[Event]], sorted(io_args["input_events"], key=str)), output_events=_t.cast(list[_t.Type[Event]], sorted(io_args["output_events"], key=str)), event_field_coverage=event_field_coverage, + runtime_validation=runtime_validation, + validate_outputs=validate_outputs, ) # Set exports for subclass cls.exports = sorted(set(exports)) @@ -364,15 +395,15 @@ async def _wrapper() -> None: await self._set_status(Status.RUNNING, publish=not self._is_running) await self._io_read_with_status_check() await self._handle_events() - self._bind_inputs() - if self._can_step: - try: + try: + self._bind_inputs() + if self._can_step: await self._step() - except Exception as e: - await self._set_status(Status.FAILED) - self._logger.exception("Component step failed") - raise e - self._bind_outputs() + self._bind_outputs() + except Exception: + await self._set_status(Status.FAILED) + self._logger.exception("Component step failed") + raise await self.io.write() self._reset_input_trackers() await self._set_status(Status.WAITING, publish=not self._is_running) @@ -473,6 +504,8 @@ def _bind_inputs(self) -> None: # Consume input data directly assigned and read from channels and reset to empty values received_inputs = dict(self.io.buf_fields[_io_key_in].flush()) self._field_inputs.update(received_inputs) + if self.io._input_validators: + self.io.validate_inputs(self._field_inputs) # Check if all input fields have been set self._field_inputs_ready = all(k in self._field_inputs for k in self.io.inputs) for field in self.io.inputs: @@ -489,6 +522,8 @@ def _bind_outputs(self) -> None: if not self._can_step: return output_data = {field: getattr(self, field, None) for field in self.io.outputs} + if self.io._output_validators: + self.io.validate_output_values(output_data) self.io.buf_fields[_io_key_out].put(output_data.items()) async def _handle_events(self) -> None: diff --git a/plugboard/component/io_controller.py b/plugboard/component/io_controller.py index 9fa6076a..a9920455 100644 --- a/plugboard/component/io_controller.py +++ b/plugboard/component/io_controller.py @@ -7,6 +7,13 @@ from functools import cache, cached_property import typing as _t +from plugboard.component.port_types import ( + PortDeclaration, + RuntimeValidation, + compile_validator, + normalize_ports, + type_name, +) from plugboard.connector import AsyncioChannel, Channel, Connector from plugboard.events import Event, StopEvent from plugboard.exceptions import ChannelClosedError, IOStreamClosedError @@ -29,22 +36,44 @@ class IOController: - """`IOController` manages input/output to/from components.""" + """`IOController` manages input/output to/from components. + + Args: + inputs: Port names (implicitly Any), or a mapping of names to Python types. + outputs: Port names (implicitly Any), or a mapping of names to Python types. + runtime_validation: Input value checking: off (default), shallow (outer type), + or deep (including built-in container contents). Values are never coerced. + validate_outputs: Also check outputs before buffering, using runtime_validation. + Defaults to False. Each output is checked once before fan-out. + """ def __init__( self, - inputs: _t.Optional[_t.Any] = None, - outputs: _t.Optional[_t.Any] = None, + inputs: PortDeclaration | None = None, + outputs: PortDeclaration | None = None, initial_values: _t.Optional[dict[str, _t.Iterable]] = None, input_events: _t.Optional[list[_t.Type[Event]]] = None, output_events: _t.Optional[list[_t.Type[Event]]] = None, event_field_coverage: _t.Optional[dict[str, list[str]]] = None, namespace: str = IO_NS_UNSET, component: _t.Optional[Component] = None, + *, + runtime_validation: RuntimeValidation = "off", + validate_outputs: bool = False, ) -> None: self.namespace = namespace - self.inputs = inputs or [] - self.outputs = outputs or [] + self.input_types = normalize_ports(inputs) + self.output_types = normalize_ports(outputs) + self.inputs = list(self.input_types) + self.outputs = list(self.output_types) + if runtime_validation not in ("off", "shallow", "deep"): + raise ValueError(f"Unknown runtime validation mode: {runtime_validation!r}") + self.runtime_validation = runtime_validation + self.validate_outputs = validate_outputs + self._input_validators = self._compile_validators(self.input_types) + self._output_validators = ( + self._compile_validators(self.output_types) if validate_outputs else {} + ) self.initial_values = initial_values or {} self.input_events = input_events or [] self.output_events = output_events or [] @@ -87,6 +116,28 @@ def __init__( self._received_events_lock = asyncio.Lock() self._has_received_events = asyncio.Event() + def _compile_validators( + self, types: dict[str, _t.Any] + ) -> dict[str, _t.Callable[[_t.Any, str], None]]: + if self.runtime_validation == "off": + return {} + return { + field: compile_validator(hint, deep=self.runtime_validation == "deep") + for field, hint in types.items() + if hint is not _t.Any + } + + def validate_inputs(self, values: dict[str, _t.Any]) -> None: + """Check supplied inputs, including event assignments and initial values.""" + for field, validate in self._input_validators.items(): + if field in values: + validate(values[field], f"{self.namespace}.{field} (input)") + + def validate_output_values(self, values: dict[str, _t.Any]) -> None: + """Check each typed output once before buffering and fan-out.""" + for field, validate in self._output_validators.items(): + validate(values[field], f"{self.namespace}.{field} (output)") + @property def is_closed(self) -> bool: """Returns `True` if the `IOController` is closed, `False` otherwise.""" @@ -418,6 +469,15 @@ def dict(self) -> dict[str, _t.Any]: # noqa: D102 "output_events": [e.safe_type() for e in self.output_events], "initial_values": {k: list(v) for k, v in self._initial_values.items()}, "event_field_coverage": {k: list(v) for k, v in self.event_field_coverage.items()}, + # Live types/validators stay out of state backend serialization. + **{ + key: {field: type_name(hint) for field, hint in types.items()} + for key, types in ( + ("input_types", self.input_types), + ("output_types", self.output_types), + ) + if any(hint is not _t.Any for hint in types.values()) + }, } diff --git a/plugboard/component/port_types.py b/plugboard/component/port_types.py new file mode 100644 index 00000000..e35f8194 --- /dev/null +++ b/plugboard/component/port_types.py @@ -0,0 +1,179 @@ +"""Type declarations and compiled, non-coercing validators for component ports.""" + +from collections.abc import Callable, Iterable, Mapping +from types import UnionType +import typing as _t + +from plugboard.exceptions import IOSetupError, ValidationError + + +PortDeclaration = Iterable[str] | Mapping[str, _t.Any] +RuntimeValidation = _t.Literal["off", "shallow", "deep"] +_Validator = Callable[[_t.Any, str], None] +_CONTAINERS = {list: 1, dict: 2, set: 1, frozenset: 1, tuple: None} +_UNIONS = (UnionType, _t.Union) + + +def type_name(hint: _t.Any) -> str: + """Return a readable type description suitable for state metadata.""" + hint = type(None) if hint is None else hint + if hint is _t.Any: + return "Any" + if _t.get_origin(hint) is not None: + return str(hint).removeprefix("typing.") + if hint.__module__ == "builtins": + return hint.__qualname__ + return f"{hint.__module__}.{hint.__qualname__}" + + +def normalize_ports(ports: PortDeclaration | None) -> dict[str, _t.Any]: + """Normalize legacy names and typed mappings, rejecting unsupported hints early.""" + result: dict[str, _t.Any] = ( + dict(ports) if isinstance(ports, Mapping) else dict.fromkeys(ports or (), _t.Any) + ) + for name, hint in result.items(): + if not isinstance(name, str): + raise IOSetupError("Port names must be strings") + result[name] = type(None) if hint is None else hint + _check_hint(result[name]) + return result + + +def _check_hint(hint: _t.Any) -> None: + if hint is _t.Any or hint is None: + return + origin, args = _t.get_origin(hint), _t.get_args(hint) + if origin in _UNIONS: + for arg in args: + _check_hint(arg) + return + if origin in _CONTAINERS: + arity = _CONTAINERS[origin] + if arity is not None and args and len(args) != arity: + raise IOSetupError(f"Invalid port type arguments: {hint!r}") + if origin is tuple and len(args) == 2 and args[1] is Ellipsis: + args = args[:1] + for arg in args: + _check_hint(arg) + return + if ( + isinstance(hint, type) + and not _t.is_typeddict(hint) + and not getattr(hint, "_is_protocol", False) + and origin is None + ): + return + raise IOSetupError( + f"Unsupported port type {hint!r}; use Any, a Python class, a union, " + "or list/dict/tuple/set/frozenset types" + ) + + +def is_assignable(source: _t.Any, target: _t.Any) -> bool | None: + """Check declared compatibility; None means unspecified generic information. + + Mutable containers are invariant; tuple and frozenset are covariant. + Any and bare generic types leave compatibility unknown. + """ + source = type(None) if source is None else (tuple if source is _t.Tuple else source) + target = type(None) if target is None else (tuple if target is _t.Tuple else target) + if source is _t.Any or target is _t.Any: + return None + if source == target: + return True + source_origin, target_origin = _t.get_origin(source), _t.get_origin(target) + source_args, target_args = _t.get_args(source), _t.get_args(target) + if source_origin in _UNIONS: + results = [is_assignable(arg, target) for arg in source_args] + return False if False in results else (None if None in results else True) + if target_origin in _UNIONS: + results = [is_assignable(source, arg) for arg in target_args] + return True if True in results else (None if None in results else False) + source_cls, target_cls = source_origin or source, target_origin or target + if not issubclass(source_cls, target_cls): + return False + if not target_args and target_origin is not tuple: + # A bare target does not restrict elements. + return True + if not source_args and source_origin is not tuple: + return None + if source_cls is not target_cls: + # Generic subclass parameters cannot be inferred from their bases here. + return None + if source_cls is tuple: + return _tuple_assignable(source_args, target_args) + if len(source_args) != len(target_args): + return None + results = [] + for src, dst in zip(source_args, target_args): + results.append(is_assignable(src, dst)) + if source_cls in (list, dict, set): + results.append(is_assignable(dst, src)) + return False if False in results else (None if None in results else True) + + +def _tuple_assignable(source: tuple[_t.Any, ...], target: tuple[_t.Any, ...]) -> bool | None: + source_variadic = len(source) == 2 and source[1] is Ellipsis + target_variadic = len(target) == 2 and target[1] is Ellipsis + if target_variadic: + elements = source[:1] if source_variadic else source + results = [is_assignable(arg, target[0]) for arg in elements] + elif source_variadic or len(source) != len(target): + return False + else: + results = [is_assignable(src, dst) for src, dst in zip(source, target)] + return False if False in results else (None if None in results else True) + + +def compile_validator(hint: _t.Any, *, deep: bool) -> _Validator: + """Compile a reusable validator without copying or coercing payloads.""" + hint = type(None) if hint is None else (tuple if hint is _t.Tuple else hint) + if hint is _t.Any: + return lambda value, path: None + origin, args = _t.get_origin(hint), _t.get_args(hint) + description = type_name(hint) + + def fail(value: _t.Any, path: str) -> _t.NoReturn: + raise ValidationError( + f"Invalid value at {path}: expected {description}, got {type_name(type(value))}" + ) + + if origin in _UNIONS: + validators = [compile_validator(arg, deep=deep) for arg in args] + + def validate_union(value: _t.Any, path: str) -> None: + for validate in validators: + try: + validate(value, path) + return + except ValidationError: + pass + fail(value, path) + + return validate_union + + cls = origin or hint + validators = ( + [compile_validator(arg, deep=True) for arg in args if arg is not Ellipsis] if deep else [] + ) + + def validate(value: _t.Any, path: str) -> None: + if not isinstance(value, cls): + fail(value, path) + if not deep or origin is None: + return + if origin is tuple: + variadic = len(args) == 2 and args[1] is Ellipsis + if not variadic and len(value) != len(args): + raise ValidationError(f"Invalid tuple length at {path}: expected {description}") + for index, item in enumerate(value): + validators[0 if variadic else index](item, f"{path}[{index}]") + elif origin is dict and validators: + for index, (key, item) in enumerate(value.items()): + validators[0](key, f"{path}.keys[{index}]") + validators[1](item, f"{path}.values[{index}]") + elif validators: + for index, item in enumerate(value): + validators[0](item, f"{path}[{index}]") + + return validate diff --git a/plugboard/process/process.py b/plugboard/process/process.py index 0f570358..cc78569f 100644 --- a/plugboard/process/process.py +++ b/plugboard/process/process.py @@ -12,6 +12,7 @@ import msgspec from plugboard.component import Component +from plugboard.component.port_types import is_assignable, type_name from plugboard.connector import Connector from plugboard.exceptions import NotInitialisedError, ValidationError from plugboard.schemas import ConfigSpec, Status, validate_process @@ -120,11 +121,35 @@ def validate(self) -> None: raise ValidationError( "Component invalid: did you forget to call super().__init__ in the constructor?" ) + self._validate_io_types() errors = validate_process(self.dict()) if errors: msg = "Process validation failed:\n" + "\n".join(errors) raise ValidationError(msg) + def _validate_io_types(self) -> None: + """Reject incompatible declared field connections without inspecting payloads.""" + errors: list[str] = [] + for connector in self.connectors.values(): + source, target = connector.spec.source, connector.spec.target + source_component = self.components.get(source.entity) + target_component = self.components.get(target.entity) + if source_component is None or target_component is None: + continue # Event connectors do not connect component field ports. + source_types = source_component.io.output_types + target_types = target_component.io.input_types + if source.descriptor not in source_types or target.descriptor not in target_types: + continue # Existing channel setup handles unknown fields. + source_type = source_types[source.descriptor] + target_type = target_types[target.descriptor] + if is_assignable(source_type, target_type) is False: + errors.append( + f"Incompatible port types: {source.id} ({type_name(source_type)}) -> " + f"{target.id} ({type_name(target_type)})" + ) + if errors: + raise ValidationError("Process type validation failed:\n" + "\n".join(errors)) + @abstractmethod async def step(self) -> None: """Executes a single step for the process.""" diff --git a/tests/integration/test_typed_ports.py b/tests/integration/test_typed_ports.py new file mode 100644 index 00000000..fffe7993 --- /dev/null +++ b/tests/integration/test_typed_ports.py @@ -0,0 +1,181 @@ +"""Typed port checks across process initialization and component execution.""" + +import typing as _t + +import msgspec +import pytest + +from plugboard.component import Component, IOController as IO +from plugboard.component.port_types import RuntimeValidation +from plugboard.connector import AsyncioConnector, Connector, RayConnector +from plugboard.events import Event +from plugboard.exceptions import ValidationError +from plugboard.process import LocalProcess, Process, RayProcess +from plugboard.schemas import ComponentArgsDict, ConnectorSpec, Status + + +class Source(Component): + """Producer that can violate its declared output contract.""" + + io = IO(outputs={"value": list[int]}) + + def __init__(self, invalid: bool = False, **kwargs: _t.Unpack[ComponentArgsDict]) -> None: + super().__init__(**kwargs) + self._invalid = invalid + + async def step(self) -> None: + """Emit one payload.""" + self.value = ["wrong"] if self._invalid else [1] + + +class DynamicSink(Component): + """Consumer whose port types and policy are declared at instance construction.""" + + io = IO(inputs=["value"]) + exports = ["received"] + + def __init__( + self, + mode: RuntimeValidation = "off", + incompatible: bool = False, + **kwargs: _t.Unpack[ComponentArgsDict], + ) -> None: + super().__init__(**kwargs) + self.io = IO( + inputs={"value": str if incompatible else list[int]}, + initial_values=self._initial_values, + input_events=self.io.input_events, + output_events=self.io.output_events, + namespace=self.name, + component=self, + runtime_validation=mode, + ) + self.received: list[_t.Any] = [] + + async def step(self) -> None: + """Record values without imposing additional checks.""" + self.received.append(self.value) + + +def _connector(cls: type[Connector] = AsyncioConnector) -> Connector: + return cls(ConnectorSpec(source="source.value", target="sink.value")) + + +async def test_incompatible_connection_fails_before_setup() -> None: + """Known type mismatches fail before initializing components or connections.""" + source, sink = Source(name="source"), DynamicSink(name="sink", incompatible=True) + process = LocalProcess([source, sink], [_connector()]) + with pytest.raises( + ValidationError, match=r"source.value \(list\[int\]\) -> sink.value \(str\)" + ): + await process.init() + assert not process.is_initialised + assert not process._state_is_connected + assert not sink.io._input_channels + + +@pytest.mark.parametrize("mode", ["off", "shallow", "deep"]) +async def test_runtime_channel_values(mode: RuntimeValidation) -> None: + """Every channel payload is checked according to the selected policy.""" + source, sink = Source(name="source"), DynamicSink(name="sink", mode=mode) + process = LocalProcess([source, sink], [_connector()]) + async with process: + assert process.is_initialised + await process.step() + assert sink.received == [[1]] + # A valid first value does not exempt later messages from checking. + source._invalid = True + if mode == "deep": + with pytest.raises(ExceptionGroup) as exc: + await process.step() + assert exc.group_contains(ValidationError, match=r"sink.value \(input\)\[0\]") + assert sink.status == Status.FAILED + assert process.status == Status.FAILED + assert sink.received == [[1]] + else: + await process.step() + assert sink.received == [[1], ["wrong"]] + # Type metadata can be stored by JSON-based state backends. + assert msgspec.json.decode(sink.json())["io"]["input_types"] == {"value": "list[int]"} + + +async def test_initial_values_are_checked() -> None: + """Initial input values reach the same validation boundary as channel data.""" + source = Source(name="source") + sink = DynamicSink(name="sink", mode="deep", initial_values={"value": [["wrong"]]}) + process = LocalProcess([source, sink], [_connector()]) + async with process: + with pytest.raises(ExceptionGroup) as exc: + await process.step() + assert exc.group_contains(ValidationError, match="expected int, got str") + assert sink.received == [] + + +class PayloadEvent(Event): + """An event that supplies an input field.""" + + type: _t.ClassVar[str] = "typed_payload" + value: _t.Any + + +class EventSink(Component): + """Consumer populated through an event handler rather than a field connector.""" + + io = IO(inputs={"value": int}, input_events=[PayloadEvent], runtime_validation="deep") + + @PayloadEvent.handler(populates_fields=["value"]) + async def receive(self, event: PayloadEvent) -> None: + """Assign the event payload to the input.""" + self.value = event.value + + async def step(self) -> None: + """Consume a valid input.""" + + +async def test_event_assigned_inputs_are_checked() -> None: + """Event assignments cannot bypass input validation.""" + sink = EventSink(name="event_sink") + sink.io.buf_events["input"].put([PayloadEvent(source="test", data={}, value="wrong")]) + with pytest.raises(ValidationError, match=r"event_sink.value \(input\).*expected int"): + await sink.step() + assert sink.status == Status.FAILED + + +async def test_output_validation_fails_before_send() -> None: + """Producer-side checking marks failures and prevents invalid output transmission.""" + + class CheckedSource(Source): + io = IO(runtime_validation="deep", validate_outputs=True) + + source = CheckedSource(name="source", invalid=True) + channel = _t.cast(AsyncioConnector, _connector()) + await source.io.connect([channel]) + try: + with pytest.raises(ValidationError, match=r"source.value \(output\)\[0\]"): + await source.step() + assert source.status == Status.FAILED + assert channel._send_channel._queue.empty() + finally: + await source.io.close() + + +@pytest.mark.parametrize("incompatible", [False, True]) +async def test_ray_typed_ports(incompatible: bool, ray_ctx: None) -> None: + """Ray preserves dynamically declared types, policy, and serialized state.""" + source = Source(name="source", invalid=True) + sink = DynamicSink(name="sink", mode="deep", incompatible=incompatible) + process: Process = RayProcess([source, sink], [_connector(RayConnector)]) + try: + if incompatible: + with pytest.raises(ValidationError, match="Incompatible port types"): + await process.init() + else: + await process.init() + # Ray task errors retain the original exception type inside the group. + with pytest.raises(ExceptionGroup) as exc: + await process.step() + assert exc.group_contains(ValidationError, match=r"sink.value \(input\)\[0\]") + assert sink.status == Status.FAILED + assert process.status == Status.FAILED + finally: + await process.destroy() diff --git a/tests/unit/test_port_types.py b/tests/unit/test_port_types.py new file mode 100644 index 00000000..1fc9fcdb --- /dev/null +++ b/tests/unit/test_port_types.py @@ -0,0 +1,330 @@ +"""Tests for optional port types, compatibility, inheritance, and value validation.""" + +from abc import ABC +import typing as _t +from unittest.mock import Mock, patch + +import msgspec +import pytest + +from plugboard.component import Component, IOController as IO, component +from plugboard.component.port_types import compile_validator, is_assignable +from plugboard.connector import AsyncioConnector +from plugboard.exceptions import IOSetupError, ValidationError +from plugboard.process import LocalProcess +from plugboard.schemas import ConnectorSpec + + +class Animal: + """Base payload class.""" + + +class Dog(Animal): + """A narrower payload class.""" + + +@pytest.mark.parametrize( + "source,target,expected", + [ + (Dog, Animal, True), + (Animal, Dog, False), + (int, str, False), + (int, float, False), + (bool, int, True), + (_t.Any, int, None), + (int, _t.Any, None), + (str, str | None, True), + (str | None, str, False), + (int | str, str | int | None, True), + (_t.Union[int, str], int, False), + (list[Dog], list[Animal], False), + (list[int], list[int], True), + (list[None], list[type(None)], True), + (list[_t.Any], list[int], None), + (list, list[int], None), + (list[int], list, True), + (dict[str, int], dict[str, str], False), + (dict[str, Dog], dict[str, Animal], False), + (frozenset[Dog], frozenset[Animal], True), + (set[Dog], set[Animal], False), + (tuple[Dog, int], tuple[Animal, int], True), + (tuple[Dog, Dog], tuple[Animal, ...], True), + (tuple[Dog, ...], tuple[Animal, ...], True), + (tuple[int, ...], tuple[int, int], False), + (tuple[int], tuple[int, int], False), + (tuple[()], tuple[int, ...], True), + (tuple[int], tuple[()], False), + (tuple, tuple[int], None), + (_t.Tuple, tuple[int], None), + (tuple[int], _t.Tuple, True), + (list[int], object, True), + ], +) +def test_connection_compatibility(source: _t.Any, target: _t.Any, expected: bool | None) -> None: + """Connections respect nominal subtyping, unions, and container variance.""" + assert is_assignable(source, target) is expected + + +def test_legacy_and_mapping_declarations() -> None: + """Lists retain Any semantics and controllers own their normalized mappings.""" + legacy = IO(inputs=["x"], outputs=("y",)) + assert legacy.inputs == ["x"] + assert legacy.input_types == {"x": _t.Any} + assert legacy.output_types == {"y": _t.Any} + assert "input_types" not in legacy.dict() + types = {"x": int, "optional": None} + typed = IO(inputs=types) + types["x"] = str + assert typed.input_types == {"x": int, "optional": type(None)} + assert typed.inputs == ["x", "optional"] + assert msgspec.json.decode(msgspec.json.encode(typed.dict()))["input_types"]["x"] == "int" + + +@pytest.mark.parametrize( + "hint", ["int", 123, list[_t.ForwardRef("Missing")], _t.Literal[1], _t.Callable] +) +def test_unsupported_types(hint: _t.Any) -> None: + """Unsupported declarations fail at setup even when runtime validation is off.""" + with pytest.raises(IOSetupError, match="Unsupported port type"): + IO(inputs={"x": hint}) + + +def test_invalid_mode() -> None: + """Typos in validation policy fail immediately.""" + with pytest.raises(ValueError, match="Unknown runtime validation mode"): + IO(runtime_validation=_t.cast(_t.Any, "typo")) + + +def test_inheritance_and_instance_copy() -> None: + """Inherited ports keep types, and legacy redeclarations do not erase them.""" + + class Parent(Component): + io = IO(inputs={"x": int}, outputs={"y": list[str]}, runtime_validation="deep") + + class Child(Parent): + io = IO(inputs=["x", "z"], runtime_validation="shallow", validate_outputs=True) + + child = Child(name="child") + assert child.io.input_types == {"x": int, "z": _t.Any} + assert child.io.output_types == {"y": list[str]} + assert child.io.runtime_validation == "shallow" + assert child.io.validate_outputs + assert child.io.input_types is not Child.io.input_types + + class Grandchild(Child): + pass + + assert Grandchild.io.input_types == child.io.input_types + assert Grandchild.io.runtime_validation == "shallow" + + +def test_conflicting_inheritance() -> None: + """Both overrides and conflicting multiple inheritance are rejected.""" + + class Left(Component, ABC): + io = IO(inputs={"x": int}) + + class Right(Component, ABC): + io = IO(inputs={"x": str}) + + with pytest.raises(IOSetupError, match="Conflicting inherited types"): + + class Override(Left): + io = IO(inputs={"x": str}) + + with pytest.raises(IOSetupError, match="Conflicting inherited types"): + + class Combined(Left, Right): + pass + + +def test_legacy_dynamic_names_and_mixin() -> None: + """Names added to legacy lists and declarations supplied by mixins still work.""" + + class Parent(Component): + io = IO(outputs=["x"]) + + Parent.io.outputs.append("y") + assert Parent(name="parent").io.output_types == {"x": _t.Any, "y": _t.Any} + + class Child(Parent): + pass + + assert Child.io.output_types == {"x": _t.Any, "y": _t.Any} + + class Mixin: + io = IO(outputs={"x": int}) + + class Mixed(Mixin, Component): + pass + + assert Mixed(name="mixed").io.output_types == {"x": int} + + +@pytest.mark.parametrize( + "hint,value", + [ + (int, 1), + (int, True), + (Animal, Dog()), + (str | None, None), + (list[int | None], [1, None]), + (list[None], [None]), + (dict[str, list[int]], {"x": [1, 2]}), + (tuple[int, str], (1, "x")), + (tuple[int, ...], (1, 2, 3)), + (tuple[()], ()), + (_t.Tuple, (1, "x")), + (set[int], {1, 2}), + (frozenset[str], frozenset({"x"})), + (_t.Any, object()), + ], +) +def test_deep_valid_values(hint: _t.Any, value: _t.Any) -> None: + """Validation supports the declared subset and does not transform values.""" + compile_validator(hint, deep=True)(value, "component.port") + + +@pytest.mark.parametrize( + "hint,value,path", + [ + (int, "1", "component.port"), + (float, 1, "component.port"), + (list[int], ["1"], "component.port[0]"), + (list[None], [1], "component.port[0]"), + (dict[str, list[int]], {"x": ["1"]}, "component.port.values[0][0]"), + (dict[str, int], {1: 2}, "component.port.keys[0]"), + (tuple[int, str], (1, 2), "component.port[1]"), + (tuple[int, ...], (1, "2"), "component.port[1]"), + (tuple[()], (1,), "component.port"), + (set[int], {"1"}, "component.port[0]"), + (str | None, 1, "component.port"), + ], +) +def test_deep_invalid_values(hint: _t.Any, value: _t.Any, path: str) -> None: + """Nested validation failures include a useful path without coercion.""" + with pytest.raises(ValidationError) as exc: + compile_validator(hint, deep=True)(value, "component.port") + assert path in str(exc.value) + + +def test_runtime_policy_and_compile_once() -> None: + """Off does not compile validators; active policies reuse them for every value.""" + with patch( + "plugboard.component.io_controller.compile_validator", wraps=compile_validator + ) as spy: + off = IO(inputs={"x": int}, outputs={"y": str}, validate_outputs=True) + off.validate_inputs({"x": "anything"}) + assert not spy.called + deep = IO(inputs={"x": list[int]}, runtime_validation="deep") + count = spy.call_count + deep.validate_inputs({"x": [1]}) + deep.validate_inputs({"x": [2]}) + assert spy.call_count == count + + shallow = IO(inputs={"x": list[int]}, runtime_validation="shallow") + shallow.validate_inputs({"x": ["unchecked contents"]}) + with pytest.raises(ValidationError, match="expected list"): + shallow.validate_inputs({"x": "wrong outer type"}) + + +def test_binding_and_output_fanout() -> None: + """Missing inputs are skipped, supplied None is checked, and outputs validate once.""" + + class Typed(Component): + io = IO( + inputs={"x": int}, + outputs={"y": int}, + runtime_validation="deep", + validate_outputs=True, + ) + + async def step(self) -> None: + self.y = self.x + + comp = Typed(name="typed") + comp._bind_inputs() + assert not comp._field_inputs_ready + comp.x = None + with pytest.raises(ValidationError, match=r"typed.x \(input\).*expected int.*NoneType"): + comp._bind_inputs() + comp.x = 1 + comp._bind_inputs() + comp.y = "wrong" + with pytest.raises(ValidationError, match=r"typed.y \(output\).*expected int.*str"): + comp._bind_outputs() + assert not comp.io.buf_fields["output"] + comp.y = 1 + validate = Mock(wraps=comp.io._output_validators["y"]) + comp.io._output_validators["y"] = validate + comp.io._output_channels = {("y", "a"): Mock(), ("y", "b"): Mock()} + comp._bind_outputs() + assert validate.call_count == 1 + + +def test_output_checking_is_opt_in() -> None: + """Enabling input checks alone does not compile output checks.""" + io = IO(inputs={"x": int}, outputs={"y": int}, runtime_validation="deep") + assert io._input_validators + assert not io._output_validators + io.validate_output_values({"y": "unchecked"}) + + +def test_initialization_checks_each_edge_and_allows_any() -> None: + """Fan-in/fan-out checks report every incompatible edge and allow untyped ports.""" + + class IntSource(Component): + io = IO(outputs={"value": int}) + + class StringSource(Component): + io = IO(outputs={"value": str}) + + class IntSink(Component): + io = IO(inputs={"value": int}) + + class UntypedSink(Component): + io = IO(inputs=["value"]) + + components = [ + IntSource(name="numbers"), + StringSource(name="strings"), + IntSink(name="sink_a"), + IntSink(name="sink_b"), + UntypedSink(name="untyped"), + ] + connectors = [ + AsyncioConnector(ConnectorSpec(source=f"{source}.value", target=f"{target}.value")) + for source, target in ( + ("numbers", "sink_a"), + ("strings", "sink_a"), + ("strings", "sink_b"), + ("strings", "untyped"), + ) + ] + process = LocalProcess(components, connectors) + with pytest.raises(ValidationError) as exc: + process._validate_io_types() + assert str(exc.value).count("Incompatible port types") == 2 + assert "strings.value (str) -> sink_a.value (int)" in str(exc.value) + assert "strings.value (str) -> sink_b.value (int)" in str(exc.value) + assert "untyped" not in str(exc.value) + process.connectors = {connectors[0].id: connectors[0], connectors[-1].id: connectors[-1]} + process._validate_io_types() + + +def test_mapping_declarations_on_decorated_components() -> None: + """Function components accept mappings without inferring types from annotations.""" + + @component(inputs={"x": int}, outputs={"y": str}) + async def transform(x: int) -> dict[str, str]: + return {"y": str(x)} + + comp = transform.component(name="transform") + assert comp.io.input_types == {"x": int} + assert comp.io.output_types == {"y": str} + + class AnnotationsOnly(Component): + io = IO(inputs=["x"]) + x: int + + assert AnnotationsOnly.io.input_types == {"x": _t.Any}