Testing¶

Warning

This API is experimental and subject to breaking changes.

Tests for the lahja library can be written using the Runner/Engine/Driver APIs. These allow for constructing reusable declarative tests against endpoints which can be run against different endpoint implementations as well as different configurations of endpoints.

Runner¶

Runners are in charge of the outermost execution layer. A Runner must be a callable which accepts *args where each argument is a Driver.

class lahja.tools.runner.RunnerAPI¶: Bases: abc.ABC

Engines¶

Engines are in charge of abstracting away how each individual endpoint implementation should be run. An Engine must implement the following API.

class lahja.tools.engine.EngineAPI¶

Bases: abc.ABC

run_drivers(*drivers) → Awaitable[None]¶: Performs the actual running of the drivers executing them with in a manner appropriate for the individual endpoint implementation.

run_with_timeout(coro: Callable[..., Awaitable[Any]], *args, timeout: int) → None¶: Runs a coroutine with the specifid positional args with a timeout. must raise the built-in TimeoutError when a timeout occurs.

sleep(seconds: float) → None¶: Sleep for the provide number of seconds in a manner appropriate for the individual endpoint implementation.

Drivers¶

Drivers are a declarative set of instructions for instrumenting the actions and lifecycle of an endpoint. A driver must be a coroutine which takes an Engine as a single argument and performs the actions declared by the driver.

Drivers should be constructed in a functional maner using the utilities provided under lahja.tools.drivers.

A driver is composed of a single Initializer followed by a variadic number of Actions.

lahja.tools.drivers.driver.driver(initializer: lahja.tools.drivers.initializers.Initializer, *actions) → Callable[lahja.tools.engine.EngineAPI, Awaitable[None]]¶: Construct a Driver. Should contain a single Initializer followed by a variadic number of Actions.

Initializers¶

lahja.tools.drivers.initializers.serve_endpoint(config: lahja.common.ConnectionConfig) → lahja.tools.drivers.initializers.Initializer¶

lahja.tools.drivers.initializers.run_endpoint(name: str) → lahja.tools.drivers.initializers.Initializer¶

Actions¶

lahja.tools.drivers.actions.broadcast(event: lahja.common.BaseEvent, config: Optional[lahja.common.BroadcastConfig] = None) → lahja.tools.drivers.actions.AsyncAction¶: See EndpointAPI.broadcast

lahja.tools.drivers.actions.connect_to_endpoints(*configs) → lahja.tools.drivers.actions.AsyncAction¶: See EndpointAPI.connect_to_endpoints

lahja.tools.drivers.actions.throws(action: Union[lahja.tools.drivers.actions.SyncAction, lahja.tools.drivers.actions.AsyncAction], exc_type: Type[Exception]) → Union[lahja.tools.drivers.actions.SyncAction, lahja.tools.drivers.actions.AsyncAction]¶: Checks that the provided Action throws the provided exception type.

lahja.tools.drivers.actions.wait_for(event_type: Type[lahja.common.BaseEvent], on_event: Optional[Callable[[lahja.base.EndpointAPI, lahja.common.BaseEvent], Any]] = None) → lahja.tools.drivers.actions.AsyncAction¶: Wait for an event of the provided request_type and call response event returned by the provide get_response function.

lahja.tools.drivers.actions.wait_until_any_endpoint_subscribed_to(event_type: Type[lahja.common.BaseEvent]) → lahja.tools.drivers.actions.AsyncAction¶: See EndpointAPI.wait_until_any_endpoint_subscribed_to

lahja.tools.drivers.actions.wait_until_connected_to(name: str) → lahja.tools.drivers.actions.AsyncAction¶: See EndpointAPI.wait_until_connected_to

lahja.tools.drivers.actions.wait_any_then_broadcast(event: lahja.common.BaseEvent, config: Optional[lahja.common.BroadcastConfig] = None) → lahja.tools.drivers.actions.AsyncAction¶: Combination of wait_until_any_endpoint_subscribed_to and broadcast

lahja.tools.drivers.actions.serve_request(request_type: Type[lahja.common.BaseRequestResponseEvent[lahja.common.BaseEvent]], get_response: Callable[[lahja.base.EndpointAPI, lahja.common.BaseRequestResponseEvent[lahja.common.BaseEvent]], lahja.common.BaseEvent]) → lahja.tools.drivers.actions.AsyncAction¶: Wait for an event of the provided request_type and respond using the response event returned by the provide get_response function.

lahja.tools.drivers.actions.request(event: lahja.common.BaseRequestResponseEvent[lahja.common.BaseEvent], config: Optional[lahja.common.BroadcastConfig] = None, on_response: Optional[Callable[[lahja.base.EndpointAPI, lahja.common.BaseEvent], Any]] = None) → lahja.tools.drivers.actions.AsyncAction¶

See EndpointAPI.connect_to_endpoints

Optionally provide a callback on_response that will be run upon receipt of the response.

lahja.tools.drivers.actions.checkpoint(name: str) → Tuple[lahja.tools.drivers.actions.AsyncAction, lahja.tools.drivers.actions.AsyncAction]¶: Generates a pair of actions that can be used in separate drivers to synchronize their action execution. Each driver will wait until this checkpoint has been hit before proceeding.

Examples¶

Driver to run an endpoint as a server and wait for a client to connect.

from lahja.tools import drivers as d

server_driver = d.driver(
    d.serve_endpoint(ConnectionConfig(...)),
    d.wait_until_connected_to('client'),
)

Driver to run a client and connect to a server.

from lahja.tools import drivers as d

server_config = ConnectionConfig(...)
client_driver = d.driver(
    d.run_endpoint(ConnectionConfig(...)),
    d.connect_to_endpoints(server_config),
)

We could then run these together against the trio implementation of the endpoint like this.

from lahja.tools.runners import TrioRunner

client_driver = ...
server_driver = ...
runner = TrioRunner()
runner(client_driver, server_driver)