esptest.adapter.port.base_port module

exception esptest.adapter.port.base_port.ExpectTimeout(message, data_in_buffer=b'')

Bases: TimeoutError

raise same ExpectTimeout rather than different Exception from different framework

Parameters:

• message (str)

• data_in_buffer (str | bytes)

Return type:

None

class esptest.adapter.port.base_port.RawPort

Bases: object

Define a minimum Dut class, the dut objects should at least support these methods

the dut should at least support these attributes: - method: write_bytes() with parameters: data[bytes] - method: read_bytes() with parameters: timeout[float]

optional attribute & method: - attribute: name with type str - attribute: read_timeout with type float

write_bytes(data)

write bytes

Parameters:

data (bytes)

Return type:

None

read_bytes(timeout=0)

blocking read bytes

Parameters:

timeout (float)

Return type:

bytes

class esptest.adapter.port.base_port.PortSpawn(raw_port, name='', log_file=None, timeout=30, **kwargs)

Bases: SpawnBase, Generic[T]

Create a new class for pexpect with port read()/write() method.

There’s some reason that we can not use pyserial with pexpect.fdpexpect directly:

• pyserial do not support fileno in windows.

• Pexpect only read from serial during expect() method.

• Can not read more than 4K data at once, the data may be lost if it is not read in time:

• https://stackoverflow.com/questions/2415074/serial-port-not-able-to-write-big-chunk-of-data

Parameters:

• raw_port (T)

• name (str)

• log_file (str | None)

• timeout (float)

• kwargs (Any)

DEFAULT_READ_INTERVAL = 0.005

property receive_callback: Callable[[str, bytes], None] | None

property raw_port: T

property read_timeout: float

property data_cache: str

write(data)

Parameters:

data (AnyStr)

Return type:

None

read_nonblocking(size=1, timeout=None)

This method was used during expect(), reads data from serial output data cache.

If the data cache is not empty, it will return immediately. Otherwise, waiting for new data.

Parameters:

• size (int, optional) – maximum size of returning data. Defaults to 1.

• timeout (t.Union[int, float], optional) – maximum block time waiting for new data.

Returns:

new serial output data.

Return type:

bytes

stop()

Return type:

None

close()

Stop and clean up

Return type:

None

esptest.adapter.port.base_port.handle_expect_timeout(func)

Raise same type exception ExpectTimeout for ports from different frameworks

Parameters:

func (Callable)

Return type:

Callable

class esptest.adapter.port.base_port.BasePort(raw_port, name='', log_file='', **kwargs)

Bases: DataMonitorMixin, _BasePort, Generic[T]

A class to simply port methods for all devices / shell / sockets to similar usage

• Create receive thread and pexpect spawn process for data read/expect

• Redefine

Parameters:

• raw_port (T)

• name (str)

• log_file (str)

• kwargs (Any)

EXPECT_TIMEOUT_EXCEPTIONS: Tuple[Type[Exception], ...] = (<class 'TimeoutError'>, <class 'pexpect.exceptions.ExceptionPexpect'>)

INIT_START_REDIRECT_THREAD: bool = True

property port: T

property raw_port: T

property name: str

property logger: Logger

property log_file: str

Get Current dut log file.

property rx_log_callback: Callable[[str, bytes], None] | None

Get Current dut log file.

set_rx_log_callback(new_callback)

Parameters:

new_callback (Callable[[str, bytes], None] | None)

Return type:

None

property monitors: List[DataMonitor]

property spawn: PortSpawn | None

Allow the use of pexpect spawn enhancements, if pexpect process is available

start_redirect_thread()

Start a new thread to read data from port and save to data cache.

Return type:

None

stop_redirect_thread()

Stop the redirect thread and pexpect process.

Return type:

bool

disable_redirect_thread()

Return type:

Generator[None, None, None]

write(data)

Parameters:

data (AnyStr)

Return type:

None

write_line(data, end='\n')

Parameters:

• data (AnyStr)

• end (str)

Return type:

None

expect_exact(pattern, timeout)

this is similar to expect(), but only uses plain string/bytes matching

Parameters:

• pattern (str | bytes)

• timeout (float)

Return type:

None

expect(pattern: str, timeout: float = PEXPECT_DEFAULT_TIMEOUT) → None

expect(pattern: bytes, timeout: float = PEXPECT_DEFAULT_TIMEOUT) → None

expect(pattern: Pattern[str], timeout: float = PEXPECT_DEFAULT_TIMEOUT) → Match[str]

expect(pattern: Pattern[bytes], timeout: float = PEXPECT_DEFAULT_TIMEOUT) → Match[bytes]

This seeks through the stream until a pattern is matched.

This expect() method is different with the one in pexpect. This method only accepts pattern type str/bytes or re.Pattern. Does not accept list, EOF or TIMEOUT. If the pattern type is str or bytes, this method is similar to expect_exact(), but returning None. If the pattern type is re.Pattern, this method will return a re.Match object if the pattern is matched. Can read all output data by pattern=re.compile(‘.+’, re.DOTALL)

Note

When matching very long data in a single read, pexpect may truncate the buffer due to its maxread limit. If the expected pattern can span a large chunk of output, increase maxread on the underlying pexpect spawn accordingly.

Parameters:

• pattern (t.Union[str, bytes, re.Pattern]) – pattern to match

• timeout (int, optional) – seconds of waiting for new data if match failed. Defaults to 30s.

Returns:

match result if the input pattern is re.Pattern

Return type:

t.Optional[re.Match]

property data_cache: str

flush_data()

Return type:

str

read_all_data(flush=True)

Parameters:

flush (bool)

Return type:

str

read_all_bytes(flush=False)

Read out all data from dut, return immediately.

Returns:

all data read from dut

Return type:

bytes

Parameters:

flush (bool)

close()

Return type:

None