tests.server package

Submodules

tests.server.config module

Server configuration.

class tests.server.config.ServerConfig(file_path: str)

Bases: object

Loads server configuration.

Attributes:

host: The host address for the server to run on. port: The port for the server to run on.

host: str

port: str

tests.server.errors module

Error messages.

tests.server.listens module

class tests.server.listens.Listens

Bases: object

A class that manages listeners and notifies them of changes in status.

Attributes:

_listeners (list[callable]): A list of callback functions to be executed when a change in status occurs. _notify_listeners_task (Task): An asyncio task that executes the coroutines of the listeners concurrently. status (RawListens): An instance of the RawListens class representing the current status. _count (int): A counter to keep track of the number of listeners.

Methods:

__init__(): Initializes the Listens object. _notify_listeners(new_status): Notifies all listeners of a change in status. add_listener(callback): Adds a new listener to the list of listeners. stop_listening(): Removes all listeners from the list.

add_listener(callback: callable) → None

Adds a new listener to the list of listeners.

Args:

callback: The callback function to be added as a listener.

status: RawListens

A descriptor class for accessing and updating the status of a Listens object.

This class provides the __get__ and __set__ methods to get and set the status of a Listens object. When the status is updated, it notifies the listeners asynchronously.

Attributes:

None

Methods:

__get__(self, obj: Listens, objtype=None) -> Any:

Get the status of the Listens object.

__set__(self, obj: Listens, value) -> None:

Set the status of the Listens object and notify the listeners.

stop_listening() → None

Removes all listeners from the list.

class tests.server.listens.RawListens

Bases: object

A descriptor class for accessing and updating the status of a Listens object.

This class provides the __get__ and __set__ methods to get and set the status of a Listens object. When the status is updated, it notifies the listeners asynchronously.

Attributes:

None

Methods:

__get__(self, obj: Listens, objtype=None) -> Any:

Get the status of the Listens object.

__set__(self, obj: Listens, value) -> None:

Set the status of the Listens object and notify the listeners.

tests.server.logger module

tests.server.manager module

The middleman between Plover and the server.

class tests.server.manager.EngineServerManager

Bases: object

Manages a server that exposes the Plover engine.

add_listener(listener: callable) → None

Adds a listener to the server.

Args:

listener (callable): The listener function to be added.

get_server_status() → ServerStatus

Gets the status of the server.

Returns:

The status of the server.

join() → None

Waits for the server to finish processing all requests and stops the server.

start() → None

Starts the server.

Raises:

AssertionError: The server failed to start. IOError: The server failed to start.

stop() → None

Stops the server.

Raises:

AssertionError: The server failed to stop.

stop_listening() → None

Stops the server from listening for status changes.

tests.server.server module

Core engine server definitions.

class tests.server.server.EngineServer(host: str, port: str)

Bases: object

A server example.

This class represents a server that can be started and stopped. It provides methods to queue messages for broadcasting and register callbacks to handle received messages.

Attributes:

listened: An instance of the Listens class representing the listeners for the server. data: An instance of the Listens class representing the data received by the server.

data: Listens = <tests.server.listens.Listens object>

join() → None

Function to stop the underlying thread.

listened: Listens = <tests.server.listens.Listens object>

queue_message(data: dict) → Future

Queues a message for the server to broadcast.

Assumes it is called from a thread different from the event loop.

Args:

data: The data in JSON format to broadcast.

queue_stop() → None

Queues the server to stop.

Assumes it is called from a thread different from the event loop.

register_message_callback(callback) → None

Registers a callback function to handle received messages.

Args:

callback: The callback function to register.

start() → None

Function to start the server (external thread).

class tests.server.server.ServerStatus(value, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Represents the status of the server.

Attributes:

Stopped: The server is stopped. Running: The server is running.

Running = 2

Stopped = 1

Module contents