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)