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
.
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-inTimeoutError
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 provideget_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
andbroadcast
-
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 provideget_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)