unitelabs.bus.testing
Packages
Attributes
- Name
Stub- Type
- Value
= typing.Callable[[bytes], typing.Optional[bytes]]
- Description
- Name
__all__- Type
- Value
= [ "DeviceMock", "SerialDeviceMock", "create_mock_connection", "MockTransport", "Stub", "stub_maker" ]
- Description
Functions
create_mock_connection(
- protocol_factory : typing.Callable[..., asyncio.Protocol],
- stubs : typing.Optional[list[Stub]],
- **kwargs
Parameters
- Name
protocol_factory- Type
- typing.Callable[..., asyncio.Protocol]
- Default
- Description
- Name
stubs- Type
- typing.Optional[list[Stub]]
- Default
- = None
- Description
- Name
**kwargs- Type
- Default
- = {}
- Description
Response
- Type
- typing.Tuple[Transport, asyncio.Protocol]
- Description
Classes
DeviceMock
Abstract base class to mock devices (hardware or software components) for testing purposes.
- Bases
- contextlib.AbstractContextManager
Methods
__init__(self, stubs : typing.Optional[list[Stub]]) -> None
Parameters
- Name
self- Type
- Default
- Description
- Name
stubs- Type
- typing.Optional[list[Stub]]
- Default
- = None
- Description
An optional list of stub methods that receive the same request as the device and optionally return the corresponding response. Stubs can be modified later on by directly modifying the `mock.stubs` list. Stubs at the end of the list have a higher priority and overwrite responses from the stubs at the beginning of the list.
get_response(self, request : bytes) -> typing.Optional[bytes]
Returns the stubbed response for a given request. Responses from stubs at the end of the list have a higher priority and overwrite responses from the stubs at the beginning of the list.
Parameters
- Name
self- Type
- Default
- Description
- Name
request- Type
- bytes
- Default
- Description
The request sent to the hardware.
Response
- Type
- typing.Optional[bytes]
- Description
The response corresponding to the received request.
Attributes
- Name
stubs- Type
- list[Stub]
- Value
= None
- Description
- Name
logger- Type
- logging.Logger
- Value
= None
- Description
A standard python logger available to debug device mocks.
MockTransport
Mocks `Transport` functionality. Must provide either `stubs` or set `autorespond=True` to get responses.
- MRO
- ⎿⎽⎽⎽Transport
- ⎿⎽⎽⎽⎽MockTransport
Methods
__init__(
- self,
- stubs : typing.Optional[list[Stub]],
- device : typing.Optional[DeviceMock],
- autorespond : bool,
- **kwargs
Parameters
- Name
self- Type
- Default
- Description
- Name
stubs- Type
- typing.Optional[list[Stub]]
- Default
- = None
- Description
List of callable functions that return responses based matching to requests.
- Name
device- Type
- typing.Optional[DeviceMock]
- Default
- = None
- Description
A `DeviceMock` through which device communication will be mocked.
- Name
autorespond- Type
- bool
- Default
- = False
- Description
Whether to return request bytes without checking `stubs`, i.e. loopback responses.
- Name
**kwargs- Type
- Default
- = {}
- Description
Any additional `Transport` init kwargs.
write(self, data : bytes) -> None
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- bytes
- Default
- Description
_write(self, data : bytes) -> int
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- bytes
- Default
- Description
Response
- Type
- int
- Description
_read(self) -> typing.Optional[bytes]
Response
- Type
- typing.Optional[bytes]
- Description
_open(self) -> None
_close(self) -> None
_ensure_reader(self) -> None
_remove_reader(self) -> None
_ensure_writer(self) -> None
_remove_writer(self) -> None
open(self) -> None
Opens the transport and sets state to allow future read operations.
close(self) -> None
Closes the transport and sets state to disallow further read operations.
is_reading(self) -> bool
Whether or not reading operations are currently being performed.
Response
- Type
- bool
- Description
True if the transport is receiving, otherwise False.
pause_reading(self) -> None
Pause the receiving end. No data will be passed to the protocol's `data_received()` method until `resume_reading()` is called.
resume_reading(self) -> None
Resume the receiving end. Data received will once again be passed to the protocol's `data_received()` method.
_abort(self, exception : typing.Optional[Exception]) -> None
Closes the transport immediately and updates state to disable further read operations.
Parameters
- Name
self- Type
- Default
- Description
- Name
exception- Type
- typing.Optional[Exception]
- Default
- = None
- Description
The Exception to propagate to the protocol when aborting, if connected.
_safe_read(self) -> None
Safely and asynchronously read data from the transport.
read_all(self) -> bytes
Read all available data from the transport. Repeatedly calls `_read` and aggregates the results until no further data is available.
Response
- Type
- bytes
- Description
All available data from the transport.
clear_read_buffer(self) -> None
Clears out all available read data without notifying the protocol. Calls `read_all` and throws away the result.
get_write_buffer_size(self) -> int
Calculate the current size of the write buffer.
Response
- Type
- int
- Description
The number of bytes in the write buffer.
get_write_buffer_limits(self) -> tuple[int, int]
Get the high and low watermarks for write flow control.
Response
- Type
- tuple[int, int]
- Description
a tuple (low, high) where low and high are positive number of bytes.
set_write_buffer_limits(self, high : typing.Optional[int], low : typing.Optional[int]) -> None
Set the high- and low-water limits for write flow control. These two values control when to call the protocol's `pause_writing()` and `resume_writing()` methods. If specified, the low-water limit must be less than or equal to the high-water limit. Neither value can be negative. The defaults are implementation-specific. If only the high-water limit is given, the low-water limit defaults to an implementation-specific value less than or equal to the high-water limit. Setting high to zero forces low to zero as well, and causes `pause_writing()` to be called whenever the buffer becomes non-empty. Setting low to zero causes `resume_writing()` to be called only once the buffer is empty. Use of zero for either limit is generally sub-optimal as it reduces opportunities for doing I/O and computation concurrently.
Parameters
- Name
self- Type
- Default
- Description
- Name
high- Type
- typing.Optional[int]
- Default
- = None
- Description
The maximum allowed number of bytes in the write buffer.
- Name
low- Type
- typing.Optional[int]
- Default
- = None
- Description
The minimum allowed number of bytes in the write buffer.
can_write_eof(self) -> bool
Whether or not this transport has implemented `write_eof()` method.
Response
- Type
- bool
- Description
True if this transport supports `write_eof()`, False if not.
write_eof(self) -> None
Close the write with end-of-file after flushing buffered data. (This is like typing ^D into a UNIX program reading from stdin.) Data may still be received.
writelines(self, list_of_data : typing.Iterable[typing.Union[bytes, bytearray, memoryview]]) -> None
Write a list (or any iterable) of data bytes to the transport. The default implementation concatenates the arguments and calls `write()` on the result.
Parameters
- Name
self- Type
- Default
- Description
- Name
list_of_data- Type
- typing.Iterable[typing.Union[bytes, bytearray, memoryview]]
- Default
- Description
The list of bytes to concatenate and write to the Transport.
flush(self) -> None
Flush the write buffer and disable further writing.
_safe_write(self) -> None
Asynchronously write buffered data. This method is called back asynchronously as a writer registered with the asyncio event-loop against the underlying file descriptor for the serial port. If this method is invoked while the transport is closing, and the write-buffer is then emptied by this method, the protocol's `connection_lost()` method will be called with None as its argument.
_maybe_pause_writing(self) -> None
To be called whenever the write-buffer size increases. Tests the current write-buffer size against the high water mark configured for this transport. If the high water mark is exceeded, the `Protocol` is instructed to `pause_writing()`.
_maybe_resume_protocol(self) -> None
To be called whenever the write-buffer size decreases. Tests the current write-buffer size against the low water mark configured for this transport. If writing is currently paused and the write-buffer size is below the low water mark, the `Protocol` is instructed to `resume_writing()`.
_set_write_buffer_limits(self, low : typing.Optional[int], high : typing.Optional[int]) -> None
Set the high- and low-water limits for write flow control. By default, the high-water limit is 4 times the high-water limit and if neither is specified, (16384, 65536).
Parameters
- Name
self- Type
- Default
- Description
- Name
low- Type
- typing.Optional[int]
- Default
- = None
- Description
The low-water limit for write flow control.
- Name
high- Type
- typing.Optional[int]
- Default
- = None
- Description
The high-water limit for write flow control.
is_closing(self) -> bool
Whether the transport is closing or closed.
Response
- Type
- bool
- Description
True if the transport is closing or closed, False otherwise.
abort(self) -> None
Close the transport immediately.
_exception(
- self,
- exception : Exception,
- message : str
Report a fatal error to the event-loop and abort the transport.
Parameters
- Name
self- Type
- Default
- Description
- Name
exception- Type
- Exception
- Default
- Description
The Exception to pass on the the loop's exception handler.
- Name
message- Type
- str
- Default
- Description
Human-readable text describing the exception's execution state, cause, etc.
Attributes
- Name
stubs- Type
- Value
= stubs or []
- Description
- Name
device- Type
- Value
= device
- Description
- Name
autorespond- Type
- Value
= autorespond
- Description
- Name
queue- Type
- Value
= queue.Queue()
- Description
- Name
mock- Type
- Value
= unittest.mock.Mock()
- Description
- Name
kwargs- Type
- Value
= kwargs
- Description
- Name
_has_reader- Type
- Value
= False
- Description
- Name
_has_writer- Type
- Value
= False
- Description
- Name
_is_writing_paused- Type
- Value
= False
- Description
- Name
_write_buffer- Type
- list[typing.Union[bytes, bytearray, memoryview]]
- Value
= []
- Description
- Name
writes_pending- Type
- bool
- Value
= None
- Description
Whether or not there is data in the write buffer waiting to be written.
- Name
_loop- Type
- Value
= asyncio.get_event_loop_policy().get_event_loop()
- Description
- Name
_protocol- Type
- typing.Optional[P_co]
- Value
= None
- Description
- Name
_is_closing- Type
- Value
= True
- Description
SerialDeviceMock
A device mock implementation that simulates a serial connection.
Methods
__enter__(self) -> None
__exit__(
- self,
- exc_type,
- exc_val,
- exc_tb
Parameters
- Name
self- Type
- Default
- Description
- Name
exc_type- Type
- Default
- Description
- Name
exc_val- Type
- Default
- Description
- Name
exc_tb- Type
- Default
- Description
open(self) -> None
close(self) -> None
receive(self) -> None
received(self, data : bytes) -> None
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- bytes
- Default
- Description
get_response(self, request : bytes) -> typing.Optional[bytes]
Returns the stubbed response for a given request. Responses from stubs at the end of the list have a higher priority and overwrite responses from the stubs at the beginning of the list.
Parameters
- Name
self- Type
- Default
- Description
- Name
request- Type
- bytes
- Default
- Description
The request sent to the hardware.
Response
- Type
- typing.Optional[bytes]
- Description
The response corresponding to the received request.
Attributes
- Name
cancel- Type
- Value
= threading.Event()
- Description
- Name
reader- Type
- Value
= threading.Thread(target=self.receive, daemon=True)
- Description
- Name
port- Type
- str
- Value
= None
- Description
- Name
stubs- Type
- list[Stub]
- Value
= None
- Description
- Name
logger- Type
- logging.Logger
- Value
= None
- Description
A standard python logger available to debug device mocks.