diff --git a/pyproject.toml b/pyproject.toml index 13063f53..9c784722 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "ezmsg" -version = "3.10.0b3" +version = "3.10.0b4" description = "A simple DAG-based computation model" authors = [ { name = "Griffin Milsap", email = "griffin.milsap@gmail.com" }, @@ -44,6 +44,9 @@ test = [ "xarray>=2025.6.1", "psutil>=7.1.0", "pytest-benchmark>=5.2.3", + # Exercises the schema extra's pydantic-present paths; the absent paths + # are tested by stubbing the soft import out. + "pydantic>=2.7", ] docs = [ {include-group = "axisarray"}, @@ -71,6 +74,12 @@ perf = [ dashboard = [ "ezmsg-dashboard; python_version >= '3.11'", ] +# Standard JSON Schema in settings metadata, and validation/coercion of +# dynamic settings field updates in the owning process. Without it, schema +# fields are None and field updates apply raw, exactly as before. +schema = [ + "pydantic>=2.7", +] [tool.pytest.ini_options] addopts = ["--import-mode=importlib"] diff --git a/src/ezmsg/core/backend.py b/src/ezmsg/core/backend.py index e6a5a04e..994a9a5a 100644 --- a/src/ezmsg/core/backend.py +++ b/src/ezmsg/core/backend.py @@ -710,16 +710,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/src/ezmsg/core/backendprocess.py b/src/ezmsg/core/backendprocess.py index e26e3a29..4fc829e4 100644 --- a/src/ezmsg/core/backendprocess.py +++ b/src/ezmsg/core/backendprocess.py @@ -50,6 +50,7 @@ from .subclient import Subscriber from .netprotocol import AddressType from .settingsmeta import ( + coerce_settings_field_value, settings_repr_value, settings_schema_from_value, settings_structured_value, @@ -422,10 +423,23 @@ async def process_request_handler( ) try: + # Authoritative validation/coercion: this process is the one + # place the settings class (and everything it references, + # enums included) is importable, so a value is checked against + # the field's annotation HERE rather than trusting callers to + # pre-cast. With the `schema` extra absent, or for annotations + # pydantic cannot resolve, the value applies raw as before; a + # refused value fails the request instead of publishing a + # value the settings type cannot hold. + coerced_value = coerce_settings_field_value( + type(current_settings[unit_address]), + update_obj.field_path, + update_obj.value, + ) patched = self._replace_settings_field( current_settings[unit_address], update_obj.field_path, - update_obj.value, + coerced_value, ) control_pub = control_publishers.get(input_topic) if control_pub is None: diff --git a/src/ezmsg/core/commands/__init__.py b/src/ezmsg/core/commands/__init__.py index 175c3c96..4faaa541 100644 --- a/src/ezmsg/core/commands/__init__.py +++ b/src/ezmsg/core/commands/__init__.py @@ -2,6 +2,7 @@ from .dashboard_cmd import setup_dashboard_cmdline from .graphviz import setup_graphviz_cmdline +from .inspect import setup_inspect_cmdline from .mermaid import setup_mermaid_cmdline from .serve import setup_serve_cmdline from .shutdown import setup_shutdown_cmdline @@ -15,3 +16,4 @@ def setup_core_cmdline(subparsers: argparse._SubParsersAction) -> None: setup_shutdown_cmdline(subparsers) setup_graphviz_cmdline(subparsers) setup_mermaid_cmdline(subparsers) + setup_inspect_cmdline(subparsers) diff --git a/src/ezmsg/core/commands/inspect.py b/src/ezmsg/core/commands/inspect.py new file mode 100644 index 00000000..45232e0d --- /dev/null +++ b/src/ezmsg/core/commands/inspect.py @@ -0,0 +1,156 @@ +"""``ezmsg inspect`` — describe an app's components without running it. + +Everything reported here is read from CLASS-level state the component +metaclasses already maintain (``__streams__``, ``__settings_type__``, +``__components__``), so nothing is instantiated, no settings are required, +and no graph connection is involved: the target module is imported and its +component classes are described as declared. Because a child's attribute +name is exactly the segment ezmsg uses for its runtime address, the +``path`` reported for each nested component matches the address it will +register under at run time (below whatever root name ``ez.run`` is given — +the class name stands in for that root segment here). + +Two things are invisible to a static description, by design: components a +Collection creates dynamically in ``configure()``, and stream addresses +(assigned at graph build). Settings schemas ride along in both the +field-list form and, when the ``schema`` extra is installed, the standard +JSON Schema — the same payload the graph metadata carries for a running +app, so consumers can share one contract for "at rest" and "live". +""" + +import argparse +import dataclasses +import importlib +import importlib.util +import json +import sys +from pathlib import Path +from typing import Any + +from ..collection import Collection +from ..component import Component +from ..settingsmeta import settings_schema_from_type +from ..stream import InputStream + +INSPECT_SCHEMA_VERSION = 1 + + +def _type_name(tp: object) -> str: + if isinstance(tp, type): + return f"{tp.__module__}.{tp.__qualname__}" + return str(tp) + + +def _load_module(source: str) -> Any: + """Import ``source`` as a file path or a dotted module name.""" + path = Path(source).expanduser() + if path.suffix == ".py" or path.exists(): + spec = importlib.util.spec_from_file_location(path.stem, path) + if spec is None or spec.loader is None: + raise ImportError(f"Cannot load a module from {source!r}") + module = importlib.util.module_from_spec(spec) + # Registered so dataclass/typing resolution that imports the module + # by name (get_type_hints, pydantic) can find it. + sys.modules[path.stem] = module + spec.loader.exec_module(module) + return module + return importlib.import_module(source) + + +def _streams_description(component_cls: type) -> list[dict[str, Any]]: + streams = getattr(component_cls, "__streams__", {}) + return [ + { + "name": name, + "kind": type(stream).__name__, + "msg_type": _type_name(getattr(stream, "msg_type", None)), + } + for name, stream in streams.items() + ] + + +def _component_description(name: str, path: str, component_cls: type) -> dict[str, Any]: + settings_type = getattr(component_cls, "__settings_type__", None) + schema = settings_schema_from_type(settings_type) if settings_type is not None else None + input_settings = getattr(component_cls, "__streams__", {}).get("INPUT_SETTINGS") + description: dict[str, Any] = { + "name": name, + "path": path, + "component_type": _type_name(component_cls), + "collection": issubclass(component_cls, Collection), + "settings_type": _type_name(settings_type) if settings_type is not None else None, + "settings_schema": dataclasses.asdict(schema) if schema is not None else None, + # Same rule the graph metadata applies to a running component: only + # an InputStream inlet named INPUT_SETTINGS accepts dynamic updates. + "dynamic_settings": isinstance(input_settings, InputStream), + "streams": _streams_description(component_cls), + } + children = getattr(component_cls, "__components__", None) + if issubclass(component_cls, Collection) and isinstance(children, dict): + description["components"] = [ + _component_description(child_name, f"{path}/{child_name}", type(child)) + for child_name, child in children.items() + ] + return description + + +def _root_component_classes(module: Any, component_name: str | None) -> list[type]: + if component_name is not None: + target = getattr(module, component_name, None) + if not (isinstance(target, type) and issubclass(target, Component)): + raise SystemExit( + f"{component_name!r} is not an ezmsg Component class in {module.__name__!r}" + ) + return [target] + roots = [ + value + for value in vars(module).values() + if isinstance(value, type) + and issubclass(value, Component) + and value.__module__ == module.__name__ + ] + if not roots: + raise SystemExit(f"No ezmsg Component classes are defined in {module.__name__!r}") + return roots + + +def handle_inspect(args: argparse.Namespace) -> None: + try: + module = _load_module(args.source) + except SystemExit: + raise + except BaseException as exc: + raise SystemExit(f"Could not import {args.source!r}: {type(exc).__name__}: {exc}") from exc + + roots = _root_component_classes(module, args.component) + result = { + "inspect_schema_version": INSPECT_SCHEMA_VERSION, + "source": args.source, + "components": [ + _component_description(root.__name__, root.__name__, root) for root in roots + ], + } + print(json.dumps(result, indent=args.indent)) + + +def setup_inspect_cmdline(subparsers: argparse._SubParsersAction) -> None: + parser = subparsers.add_parser( + "inspect", + help="describe a module's components and settings schemas without running them", + ) + parser.add_argument( + "source", + help="path to a .py file, or a dotted module name, defining ezmsg Components", + ) + parser.add_argument( + "--component", + default=None, + help="describe only this Component class (default: every Component the module defines)", + ) + parser.add_argument( + "--indent", + type=int, + default=None, + help="pretty-print the JSON with this indent (default: one line)", + ) + parser.set_defaults(_handler=handle_inspect) diff --git a/src/ezmsg/core/graphmeta.py b/src/ezmsg/core/graphmeta.py index ba25f46a..5d1587da 100644 --- a/src/ezmsg/core/graphmeta.py +++ b/src/ezmsg/core/graphmeta.py @@ -122,6 +122,12 @@ class SettingsSchemaMetadata: provider: str settings_type: str fields: list[SettingsFieldMetadata] + # Standard JSON Schema for the settings type, generated by pydantic when + # the optional `schema` extra is installed (None otherwise, and for types + # pydantic cannot model). Rides everywhere this schema already goes — + # component metadata, settings snapshots, settings events — so any + # consumer can validate values without parsing the field_type strings. + json_schema: dict[str, Any] | None = None SettingsReprType: TypeAlias = dict[str, Any] | str diff --git a/src/ezmsg/core/settingsmeta.py b/src/ezmsg/core/settingsmeta.py index c7478945..7d91f248 100644 --- a/src/ezmsg/core/settingsmeta.py +++ b/src/ezmsg/core/settingsmeta.py @@ -1,12 +1,145 @@ from __future__ import annotations -from dataclasses import MISSING, asdict, fields as dataclass_fields, is_dataclass import enum +import types +import typing from collections.abc import Mapping -from typing import Any, get_args, get_origin +from dataclasses import MISSING, asdict, is_dataclass +from dataclasses import fields as dataclass_fields +from typing import Annotated, Any, get_args, get_origin, get_type_hints from .graphmeta import SettingsFieldMetadata, SettingsSchemaMetadata +# Soft dependency (the `schema` extra): with pydantic installed, settings +# metadata carries a standard JSON Schema and dynamic settings field updates +# are validated/coerced in the owning process; without it, schema fields are +# None and updates apply raw, exactly as before. +# +# Resolved on first use rather than at import. This module is reached by a +# plain ``import ezmsg.core``, so importing pydantic here would charge every +# ezmsg process ~200 ms and ~5 MB — including every spawned backend process, +# which on spawn platforms re-imports from scratch — whether or not it ever +# describes or coerces a setting. Deferring it means only the process that +# builds graph metadata, and processes that actually handle settings, pay. +_UNRESOLVED = object() +_TYPE_ADAPTER: Any = _UNRESOLVED + + +def _type_adapter() -> Any: + """``pydantic.TypeAdapter`` if it can be imported, else None (cached).""" + global _TYPE_ADAPTER + if _TYPE_ADAPTER is _UNRESOLVED: + try: + from pydantic import TypeAdapter + except ImportError: # pragma: no cover — exercised via the stubbed tests + TypeAdapter = None + _TYPE_ADAPTER = TypeAdapter + return _TYPE_ADAPTER + + +class SettingsCoercionError(ValueError): + """A dynamic settings field update refused by the settings type. + + Raised only when pydantic is installed AND resolved the field's + annotation AND rejected the value — never for gaps in coverage, which + fall back to applying the value raw. + """ + + +def settings_json_schema(settings_type: object) -> dict[str, Any] | None: + """Standard JSON Schema for a settings type, or None. Never raises. + + None means pydantic is not installed, or it cannot model this type (an + arbitrary-typed field, an exotic annotation); consumers treat that as + "no schema" and keep whatever behavior they had. + """ + type_adapter = _type_adapter() + if type_adapter is None or not isinstance(settings_type, type): + return None + try: + schema = type_adapter(settings_type).json_schema(mode="validation") + except Exception: + return None + return schema if isinstance(schema, dict) else None + + +def _unwrap_annotation(annotation: Any) -> Any: + """Peel Annotated and single-member-Optional wrappers for shape checks. + + Only used to decide how to DESCEND a dotted field path; the full + (wrapped) annotation is what gets validated against, so metadata like + pydantic Field constraints is never lost. + """ + while True: + if get_origin(annotation) is Annotated: + annotation = get_args(annotation)[0] + continue + origin = get_origin(annotation) + if origin in (typing.Union, types.UnionType): + non_none = [arg for arg in get_args(annotation) if arg is not type(None)] + if len(non_none) == 1: + annotation = non_none[0] + continue + return annotation + + +def _annotation_at_path(settings_type: type, field_path: str) -> Any | None: + """The annotation a dotted settings field path addresses, or None. + + Walks dataclass/pydantic-model fields by name and str-keyed mappings by + value type — the same shapes ``_replace_settings_field`` patches. None + (unknown field, unresolvable hints, an unsupported container in the + middle of the path) means "no annotation available", not an error. + """ + current: Any = settings_type + for segment in str(field_path).split("."): + base = _unwrap_annotation(current) + if is_dataclass(base) or hasattr(base, "model_fields"): + try: + hints = get_type_hints(base, include_extras=True) + except Exception: + return None + if segment not in hints: + return None + current = hints[segment] + continue + if get_origin(base) is dict or (isinstance(get_origin(base), type) and issubclass(get_origin(base), Mapping)): + args = get_args(base) + if len(args) != 2: + return None + current = args[1] + continue + return None + return current + + +def coerce_settings_field_value(settings_type: object, field_path: str, value: Any) -> Any: + """Validate/coerce one dynamic settings field update against its annotation. + + The authoritative reconstruction point: this runs in the process that + owns the unit, the one place the settings class and everything it + references (enums included) is importable. Returns the coerced value; + returns ``value`` unchanged when pydantic is absent or the annotation + cannot be resolved/modelled (raw behavior, as before this existed); and + raises :class:`SettingsCoercionError` when pydantic refuses the value — + a refusal the process control response reports back to the caller + instead of publishing a value the settings type cannot hold. + """ + type_adapter = _type_adapter() + if type_adapter is None or not isinstance(settings_type, type): + return value + annotation = _annotation_at_path(settings_type, field_path) + if annotation is None: + return value + try: + adapter = type_adapter(annotation) + except Exception: + return value + try: + return adapter.validate_python(value) + except Exception as exc: + raise SettingsCoercionError(f"Invalid value for settings field '{field_path}': {exc}") from None + def _type_name(tp: object) -> str: if isinstance(tp, type): @@ -41,6 +174,19 @@ def settings_structured_value(value: object) -> dict[str, Any] | None: if value is None: return None + # Pydantic's JSON-mode dump first: it round-trips exactly what the + # type's JSON Schema describes (enums by value, tuples as arrays, paths + # and datetimes as strings). A type pydantic cannot model fails at + # adapter construction and falls through to the legacy renderings. + type_adapter = _type_adapter() + if type_adapter is not None: + try: + dumped = type_adapter(type(value)).dump_python(value, mode="json") + if isinstance(dumped, dict): + return dumped + except Exception: + pass + if is_dataclass(value): try: asdict_value = asdict(value) @@ -144,6 +290,8 @@ def settings_schema_from_type(settings_type: object) -> SettingsSchemaMetadata | if not isinstance(settings_type, type): return None + json_schema = settings_json_schema(settings_type) + if is_dataclass(settings_type): fields: list[SettingsFieldMetadata] = [] for f in dataclass_fields(settings_type): @@ -193,6 +341,7 @@ def settings_schema_from_type(settings_type: object) -> SettingsSchemaMetadata | provider="dataclass", settings_type=_type_name(settings_type), fields=fields, + json_schema=json_schema, ) if hasattr(settings_type, "model_fields"): @@ -233,6 +382,7 @@ def settings_schema_from_type(settings_type: object) -> SettingsSchemaMetadata | provider="pydantic", settings_type=_type_name(settings_type), fields=fields, + json_schema=json_schema, ) if hasattr(settings_type, "__fields__"): @@ -266,6 +416,7 @@ def settings_schema_from_type(settings_type: object) -> SettingsSchemaMetadata | provider="pydantic", settings_type=_type_name(settings_type), fields=fields, + json_schema=json_schema, ) param_ns = getattr(settings_type, "param", None) @@ -323,6 +474,7 @@ def settings_schema_from_type(settings_type: object) -> SettingsSchemaMetadata | provider="param", settings_type=_type_name(settings_type), fields=fields, + json_schema=json_schema, ) return None 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_inspect_command.py b/tests/test_inspect_command.py new file mode 100644 index 00000000..efff22e1 --- /dev/null +++ b/tests/test_inspect_command.py @@ -0,0 +1,133 @@ +"""``ezmsg inspect``: static component/settings description, no run needed.""" + +import json + +import pytest + +from ezmsg.core.command import cmdline + +DEMO_MODULE = ''' +import ezmsg.core as ez +from ezmsg.util.messages.axisarray import AxisArray + + +class GainSettings(ez.Settings): + gain: float = 1.0 + label: str = "" + + +class GainUnit(ez.Unit): + SETTINGS = GainSettings + + INPUT_SETTINGS = ez.InputStream(GainSettings) + INPUT_SIGNAL = ez.InputStream(AxisArray) + OUTPUT_SIGNAL = ez.OutputStream(AxisArray) + + @ez.subscriber(INPUT_SETTINGS) + async def on_settings(self, msg: GainSettings) -> None: + self.apply_settings(msg) + + +class TopicalSystem(ez.Collection): + # An InputTopic named INPUT_SETTINGS: discoverable, but NOT dynamic + # settings — a topic is not an InputStream (the SpikeRateFromRaw shape). + SETTINGS = GainSettings + + INPUT_SETTINGS = ez.InputTopic(GainSettings) + + GAIN = GainUnit() + + def network(self) -> ez.NetworkDefinition: + return () + + +class DemoSystem(ez.Collection): + SETTINGS = GainSettings + + GAIN = GainUnit() + + def network(self) -> ez.NetworkDefinition: + return () +''' + + +@pytest.fixture +def demo_module_path(tmp_path): + path = tmp_path / "inspect_demo_system.py" + path.write_text(DEMO_MODULE, encoding="utf-8") + return path + + +def _inspect(capsys, *argv): + cmdline(argv=["inspect", *argv]) + return json.loads(capsys.readouterr().out) + + +def _by_name(components, name): + return next(component for component in components if component["name"] == name) + + +class TestInspectCommand: + def test_describes_every_component_class_in_the_module(self, capsys, demo_module_path): + result = _inspect(capsys, str(demo_module_path)) + assert result["inspect_schema_version"] == 1 + assert {component["name"] for component in result["components"]} == { + "GainUnit", + "TopicalSystem", + "DemoSystem", + } + + def test_collection_children_carry_runtime_address_paths(self, capsys, demo_module_path): + result = _inspect(capsys, str(demo_module_path)) + system = _by_name(result["components"], "DemoSystem") + assert system["collection"] is True + children = {child["name"]: child for child in system["components"]} + assert children["GAIN"]["path"] == "DemoSystem/GAIN" + assert children["GAIN"]["component_type"].endswith("GainUnit") + assert children["GAIN"]["collection"] is False + + def test_dynamic_settings_requires_an_input_stream_inlet(self, capsys, demo_module_path): + result = _inspect(capsys, str(demo_module_path)) + system = _by_name(result["components"], "DemoSystem") + children = {child["name"]: child for child in system["components"]} + assert children["GAIN"]["dynamic_settings"] is True + # An InputTopic named INPUT_SETTINGS does not accept dynamic updates — + # the same rule the running graph's metadata applies. + topical = _by_name(result["components"], "TopicalSystem") + assert topical["dynamic_settings"] is False + + def test_settings_schema_and_json_schema_ride_along(self, capsys, demo_module_path): + result = _inspect(capsys, str(demo_module_path)) + unit = _by_name(result["components"], "GainUnit") + schema = unit["settings_schema"] + assert unit["settings_type"].endswith("GainSettings") + assert {field["name"] for field in schema["fields"]} == {"gain", "label"} + json_schema = schema["json_schema"] + assert json_schema is not None + assert json_schema["properties"]["gain"]["type"] == "number" + + def test_streams_are_listed_with_kind_and_message_type(self, capsys, demo_module_path): + result = _inspect(capsys, str(demo_module_path)) + unit = _by_name(result["components"], "GainUnit") + streams = {stream["name"]: stream for stream in unit["streams"]} + assert streams["INPUT_SIGNAL"]["kind"] == "InputStream" + assert streams["INPUT_SIGNAL"]["msg_type"].endswith("AxisArray") + assert streams["OUTPUT_SIGNAL"]["kind"] == "OutputStream" + + def test_component_flag_selects_one_root(self, capsys, demo_module_path): + result = _inspect(capsys, str(demo_module_path), "--component", "DemoSystem") + assert [component["name"] for component in result["components"]] == ["DemoSystem"] + + def test_unknown_component_name_is_a_clear_error(self, demo_module_path): + with pytest.raises(SystemExit, match="Nope"): + cmdline(argv=["inspect", str(demo_module_path), "--component", "Nope"]) + + def test_module_without_components_is_a_clear_error(self): + with pytest.raises(SystemExit, match="No ezmsg Component classes"): + cmdline(argv=["inspect", "json"]) + + def test_broken_module_reports_the_import_failure(self, tmp_path): + path = tmp_path / "broken_module.py" + path.write_text("this is not python(\n", encoding="utf-8") + with pytest.raises(SystemExit, match="Could not import"): + cmdline(argv=["inspect", str(path)]) diff --git a/tests/test_run.py b/tests/test_run.py index 6923c4ab..d4ea1c70 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): diff --git a/tests/test_settings_json_schema.py b/tests/test_settings_json_schema.py new file mode 100644 index 00000000..6de0afb3 --- /dev/null +++ b/tests/test_settings_json_schema.py @@ -0,0 +1,144 @@ +"""Settings JSON Schema metadata and authoritative field coercion. + +Covers the `schema` extra's three behaviors — a standard JSON Schema riding +``SettingsSchemaMetadata``, pydantic-JSON structured snapshot values, and +validation/coercion of dynamic settings field updates in the owning process +— plus the soft-dependency contract: with pydantic stubbed out, schemas are +None and updates apply raw, exactly as before the extra existed. +""" + +import enum +from dataclasses import dataclass, field + +import pytest + +from ezmsg.core import settingsmeta +from ezmsg.core.settingsmeta import ( + SettingsCoercionError, + coerce_settings_field_value, + settings_json_schema, + settings_schema_from_type, + settings_schema_from_value, + settings_structured_value, +) + + +class Flavor(enum.Enum): + VANILLA = "vanilla" + MINT = "mint" + + +@dataclass +class SubSettings: + depth: int = 3 + + +@dataclass +class DemoSettings: + gain: float = 1.0 + taps: tuple[float, ...] = () + weights: dict[str, float] = field(default_factory=dict) + flavor: Flavor = Flavor.VANILLA + label: str | None = None + sub: SubSettings = field(default_factory=SubSettings) + + +class Opaque: + pass + + +@dataclass +class UnmodelableSettings: + handle: Opaque = None # type: ignore[assignment] + + +class TestSchemaAttachment: + def test_dataclass_metadata_carries_json_schema(self): + meta = settings_schema_from_type(DemoSettings) + assert meta is not None and meta.provider == "dataclass" + schema = meta.json_schema + assert schema is not None and schema["type"] == "object" + assert schema["properties"]["gain"]["type"] == "number" + assert schema["properties"]["taps"]["items"] == {"type": "number"} + assert schema["$defs"]["Flavor"]["enum"] == ["vanilla", "mint"] + assert schema["$defs"]["SubSettings"]["properties"]["depth"]["type"] == "integer" + + def test_schema_from_value_carries_it_too(self): + meta = settings_schema_from_value(DemoSettings()) + assert meta is not None and meta.json_schema is not None + + def test_unmodelable_type_keeps_fields_but_no_json_schema(self): + meta = settings_schema_from_type(UnmodelableSettings) + assert meta is not None + assert [f.name for f in meta.fields] == ["handle"] + assert meta.json_schema is None + assert settings_json_schema(UnmodelableSettings) is None + + def test_without_pydantic_schema_is_none(self, monkeypatch): + monkeypatch.setattr(settingsmeta, "_TYPE_ADAPTER", None) + assert settings_json_schema(DemoSettings) is None + meta = settings_schema_from_type(DemoSettings) + assert meta is not None and meta.json_schema is None + + +class TestStructuredValue: + def test_pydantic_json_mode_dump(self): + structured = settings_structured_value(DemoSettings(taps=(1.0, 2.0), flavor=Flavor.MINT)) + assert structured == { + "gain": 1.0, + "taps": [1.0, 2.0], + "weights": {}, + "flavor": "mint", + "label": None, + "sub": {"depth": 3}, + } + + def test_falls_back_for_unmodelable_values(self): + structured = settings_structured_value(UnmodelableSettings()) + # Legacy sanitize path: the opaque handle renders as repr. + assert structured is not None and set(structured) == {"handle"} + + def test_falls_back_without_pydantic(self, monkeypatch): + monkeypatch.setattr(settingsmeta, "_TYPE_ADAPTER", None) + structured = settings_structured_value(DemoSettings(flavor=Flavor.MINT)) + assert structured is not None and structured["flavor"] == "mint" + + +class TestFieldCoercion: + def test_int_promotes_to_declared_float(self): + coerced = coerce_settings_field_value(DemoSettings, "gain", 2) + assert coerced == 2.0 and isinstance(coerced, float) + + def test_list_becomes_the_declared_variadic_tuple(self): + coerced = coerce_settings_field_value(DemoSettings, "taps", [1, 2.5]) + assert coerced == (1.0, 2.5) and isinstance(coerced, tuple) + + def test_enum_reconstructs_by_value(self): + assert coerce_settings_field_value(DemoSettings, "flavor", "mint") is Flavor.MINT + + def test_enum_member_name_is_refused(self): + # The wire convention is member VALUES; a name is a refusal the + # caller sees, not a silent raw publish. + with pytest.raises(SettingsCoercionError): + coerce_settings_field_value(DemoSettings, "flavor", "MINT") + + def test_nested_dataclass_path(self): + assert coerce_settings_field_value(DemoSettings, "sub.depth", "4") == 4 + + def test_dict_value_path(self): + coerced = coerce_settings_field_value(DemoSettings, "weights.alpha", "0.5") + assert coerced == 0.5 and isinstance(coerced, float) + + def test_refused_value_raises_with_field_path(self): + with pytest.raises(SettingsCoercionError, match="'gain'"): + coerce_settings_field_value(DemoSettings, "gain", "loud") + + def test_unknown_field_passes_raw(self): + assert coerce_settings_field_value(DemoSettings, "nope", "x") == "x" + + def test_unresolvable_annotation_passes_raw(self): + assert coerce_settings_field_value(UnmodelableSettings, "handle", "x") == "x" + + def test_without_pydantic_everything_passes_raw(self, monkeypatch): + monkeypatch.setattr(settingsmeta, "_TYPE_ADAPTER", None) + assert coerce_settings_field_value(DemoSettings, "gain", "loud") == "loud"