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