Skip to content

IstosTestClient API Reference

In-process testing without a live Zenoh network.

In-process test client for handlers, streams and subscribers — no network.

IstosTestClient

Invoke handlers, streams and subscribers in-process. The authorizer gate, validation, DI and durability ledger all still run.

istos = Istos()

@istos.handle("robot/move")
async def move(distance: int):
    return {"moved": distance}

client = IstosTestClient(istos)
assert await client.query("robot/move", distance=10) == {"moved": 10}

Pass token= to drive an authorizer; a denied request raises UnauthorizedError.

Source code in src/istos/testing/testclient.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
class IstosTestClient:
    """
    Invoke handlers, streams and subscribers in-process. The authorizer gate,
    validation, DI and durability ledger all still run.

        istos = Istos()

        @istos.handle("robot/move")
        async def move(distance: int):
            return {"moved": distance}

        client = IstosTestClient(istos)
        assert await client.query("robot/move", distance=10) == {"moved": 10}

    Pass ``token=`` to drive an authorizer; a denied request raises
    ``UnauthorizedError``.
    """

    def __init__(self, app: Istos) -> None:
        self.app = app

    def _find_handler(self, prefix: str) -> Any:
        for handler in self.app._handlers:
            if handler.prefix == prefix:
                return handler
        raise KeyError(f"No handler registered for prefix: {prefix!r}")

    def _find_stream(self, prefix: str) -> Any:
        for wrapper in self.app._streams:
            if wrapper.prefix == prefix:
                return wrapper
        raise KeyError(f"No stream registered for prefix: {prefix!r}")

    def _find_subscribers(self, prefix: str) -> List[Any]:
        return [s for s in self.app._subscribers if s.prefix == prefix]

    def _find_channel(self, prefix: str) -> Any:
        for wrapper in self.app._channels:
            if wrapper.prefix == prefix:
                return wrapper
        raise KeyError(f"No channel registered for prefix: {prefix!r}")

    async def _gate(self, wrapper: Any, prefix: str, params: dict, token: Optional[str]) -> None:
        """Run the authorizer and set a request context so the body can inject
        the resolved principal/token. A denial raises ``UnauthorizedError``."""
        attachment = RequestEnvelope(token=token).to_attachment() if token is not None else None
        principal = await check_authorized(
            getattr(wrapper, "_authorizer", None),
            AuthContext(prefix=prefix, key_expr=prefix, params=params, attachment=attachment),
        )
        set_request_context(RequestContext(
            prefix=prefix, principal=principal, attachment=attachment,
        ))

    async def query(self, prefix: str, token: Optional[str] = None, **kwargs: Any) -> Any:
        """Invoke a handler in-process."""
        handler = self._find_handler(prefix)
        # db / Depends(...) are injected by the handler, not validated here.
        skip = getattr(handler, "_injected_params", None)
        validated = validate_params(handler.func, kwargs, skip_params=skip)
        validated.pop("db", None)
        validated.pop("session", None)
        await self._gate(handler, prefix, validated, token)
        return await handler(**validated)

    async def stream(
        self, prefix: str, token: Optional[str] = None, **kwargs: Any
    ) -> AsyncIterator[Any]:
        """Consume a ``@stream`` handler in-process, yielding each chunk.

            async for chunk in client.stream("llm/generate", prompt="hi"):
                ...
        """
        wrapper = self._find_stream(prefix)
        skip = getattr(wrapper, "_injected_params", None)
        validated = validate_params(wrapper.func, kwargs, skip_params=skip)
        validated.pop("db", None)
        await self._gate(wrapper, prefix, validated, token)

        async with AsyncExitStack() as di_stack:
            call_kwargs = dict(validated)
            if getattr(wrapper, "_has_depends", False):
                call_kwargs = await resolve_dependencies(
                    wrapper.func, call_kwargs, di_stack, cache={},
                    overrides=getattr(wrapper, "_dependency_overrides", {}),
                )
            agen = wrapper.func(**call_kwargs)
            try:
                async for chunk in agen:
                    yield chunk
            finally:
                if hasattr(agen, "aclose"):
                    await agen.aclose()

    async def publish(self, prefix: str, data: Any) -> None:
        """Deliver data to all matching subscribers in-process."""
        subscribers = self._find_subscribers(prefix)
        if not subscribers:
            raise KeyError(f"No subscribers registered for prefix: {prefix!r}")
        for sub in subscribers:
            await sub(data)

    def channel(
        self, prefix: str, token: Optional[str] = None,
        conversation_id: Optional[str] = None, **params: Any,
    ) -> "_TestChannel":
        """Open a ``@channel`` in-process and return a duplex handle. Use it as an
        async context manager; ``send`` / ``receive`` / ``async for`` from the
        caller's side while the handler runs::

            async with client.channel("agent/chat") as chan:
                await chan.send("hi")
                assert await chan.receive() == {"role": "system", "text": "ready"}

        The authorizer, validation and DI run exactly as on the WebSocket/fabric
        transport; a denied ``token`` raises ``UnauthorizedError`` on enter.
        """
        wrapper = self._find_channel(prefix)
        return _TestChannel(wrapper, token=token, conversation_id=conversation_id, params=params)

    def run_query(self, prefix: str, **kwargs: Any) -> Any:
        """Synchronous wrapper around query()."""
        return asyncio.run(self.query(prefix, **kwargs))

    def run_publish(self, prefix: str, data: Any) -> None:
        """Synchronous wrapper around publish()."""
        return asyncio.run(self.publish(prefix, data))

