From 14a762c73e2cb5eebccd0d6ce7cf5718958218be Mon Sep 17 00:00:00 2001 From: Andre Merzky Date: Sun, 30 Aug 2026 22:31:03 +0200 Subject: [PATCH 1/4] client: add_input and add_data_join reach the service (#30) The runtime had both; the service client had neither, which is what blocks the servicified demo. Two new graph verbs, passthrough on the session side: `add_input` carries data only -- dtype, channel, codec -- because the producer lives outside the framework, and `add_data_join` carries the joined dtype. Unit tests cover the binding, the runtime's channel refusal surfacing as 409, and join registration. `add_barrier` / `add_data_split_task` from #30 stay open: no consumer exists yet, and the barrier verb carries a wire-design decision (instances hold asyncio primitives and must not cross; a declarative spec should). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_016Npyz3Hbnwos12ESsdJ2YU --- src/digitaltwin/service/client.py | 29 ++++++++++- src/digitaltwin/service/session.py | 23 ++++++++- test/unit/test_service.py | 80 ++++++++++++++++++++++++++++++ 3 files changed, 129 insertions(+), 3 deletions(-) diff --git a/src/digitaltwin/service/client.py b/src/digitaltwin/service/client.py index 42af91e..5885138 100644 --- a/src/digitaltwin/service/client.py +++ b/src/digitaltwin/service/client.py @@ -27,7 +27,8 @@ from radical.orbit.client import PluginClient -from ..components import DataType, TypedData +from ..components import DataType, JoinDataType, TypedData +from ..streaming import CODEC_JSON from .wire import ( Package, decode, @@ -239,6 +240,32 @@ def add_agent( twin_id, "add_agent", package, input_dtype, output_dtype, *args, **kwargs )["state"] + def add_input( + self, + twin_id: str, + dtype: DataType, + channel: str, + codec: str = CODEC_JSON, + ) -> str: + """Bind an external channel to one of the twin's input dtypes. + + The producer lives outside the framework and publishes to + `channel`; the twin receives every message on it, decoded by + `codec` (`json` for plain scripts and instruments, `raw` for + bytes, `cloudpickle` only inside one trust domain). + """ + + return self._verb(twin_id, "add_input", dtype, channel, codec)["state"] + + def add_data_join(self, twin_id: str, join_dtype: JoinDataType) -> str: + """Register a join: one output event per complete set of inputs. + + `join_dtype` names the member dtypes; components downstream + consume the joined dtype like any other. + """ + + return self._verb(twin_id, "add_data_join", join_dtype)["state"] + def start(self, twin_id: str) -> str: """Start a twin. Starting a running twin is a no-op.""" diff --git a/src/digitaltwin/service/session.py b/src/digitaltwin/service/session.py index 2e14f05..996f1d6 100644 --- a/src/digitaltwin/service/session.py +++ b/src/digitaltwin/service/session.py @@ -23,9 +23,9 @@ from radical.orbit.plugin_session_base import PluginSession from rhapsody.backends.execution.orbit import OrbitExecutionBackend # type: ignore -from ..components import DataType, TypedData +from ..components import DataType, JoinDataType, TypedData from ..runtime import DTRuntime, RuntimeState -from ..streaming import PubSubClient +from ..streaming import CODEC_JSON, PubSubClient from .wire import Package, check_versions, decode, encode log = logging.getLogger("radical.orbit") @@ -77,6 +77,8 @@ "add_task", "add_investigator", "add_agent", + "add_input", + "add_data_join", "start", "stop", "describe", @@ -443,6 +445,23 @@ async def _verb_add_agent( component = self._instantiate(package, twin) twin.runtime.add_agent(component, input_dtype, output_dtype, *args, **kwargs) + async def _verb_add_input( + self, + twin: TwinInstance, + dtype: DataType, + channel: str, + codec: str = CODEC_JSON, + ) -> None: + # data-only arguments, no Package: the producer lives outside the + # framework, only the binding crosses the wire. The runtime + # validates channel and codec and subscribes at bind time. + twin.runtime.add_input(dtype, channel, codec) + + async def _verb_add_data_join( + self, twin: TwinInstance, join_dtype: JoinDataType + ) -> None: + twin.runtime.add_data_join(join_dtype) + async def _verb_start(self, twin: TwinInstance) -> None: # a running twin is left running: an idempotent retry, not an error twin.runtime.start() diff --git a/test/unit/test_service.py b/test/unit/test_service.py index 42e735b..23571d5 100644 --- a/test/unit/test_service.py +++ b/test/unit/test_service.py @@ -902,3 +902,83 @@ async def test_endpoint_loss_spares_pool_backed_roles(): assert session.endpoints_lost({"hpc1", "exsitu", "pool"}) == () assert not session._lost + + +# --------------------------------------------------------------------------- +# graph verbs: external inputs and joins (#30) +# --------------------------------------------------------------------------- + +class _BindingStream(_FakeStream): + """A `_FakeStream` that also accepts channel subscriptions.""" + + def __init__(self): + self.subscribed = [] + + async def subscribe_to_channel(self, channel, dtype, queue, codec): + self.subscribed.append((channel, dtype, codec)) + + +async def test_add_input_binds_a_channel_through_the_verb(): + """`add_input` carries data only -- dtype, channel, codec -- and the + binding lands in the runtime exactly as a direct call would leave it.""" + + session = DTSession("s1") + twin = _running_twin(session, "t1") + twin.runtime.streamer = _BindingStream() + + x = DataType("x") + await session.twin_call( + "t1", "add_input", + encode({"args": (x, "lab/raw")}), version_stamp()) + + binding = twin.runtime.inputs[0] + assert (binding.dtype, binding.channel, binding.codec) == (x, "lab/raw", + "json") + # subscribed at bind time, so nothing published before start() is lost + await asyncio.sleep(0) + assert twin.runtime.streamer.subscribed == [("lab/raw", x, "json")] + assert twin.summary()["calls"] == {"add_input": 1} + + await twin.close() + + +async def test_add_input_rejects_a_bad_channel_as_client_error(): + """The runtime's own refusal (reserved prefix) surfaces as a 409, + the same contract every other graph verb has.""" + + session = DTSession("s1") + twin = _running_twin(session, "t1") + twin.runtime.streamer = _BindingStream() + + with pytest.raises(HTTPException) as raised: + await session.twin_call( + "t1", "add_input", + encode({"args": (DataType("x"), "dt/oops")}), version_stamp()) + + assert raised.value.status_code == 409 + assert twin.runtime.inputs == [] + + await twin.close() + + +async def test_add_data_join_registers_through_the_verb(): + """One joined dtype, consumable downstream like any other.""" + + from digitaltwin.components import JoinDataType + + session = DTSession("s1") + twin = _running_twin(session, "t1") + + a, b = DataType("a"), DataType("b") + joined = JoinDataType([a, b]) + await session.twin_call( + "t1", "add_data_join", encode({"args": (joined,)}), version_stamp()) + + registered = [ant.component.out_dtype + for ants in twin.runtime.components.values() + for ant in ants + if getattr(ant.component, "out_dtype", None) == joined] + assert registered, "join component not registered" + assert twin.summary()["calls"] == {"add_data_join": 1} + + await twin.close() From 83cdfdd8019a454817e3af85774e114a3abf9607 Mon Sep 17 00:00:00 2001 From: Andre Merzky Date: Sun, 30 Aug 2026 22:31:03 +0200 Subject: [PATCH 2/4] orbit stream: a None connect timeout no longer crashes the participant `ChannelPublisher.open()` defaults to `timeout=None` -- the streaming contract's "wait forever" -- but the orbit backend forwarded that None into ORBIT's `start(timeout: float)`, which dies on `monotonic() + None`. Every external producer on the orbit data plane with the default timeout hit this. None is now approximated by a day. The new integration test runs the servicified demo shape end to end: two external channels bound with `add_input`, joined with `add_data_join`, consumed by a `JoinSink` -- producers are plain `ChannelPublisher`s using exactly that default. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_016Npyz3Hbnwos12ESsdJ2YU --- src/digitaltwin/streaming_orbit.py | 7 ++- test/integration/test_dt_orbit_stream.py | 64 +++++++++++++++++++++++- test/integration/twin_components.py | 9 ++++ 3 files changed, 77 insertions(+), 3 deletions(-) diff --git a/src/digitaltwin/streaming_orbit.py b/src/digitaltwin/streaming_orbit.py index 64a47aa..f9e7af9 100644 --- a/src/digitaltwin/streaming_orbit.py +++ b/src/digitaltwin/streaming_orbit.py @@ -165,7 +165,12 @@ def _start_runtime(self, timeout: Optional[float]) -> EndpointRuntime: role="stream") try: - runtime.start(wait=True, timeout=timeout) + # ORBIT's start() takes a finite deadline only; the streaming + # contract's "None waits forever" is approximated by a day -- a + # participant that cannot register within that is stuck, not + # waiting. + runtime.start(wait=True, + timeout=86400.0 if timeout is None else timeout) # start() returns *silently* when it merely timed out, so the # registration has to be checked explicitly -- otherwise an diff --git a/test/integration/test_dt_orbit_stream.py b/test/integration/test_dt_orbit_stream.py index e97cb2d..d089d62 100644 --- a/test/integration/test_dt_orbit_stream.py +++ b/test/integration/test_dt_orbit_stream.py @@ -15,10 +15,14 @@ import pytest -from digitaltwin.components import TRUTHY, TypedData +from digitaltwin.components import NULL_DTYPE, TRUTHY, DataType, JoinDataType, TypedData from digitaltwin.config import BACKEND_ORBIT from digitaltwin.service import register_user_modules -from digitaltwin.streaming import connect_stream_client +from digitaltwin.streaming import ( + ChannelPublisher, + PubSubConfig, + connect_stream_client, +) import learner_components import twin_components @@ -286,3 +290,59 @@ def test_a_persistent_component_publishes_through_the_injected_client( assert seen == sorted(seen), seen orbit_dt.twin_close(twin_id) + + +# --------------------------------------------------------------------------- +# external channels and joins through the client verbs (#30) +# --------------------------------------------------------------------------- + +def test_external_channels_join_through_the_client_verbs(orbit_dt, twin_id): + """The servicified demo shape: two external channels bound with + `add_input`, joined with `add_data_join`, consumed downstream. The + producers are plain `ChannelPublisher`s -- outside the framework, + knowing nothing about twins.""" + + from twin_components import JoinSink + + a = DataType("chan-a") + b = DataType("chan-b") + joined = JoinDataType([a, b]) + + chan_a = f"itest/{twin_id[:8]}/a" + chan_b = f"itest/{twin_id[:8]}/b" + + orbit_dt.create_twin(twin_id) + orbit_dt.add_input(twin_id, a, chan_a) + orbit_dt.add_input(twin_id, b, chan_b) + orbit_dt.add_data_join(twin_id, joined) + orbit_dt.add_task(twin_id, orbit_dt.package(JoinSink), joined, NULL_DTYPE) + assert orbit_dt.start(twin_id) == "running" + + async def feed_and_collect(count=3): + config = PubSubConfig(kind=BACKEND_ORBIT, broker_url=ORBIT_BROKER_URL) + collector = await connect_stream_client( + twin_id, backend=BACKEND_ORBIT, broker_url=ORBIT_BROKER_URL) + queue: asyncio.Queue = asyncio.Queue() + pub_a = await ChannelPublisher.open(chan_a, config=config) + pub_b = await ChannelPublisher.open(chan_b, config=config) + + try: + await collector.subscribe_to_dtype(ECHO_DTYPE, queue) + for value in range(count): + await pub_a.publish(value) + await pub_b.publish(value * 10) + return [ + (await asyncio.wait_for(queue.get(), COLLECT_TIMEOUT)).data + for _ in range(count) + ] + finally: + await pub_a.close() + await pub_b.close() + await collector.close() + + seen = asyncio.run(feed_and_collect()) + + # every output is one complete (a, b) pair; joins arrive in order + assert seen == [0, 11, 22], seen + + assert orbit_dt.twin_close(twin_id) == "closed" diff --git a/test/integration/twin_components.py b/test/integration/twin_components.py index 2276aa4..82b3eff 100644 --- a/test/integration/twin_components.py +++ b/test/integration/twin_components.py @@ -121,6 +121,15 @@ async def main_loop(self, runtime, in_data): await runtime.stream.publish(ECHO_DTYPE, in_data.data) +class JoinSink(UtilityTask): + """Terminal for a joined stream: republishes the member sum, so a + client can watch complete join sets arrive over the twin's stream.""" + + async def main_loop(self, runtime, in_data): + await runtime.stream.publish( + ECHO_DTYPE, sum(item.data for item in in_data.data)) + + class CrashingTask(UtilityTask): """Persistent component that dies -- the twin must land in `failed` with the reason visible in `twin_list`.""" From c12bd4e05c81d2911ae4340cc537dbca8a3f2689 Mon Sep 17 00:00:00 2001 From: Andre Merzky Date: Sun, 30 Aug 2026 23:05:56 +0200 Subject: [PATCH 3/4] review round: the binding is live when add_input answers Three review findings applied: - `runtime.add_input` returns its subscription task and the service verb awaits it, so the client's success means the binding is live -- a producer publishing right after the call no longer races the broker registration, and a subscribe failure reaches the caller instead of only a server-side log. - a re-bind that changes the codec is refused (ValueError -> 409): the stream client dedupes on (channel, dtype), so the change would be recorded in `describe()` yet never applied. - `ChannelPublisher.open` defaults to CLIENT_CONNECT_TIMEOUT like every other orbit connect path -- an external producer pointed at an unreachable broker fails in 30s instead of holding an uncancellable thread for the day-long None approximation, which stays as the backstop for an explicit None (and its timeout message now names the effective deadline, not "None"). Plus: the client docstring states the stream's real delivery contract, the join test asserts against `join_components` directly, and the None substitution is pinned at unit level. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_016Npyz3Hbnwos12ESsdJ2YU --- src/digitaltwin/runtime.py | 23 ++++++++++++++----- src/digitaltwin/service/client.py | 6 +++-- src/digitaltwin/service/session.py | 9 +++++++- src/digitaltwin/streaming.py | 5 ++++- src/digitaltwin/streaming_orbit.py | 15 +++++++------ test/unit/test_service.py | 36 +++++++++++++++++++++++++----- test/unit/test_streaming_orbit.py | 32 ++++++++++++++++++++++++++ 7 files changed, 105 insertions(+), 21 deletions(-) diff --git a/src/digitaltwin/runtime.py b/src/digitaltwin/runtime.py index f40a8cc..b456ee8 100644 --- a/src/digitaltwin/runtime.py +++ b/src/digitaltwin/runtime.py @@ -1070,6 +1070,10 @@ def add_input(self, dtype: DataType, channel: str, codec: str = CODEC_JSON): Internal producers keep their own path: a persistent component publishes through `RuntimeAPI.stream`. + + Returns the subscription task (`None` for an idempotent re-bind, + or when the twin is tearing down): a caller that needs the + binding live before producers publish awaits it. """ self._check_mutable() @@ -1077,15 +1081,24 @@ def add_input(self, dtype: DataType, channel: str, codec: str = CODEC_JSON): PubSubClient.check_channel(channel) check_codec(codec) - binding = _InputBinding(dtype, channel, codec) - if binding in self.inputs: - return - self.inputs.append(binding) + for existing in self.inputs: + if (existing.dtype, existing.channel) == (dtype, channel): + if existing.codec == codec: + return None + # the stream client dedupes on (channel, dtype), so a + # changed codec would be recorded here yet never applied + # -- refuse it rather than decode with the old one forever + raise ValueError( + f"channel {channel!r} is already bound to {dtype} with" + f" codec {existing.codec!r}; a binding cannot change" + " its codec" + ) + self.inputs.append(_InputBinding(dtype, channel, codec)) # subscribe now, so nothing published before start() is lost: the # queue buffers it and the consumers wait for start anyway logger.info(f"Bind channel {channel!r} ({codec}) to dtype: {dtype}") - self._to_asyncio_task( + return self._to_asyncio_task( self.streamer.subscribe_to_channel, channel, dtype, diff --git a/src/digitaltwin/service/client.py b/src/digitaltwin/service/client.py index 5885138..52b1281 100644 --- a/src/digitaltwin/service/client.py +++ b/src/digitaltwin/service/client.py @@ -250,9 +250,11 @@ def add_input( """Bind an external channel to one of the twin's input dtypes. The producer lives outside the framework and publishes to - `channel`; the twin receives every message on it, decoded by + `channel`; the twin receives the channel's traffic, decoded by `codec` (`json` for plain scripts and instruments, `raw` for - bytes, `cloudpickle` only inside one trust domain). + bytes, `cloudpickle` only inside one trust domain). Delivery is + the stream's contract: at most once, newest kept under pressure. + The call returns once the binding is live. """ return self._verb(twin_id, "add_input", dtype, channel, codec)["state"] diff --git a/src/digitaltwin/service/session.py b/src/digitaltwin/service/session.py index 996f1d6..398232d 100644 --- a/src/digitaltwin/service/session.py +++ b/src/digitaltwin/service/session.py @@ -455,7 +455,14 @@ async def _verb_add_input( # data-only arguments, no Package: the producer lives outside the # framework, only the binding crosses the wire. The runtime # validates channel and codec and subscribes at bind time. - twin.runtime.add_input(dtype, channel, codec) + subscribed = twin.runtime.add_input(dtype, channel, codec) + + # the verb answers only once the binding is live: a producer + # publishing right after the call must not race the broker + # registration, and a subscribe failure belongs to the caller, + # not to a server-side log alone + if subscribed is not None: + await subscribed async def _verb_add_data_join( self, twin: TwinInstance, join_dtype: JoinDataType diff --git a/src/digitaltwin/streaming.py b/src/digitaltwin/streaming.py index 48f98f8..ae03494 100644 --- a/src/digitaltwin/streaming.py +++ b/src/digitaltwin/streaming.py @@ -926,12 +926,15 @@ async def open( channel: str, codec: str = CODEC_JSON, config: Optional[PubSubConfig] = None, - timeout: Optional[float] = None, + timeout: Optional[float] = CLIENT_CONNECT_TIMEOUT, ) -> "ChannelPublisher": """Connect to a broker and publish to `channel` on it. `config` defaults to the configured broker (environment, else the loopback defaults); its namespace, if it has one, is ignored. + + The connect is bounded by `timeout` -- an external producer + pointed at an unreachable broker should fail fast, not hang. """ config = config or PubSubConfig.resolve() diff --git a/src/digitaltwin/streaming_orbit.py b/src/digitaltwin/streaming_orbit.py index f9e7af9..51fc098 100644 --- a/src/digitaltwin/streaming_orbit.py +++ b/src/digitaltwin/streaming_orbit.py @@ -164,13 +164,14 @@ def _start_runtime(self, timeout: Optional[float]) -> EndpointRuntime: runtime = EndpointRuntime(broker_url=self.broker_url, name=self.name, role="stream") + # ORBIT's start() takes a finite deadline only; the streaming + # contract's "None waits forever" is approximated by a day -- a + # participant that cannot register within that is stuck, not + # waiting. + effective = 86400.0 if timeout is None else timeout + try: - # ORBIT's start() takes a finite deadline only; the streaming - # contract's "None waits forever" is approximated by a day -- a - # participant that cannot register within that is stuck, not - # waiting. - runtime.start(wait=True, - timeout=86400.0 if timeout is None else timeout) + runtime.start(wait=True, timeout=effective) # start() returns *silently* when it merely timed out, so the # registration has to be checked explicitly -- otherwise an @@ -179,7 +180,7 @@ def _start_runtime(self, timeout: Optional[float]) -> EndpointRuntime: if not runtime.wait_registered(timeout=0): raise TimeoutError( f"ORBIT broker at {runtime.broker_url} did not register" - f" {self.name} within {timeout}s" + f" {self.name} within {effective}s" ) except BaseException: diff --git a/test/unit/test_service.py b/test/unit/test_service.py index 23571d5..e4204db 100644 --- a/test/unit/test_service.py +++ b/test/unit/test_service.py @@ -974,11 +974,37 @@ async def test_add_data_join_registers_through_the_verb(): await session.twin_call( "t1", "add_data_join", encode({"args": (joined,)}), version_stamp()) - registered = [ant.component.out_dtype - for ants in twin.runtime.components.values() - for ant in ants - if getattr(ant.component, "out_dtype", None) == joined] - assert registered, "join component not registered" + assert joined in twin.runtime.join_components assert twin.summary()["calls"] == {"add_data_join": 1} await twin.close() + + +async def test_add_input_refuses_a_codec_change_on_a_bound_channel(): + """The stream client dedupes on (channel, dtype): a re-bind with a + different codec would be recorded yet never applied, so it is + refused instead of silently decoding with the old codec forever.""" + + session = DTSession("s1") + twin = _running_twin(session, "t1") + twin.runtime.streamer = _BindingStream() + + x = DataType("x") + await session.twin_call( + "t1", "add_input", encode({"args": (x, "lab/raw")}), version_stamp()) + + # the identical re-bind is an idempotent no-op + await session.twin_call( + "t1", "add_input", encode({"args": (x, "lab/raw")}), version_stamp()) + assert len(twin.runtime.inputs) == 1 + + with pytest.raises(HTTPException) as raised: + await session.twin_call( + "t1", "add_input", + encode({"args": (x, "lab/raw", "raw")}), version_stamp()) + + assert raised.value.status_code == 409 + assert "codec" in raised.value.detail + assert len(twin.runtime.inputs) == 1 + + await twin.close() diff --git a/test/unit/test_streaming_orbit.py b/test/unit/test_streaming_orbit.py index f385a74..21bf415 100644 --- a/test/unit/test_streaming_orbit.py +++ b/test/unit/test_streaming_orbit.py @@ -518,3 +518,35 @@ async def test_participant_names_are_unique(): # ... unless the caller insists assert OrbitPubSubBackend(name="fixed").name == "fixed" + + +def test_a_none_connect_timeout_reaches_orbit_as_a_finite_deadline( + monkeypatch): + """ORBIT's `start()` takes a finite float only; the streaming + contract's `None` ("wait forever") must arrive as the documented + day-long approximation, not as `None` (`monotonic() + None` dies).""" + + recorded = {} + + class _Recording: + broker_url = "loopback://broker" + + def __init__(self, broker_url=None, name=None, role=None): + pass + + def start(self, wait=True, timeout=None): + recorded["timeout"] = timeout + + def wait_registered(self, timeout=0): + return True + + monkeypatch.setattr( + "digitaltwin.streaming_orbit.EndpointRuntime", _Recording) + + backend = OrbitPubSubBackend() + backend._start_runtime(None) + + assert recorded["timeout"] == 86400.0 + + backend._start_runtime(5.0) + assert recorded["timeout"] == 5.0 From 455b1f88a8913f923d69999ac32bd7878d9c0438 Mon Sep 17 00:00:00 2001 From: Andre Merzky Date: Sun, 30 Aug 2026 23:19:08 +0200 Subject: [PATCH 4/4] KISS round: the tests say what they pin, nothing more - the add_input unit test drops its settling sleep -- the verb answers only once the subscription is live, and the test now pins exactly that (with the sleep, a regression back to fire-and-forget would still pass) - local imports hoisted to the module blocks they belong in - the integration helper loses a parameter nothing passed - `add_input` carries its new return contract in the signature - a dead fake attribute removed Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_016Npyz3Hbnwos12ESsdJ2YU --- src/digitaltwin/runtime.py | 3 ++- test/integration/test_dt_orbit_stream.py | 9 ++++----- test/unit/test_service.py | 12 +++++++----- test/unit/test_streaming_orbit.py | 2 -- 4 files changed, 13 insertions(+), 13 deletions(-) diff --git a/src/digitaltwin/runtime.py b/src/digitaltwin/runtime.py index b456ee8..96f8ea4 100644 --- a/src/digitaltwin/runtime.py +++ b/src/digitaltwin/runtime.py @@ -1053,7 +1053,8 @@ def _ensure_dtype_queue(self, dtype: DataType) -> asyncio.Queue: return self.dtype_queues[dtype] - def add_input(self, dtype: DataType, channel: str, codec: str = CODEC_JSON): + def add_input(self, dtype: DataType, channel: str, + codec: str = CODEC_JSON) -> Optional[asyncio.Task]: """Open the graph at its input edge: bind an external channel. Sensors and other producers live outside the framework. They diff --git a/test/integration/test_dt_orbit_stream.py b/test/integration/test_dt_orbit_stream.py index d089d62..a3a4fb3 100644 --- a/test/integration/test_dt_orbit_stream.py +++ b/test/integration/test_dt_orbit_stream.py @@ -34,6 +34,7 @@ ECHO_DTYPE, INFERENCE_DTYPE, SENSOR_DTYPE, + JoinSink, OffsetModel, ) @@ -302,8 +303,6 @@ def test_external_channels_join_through_the_client_verbs(orbit_dt, twin_id): producers are plain `ChannelPublisher`s -- outside the framework, knowing nothing about twins.""" - from twin_components import JoinSink - a = DataType("chan-a") b = DataType("chan-b") joined = JoinDataType([a, b]) @@ -318,7 +317,7 @@ def test_external_channels_join_through_the_client_verbs(orbit_dt, twin_id): orbit_dt.add_task(twin_id, orbit_dt.package(JoinSink), joined, NULL_DTYPE) assert orbit_dt.start(twin_id) == "running" - async def feed_and_collect(count=3): + async def feed_and_collect(): config = PubSubConfig(kind=BACKEND_ORBIT, broker_url=ORBIT_BROKER_URL) collector = await connect_stream_client( twin_id, backend=BACKEND_ORBIT, broker_url=ORBIT_BROKER_URL) @@ -328,12 +327,12 @@ async def feed_and_collect(count=3): try: await collector.subscribe_to_dtype(ECHO_DTYPE, queue) - for value in range(count): + for value in range(3): await pub_a.publish(value) await pub_b.publish(value * 10) return [ (await asyncio.wait_for(queue.get(), COLLECT_TIMEOUT)).data - for _ in range(count) + for _ in range(3) ] finally: await pub_a.close() diff --git a/test/unit/test_service.py b/test/unit/test_service.py index e4204db..4d3a814 100644 --- a/test/unit/test_service.py +++ b/test/unit/test_service.py @@ -16,7 +16,12 @@ from fastapi import FastAPI, HTTPException # noqa: E402 from starlette.testclient import TestClient # noqa: E402 -from digitaltwin.components import TRUTHY, DataType, UtilityTask # noqa: E402 +from digitaltwin.components import ( # noqa: E402 + TRUTHY, + DataType, + JoinDataType, + UtilityTask, +) from digitaltwin.runtime import DTRuntime # noqa: E402 from digitaltwin.service.plugin import UI_ASSETS, PluginDT # noqa: E402 from digitaltwin.service.session import DTSession, TwinInstance # noqa: E402 @@ -934,8 +939,7 @@ async def test_add_input_binds_a_channel_through_the_verb(): binding = twin.runtime.inputs[0] assert (binding.dtype, binding.channel, binding.codec) == (x, "lab/raw", "json") - # subscribed at bind time, so nothing published before start() is lost - await asyncio.sleep(0) + # the verb answered, so the subscription is already live -- no settling assert twin.runtime.streamer.subscribed == [("lab/raw", x, "json")] assert twin.summary()["calls"] == {"add_input": 1} @@ -964,8 +968,6 @@ async def test_add_input_rejects_a_bad_channel_as_client_error(): async def test_add_data_join_registers_through_the_verb(): """One joined dtype, consumable downstream like any other.""" - from digitaltwin.components import JoinDataType - session = DTSession("s1") twin = _running_twin(session, "t1") diff --git a/test/unit/test_streaming_orbit.py b/test/unit/test_streaming_orbit.py index 21bf415..631c28b 100644 --- a/test/unit/test_streaming_orbit.py +++ b/test/unit/test_streaming_orbit.py @@ -529,8 +529,6 @@ def test_a_none_connect_timeout_reaches_orbit_as_a_finite_deadline( recorded = {} class _Recording: - broker_url = "loopback://broker" - def __init__(self, broker_url=None, name=None, role=None): pass