channel(prefix, token=None, conversation_id=None, **params)

Open a @channel in-process and return a duplex handle. Use it as an async context manager; send / receive / async for from the caller's side while the handler runs::

async with client.channel("agent/chat") as chan:
    await chan.send("hi")
    assert await chan.receive() == {"role": "system", "text": "ready"}

The authorizer, validation and DI run exactly as on the WebSocket/fabric transport; a denied token raises UnauthorizedError on enter.

Source code in src/istos/testing/testclient.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def channel(
    self, prefix: str, token: Optional[str] = None,
    conversation_id: Optional[str] = None, **params: Any,
) -> "_TestChannel":
    """Open a ``@channel`` in-process and return a duplex handle. Use it as an
    async context manager; ``send`` / ``receive`` / ``async for`` from the
    caller's side while the handler runs::

        async with client.channel("agent/chat") as chan:
            await chan.send("hi")
            assert await chan.receive() == {"role": "system", "text": "ready"}

    The authorizer, validation and DI run exactly as on the WebSocket/fabric
    transport; a denied ``token`` raises ``UnauthorizedError`` on enter.
    """
    wrapper = self._find_channel(prefix)
    return _TestChannel(wrapper, token=token, conversation_id=conversation_id, params=params)

publish(prefix, data) async

Deliver data to all matching subscribers in-process.

Source code in src/istos/testing/testclient.py 
111
112
113
114
115
116
117
async def publish(self, prefix: str, data: Any) -> None:
    """Deliver data to all matching subscribers in-process."""
    subscribers = self._find_subscribers(prefix)
    if not subscribers:
        raise KeyError(f"No subscribers registered for prefix: {prefix!r}")
    for sub in subscribers:
        await sub(data)

query(prefix, token=None, **kwargs) async

Invoke a handler in-process.

Source code in src/istos/testing/testclient.py 
71
72
73
74
75
76
77
78
79
80
async def query(self, prefix: str, token: Optional[str] = None, **kwargs: Any) -> Any:
    """Invoke a handler in-process."""
    handler = self._find_handler(prefix)
    # db / Depends(...) are injected by the handler, not validated here.
    skip = getattr(handler, "_injected_params", None)
    validated = validate_params(handler.func, kwargs, skip_params=skip)
    validated.pop("db", None)
    validated.pop("session", None)
    await self._gate(handler, prefix, validated, token)
    return await handler(**validated)

run_publish(prefix, data)

Synchronous wrapper around publish().

Source code in src/istos/testing/testclient.py 
141
142
143
def run_publish(self, prefix: str, data: Any) -> None:
    """Synchronous wrapper around publish()."""
    return asyncio.run(self.publish(prefix, data))

run_query(prefix, **kwargs)

Synchronous wrapper around query().

Source code in src/istos/testing/testclient.py 
137
138
139
def run_query(self, prefix: str, **kwargs: Any) -> Any:
    """Synchronous wrapper around query()."""
    return asyncio.run(self.query(prefix, **kwargs))

stream(prefix, token=None, **kwargs) async

Consume a @stream handler in-process, yielding each chunk.

async for chunk in client.stream("llm/generate", prompt="hi"): ...

Source code in src/istos/testing/testclient.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
async def stream(
    self, prefix: str, token: Optional[str] = None, **kwargs: Any
) -> AsyncIterator[Any]:
    """Consume a ``@stream`` handler in-process, yielding each chunk.

        async for chunk in client.stream("llm/generate", prompt="hi"):
            ...
    """
    wrapper = self._find_stream(prefix)
    skip = getattr(wrapper, "_injected_params", None)
    validated = validate_params(wrapper.func, kwargs, skip_params=skip)
    validated.pop("db", None)
    await self._gate(wrapper, prefix, validated, token)

    async with AsyncExitStack() as di_stack:
        call_kwargs = dict(validated)
        if getattr(wrapper, "_has_depends", False):
            call_kwargs = await resolve_dependencies(
                wrapper.func, call_kwargs, di_stack, cache={},
                overrides=getattr(wrapper, "_dependency_overrides", {}),
            )
        agen = wrapper.func(**call_kwargs)
        try:
            async for chunk in agen:
                yield chunk
        finally:
            if hasattr(agen, "aclose"):
                await agen.aclose()