pytest-embedded

class pytest_embedded.app.App(app_path: Optional[str] = None, build_dir: Optional[str] = None, **kwargs)

Bases: object

Built binary files base class

app_path

application folder path

Type:: str

binary_path

binary folder path

Type:: str

class pytest_embedded.dut.Dut(*args, **kwargs)

Bases: _InjectMixinCls

Device under test (DUT) base class

pexpect_proc

PexpectProcess instance

Type:: PexpectProcess

app

App instance

Type:: App

logfile

log file path

Type:: str

test_case_name

test case name

Type:: str

close() → None

expect(pattern, **kwargs) → Match

Expect the pattern from the internal buffer. All the arguments will be passed to pexpect.expect().

Parameters:

pattern – string, or compiled regex, or a list of string and compiled regex.

Keyword Arguments:

timeout (float) – would raise pexpect.TIMEOUT exception when pattern is not matched after timeout
expect_all (bool) – need to match all specified patterns if this flag is True. Otherwise match any of them could pass
not_matching – string, or compiled regex, or a list of string and compiled regex.

Returns:

AnyStr or re.Match

AnyStr: if you’re matching pexpect.EOF or pexpect.TIMEOUT to get all the current buffers.
re.Match: if matched given string.

expect_exact(pattern, **kwargs) → Match

Expect the pattern from the internal buffer. All the arguments will be passed to pexpect.expect_exact().

Parameters:

pattern – string, or a list of string

Keyword Arguments:

timeout (float) – would raise pexpect.TIMEOUT exception when pattern is not matched after timeout
expect_all (bool) – need to match all specified patterns if this flag is True. Otherwise match any of them could pass
not_matching – string, or compiled regex, or a list of string and compiled regex.

Returns:

AnyStr or re.Match

AnyStr: if you’re matching pexpect.EOF or pexpect.TIMEOUT to get all the current buffers.
re.Match: if matched given string.

expect_unity_test_output(remove_asci_escape_code: bool = True, timeout: float = 60, extra_before: Optional[AnyStr] = None) → None

Expect a unity test summary block and parse the output into junit report.

Would combine the junit report into the main one if you use pytest --junitxml feature.

Parameters:

remove_asci_escape_code – remove asci escape code in the message field. (default: True)
timeout – timeout. (default: 60 seconds)
extra_before – would append before the expected bytes. Use this argument when need to run expect functions between one unity test call.

Note

Would raise AssertionError at the end of the test if any unity test case result is “FAIL”
Would raise TIMEOUT exception at the end of the test if any unity test case execution took longer than timeout value

Warning

All unity test cases record would be missed if the final report block is uncaught.

property logdir

run_all_single_board_cases(group: Optional[str] = None, reset: bool = False, timeout: float = 30, run_ignore_cases: bool = False) → None

Run all multi_stage cases

Parameters:

group – test case group
reset – whether to perform a hardware reset before running a case
timeout – timeout. (Default: 30 seconds)
run_ignore_cases – run ignored test cases or not

Warning

requires enable service idf

write(s: AnyStr) → None: Write to the MessageQueue instance

class pytest_embedded.dut_factory.DutFactory

Bases: object

classmethod close()

classmethod create(*, embedded_services: str = '', app_path: str = '', build_dir: str = 'build', port: Optional[str] = None, port_location: Optional[str] = None, port_mac: Optional[str] = None, target: Optional[str] = None, beta_target: Optional[str] = None, baud: Optional[int] = None, flash_port: Optional[str] = None, skip_autoflash: Optional[bool] = None, erase_all: Optional[bool] = None, esptool_baud: Optional[int] = None, esp_flash_force: Optional[bool] = False, part_tool: Optional[str] = None, confirm_target_elf_sha256: Optional[bool] = None, erase_nvs: Optional[bool] = None, skip_check_coredump: Optional[bool] = None, panic_output_decode_script: Optional[str] = None, openocd_prog_path: Optional[str] = None, openocd_cli_args: Optional[str] = None, gdb_prog_path: Optional[str] = None, gdb_cli_args: Optional[str] = None, no_gdb: Optional[bool] = None, qemu_image_path: Optional[str] = None, qemu_prog_path: Optional[str] = None, qemu_cli_args: Optional[str] = None, qemu_extra_args: Optional[str] = None, wokwi_cli_path: Optional[str] = None, wokwi_timeout: Optional[int] = 0, wokwi_scenario: Optional[str] = None, wokwi_diagram: Optional[str] = None, skip_regenerate_image: Optional[bool] = None, encrypt: Optional[bool] = None, keyfile: Optional[str] = None)

Create a Device Under Test (DUT) object with customizable parameters.

Note

If you want to add an additional ‘arg’ parameter here, you also need to change plugin.py. Steps to add an argument: 1. (plugin.py) Create a fixture to read the argument value. 2. (plugin.py) Add the argument to parametrize_fixtures. 3. (dut_factory.py) Add it to _fixture_classes_and_options_fn. 4. (dut_factory.py) Add it to DutFactory.create.

Parameters:

embedded_services – Comma-separated list of embedded services.
app_path – Path to the application.
build_dir – Directory for build output (default is ‘build’).
port – Port configuration.
port_location – Port location.
port_mac – Port MAC address.
target – Target configuration.
beta_target – Beta target configuration.
baud – Baud rate.
flash_port – Port used for flashing the app.
skip_autoflash – Skip autoflash flag.
erase_all – Erase all flag.
esptool_baud – ESP tool baud rate.
esp_flash_force – ESP flash force flag.
part_tool – Part tool configuration.
confirm_target_elf_sha256 – Confirm target ELF SHA256.
erase_nvs – Erase NVS flag.
skip_check_coredump – Skip coredump check flag.
panic_output_decode_script – Panic output decode script.
openocd_prog_path – OpenOCD program path.
openocd_cli_args – OpenOCD CLI arguments.
gdb_prog_path – GDB program path.
gdb_cli_args – GDB CLI arguments.
no_gdb – No GDB flag.
qemu_image_path – QEMU image path.
qemu_prog_path – QEMU program path.
qemu_cli_args – QEMU CLI arguments.
qemu_extra_args – Additional QEMU arguments.
wokwi_cli_path – Wokwi CLI path.
wokwi_timeout – Wokwi timeout.
wokwi_scenario – Wokwi scenario path.
wokwi_diagram – Wokwi diagram path.
skip_regenerate_image – Skip image regeneration flag.
encrypt – Encryption flag.
keyfile – Keyfile for encryption.

Returns:

The created Device Under Test object.

Return type:

DUT object

Examples

>>> foo = DutFactory.create(embedded_services='idf,esp', app_path='path_to_hello_world')
>>> foo.expect_exact('Hello world!')

obj_stack: ClassVar[List[List[Any]]] = []

classmethod unity_tester(*args: IdfDut)

pytest_embedded.dut_factory.app_fn(_fixture_classes_and_options: ClassCliOptions) → App

pytest_embedded.dut_factory.dut_gn(_fixture_classes_and_options: ClassCliOptions, openocd: Optional[OpenOcd], gdb: Optional[Gdb], app: App, serial: Optional[Union[Serial, LinuxSerial]], qemu: Optional[Qemu], wokwi: Optional[WokwiCLI]) → Union[Dut, List[Dut]]

pytest_embedded.dut_factory.gdb_gn(_fixture_classes_and_options: ClassCliOptions) → Optional[Gdb]

pytest_embedded.dut_factory.msg_queue_gn() → MessageQueue

pytest_embedded.dut_factory.openocd_gn(_fixture_classes_and_options: ClassCliOptions) → Optional[OpenOcd]

pytest_embedded.dut_factory.pexpect_proc_fn(_pexpect_fr) → PexpectProcess

pytest_embedded.dut_factory.qemu_gn(_fixture_classes_and_options: ClassCliOptions, app) → Optional[Qemu]

pytest_embedded.dut_factory.serial_gn(_fixture_classes_and_options, msg_queue, app) → Optional[Union[Serial, LinuxSerial]]

pytest_embedded.dut_factory.set_parametrized_fixtures_cache(values: Dict)

pytest_embedded.dut_factory.wokwi_gn(_fixture_classes_and_options: ClassCliOptions, app) → Optional[WokwiCLI]: A wokwi subprocess that could read/redirect/write

class pytest_embedded.log.DuplicateStdoutPopen(msg_queue: MessageQueue, cmd: Union[str, List[str]], meta: Optional[Meta] = None, **kwargs)

Bases: Popen

Subclass of subprocess.Popen that redirect the output into the MessageQueue instance

REDIRECT_CLS: alias of _PopenRedirectProcess

SOURCE = 'POPEN'

close()

terminate(): Terminate the process with SIGTERM

write(s: AnyStr) → None

Write to stdin via stdin.write.

If the input is str, will encode to bytes and add a b’\n’ automatically in the end.
If the input is bytes, will pass this directly.

Parameters:: s – bytes or str

class pytest_embedded.log.MessageQueue(*args, **kwargs)

Bases: Queue

flush()

isatty()

put(obj, **kwargs)

write(s: AnyStr)

class pytest_embedded.log.PexpectProcess(fd, args=None, timeout=30, maxread=2000, searchwindowsize=None, logfile=None, encoding=None, codec_errors='strict', use_poll=False)

Bases: fdspawn

Use a temp file to gather multiple inputs into one output, and do pexpect.expect() from one place.

property buffer_debug_str

read_nonblocking(size: int = 1, timeout: int = -1) → bytes

Since we’re using real file stream, here we only raise an EOF error when the file stream has been closed. This could solve the os.read() blocked issue.

Parameters:

size – Read at most size bytes.
timeout – Wait timeout seconds for file descriptor to be ready to read. When -1 (default), use self.timeout. When 0, poll.

Returns:

String containing the bytes read

terminate(force=False): Close the temporary file stream and itself.

pytest_embedded.log.live_print_call(*args, msg_queue: Optional[MessageQueue] = None, expect_returncode: int = 0, **kwargs)

live print the subprocess.Popen process

Parameters:

msg_queue – MessageQueue instance, would redirect to message queue instead of sys.stdout if specified
expect_returncode – expect return code. (Default 0). Would raise exception when return code is different

Note

This function behaves the same as subprocess.call(), it would block your current process.

class pytest_embedded.plugin.PytestEmbedded(parallel_count: int = 1, parallel_index: int = 1, check_duplicates: bool = False, prettify_junit_report: bool = False, add_target_as_marker: bool = False)

Bases: object

pytest_collection_modifyitems(items: List[Function])

pytest_runtest_call(item: Function)

pytest_sessionfinish(session: Session, exitstatus: int) → None

pytest_embedded.plugin.app(_fixture_classes_and_options: ClassCliOptions) → App: A pytest fixture to gather information from the specified built binary folder

pytest_embedded.plugin.app_path(request: FixtureRequest, test_file_path: str) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.baud(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.beta_target(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.build_dir(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.cache_dir(request: FixtureRequest) → str: Cache dir for pytest-embedded

pytest_embedded.plugin.close_factory_duts()

pytest_embedded.plugin.confirm_target_elf_sha256(request: FixtureRequest) → Optional[bool]: Enable parametrization for the same cli option

pytest_embedded.plugin.count(request): Enable parametrization for the same cli option. Inject to global variable COUNT.

pytest_embedded.plugin.dut(_fixture_classes_and_options: ClassCliOptions, openocd: Optional[OpenOcd], gdb: Optional[Gdb], app: App, serial: Optional[Union[Serial, LinuxSerial]], qemu: Optional[Qemu], wokwi: Optional[WokwiCLI]) → Union[Dut, List[Dut]]: A device under test (DUT) object that could gather output from various sources and redirect them to the pexpect process, and run expect() via its pexpect process.

pytest_embedded.plugin.dut_index(**kwargs)

pytest_embedded.plugin.dut_total()

pytest_embedded.plugin.embedded_services(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.encrypt(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.erase_all(request: FixtureRequest) → Optional[bool]: Enable parametrization for the same cli option

pytest_embedded.plugin.erase_nvs(request: FixtureRequest) → Optional[bool]: Enable parametrization for the same cli option

pytest_embedded.plugin.esp_flash_force(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.esptool_baud(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.flash_port(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.gdb(_fixture_classes_and_options: ClassCliOptions) → Optional[Gdb]: A gdb subprocess that could read/redirect/write

pytest_embedded.plugin.gdb_cli_args(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.gdb_prog_path(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.keyfile(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.logfile_extension(request: FixtureRequest) → str: Enable parametrization for the same cli option

pytest_embedded.plugin.msg_queue() → MessageQueue

pytest_embedded.plugin.multi_dut_argument(func) → Callable[[...], Union[str, None, Tuple[Optional[str]]]]: Used for parse the multi-dut argument according to the count amount.

pytest_embedded.plugin.multi_dut_fixture(func) → Callable[[...], Union[Any, Tuple[Any]]]

Apply the multi-dut arguments to each fixture.

Note

Run the func(*args, **kwargs) for multiple times by iterating all kwargs via itemgetter

For example:

input: {key1: (v1, v2), key2: (v1, v2)}
output: (func(**{key1: v1, key2: v1}), func(**{key1: v2, key2: v2}))

Returns:: The return value, if count is 1. The tuple of return values, if count is greater than 1.

pytest_embedded.plugin.multi_dut_generator_fixture(func) → Callable[[...], Generator[Union[Any, Tuple[Any]], Any, None]]

Apply the multi-dut arguments to each fixture.

Note

Run the func() for multiple times by iterating all kwargs via itemgetter. Auto call close() or terminate() method of the object after it yield back.

Yields:: The return value, if count is 1. The tuple of return values, if count is greater than 1.

pytest_embedded.plugin.no_gdb(request: FixtureRequest) → bool

pytest_embedded.plugin.openocd(_fixture_classes_and_options: ClassCliOptions) → Optional[OpenOcd]: An openocd subprocess that could read/redirect/write

pytest_embedded.plugin.openocd_cli_args(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.openocd_prog_path(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.panic_output_decode_script(request: FixtureRequest) → Optional[bool]: Enable parametrization for the same cli option

pytest_embedded.plugin.parametrize_fixtures(_services, app_path, build_dir, port, port_location, port_mac, target, beta_target, baud, flash_port, skip_autoflash, erase_all, esptool_baud, esp_flash_force, part_tool, confirm_target_elf_sha256, erase_nvs, skip_check_coredump, panic_output_decode_script, openocd_prog_path, openocd_cli_args, gdb_prog_path, gdb_cli_args, no_gdb, qemu_image_path, qemu_prog_path, qemu_cli_args, qemu_extra_args, wokwi_cli_path, wokwi_timeout, wokwi_scenario, wokwi_diagram, skip_regenerate_image, encrypt, keyfile, test_case_name, _meta)

pytest_embedded.plugin.parse_multi_dut_args(count: int, s: str) → Union[Any, Tuple[Any]]

Parse multi-dut argument by the following rules:

When the return value is a string, split the string by |.
If the configuration value only has one item, duplicate it by the “count” amount.
If the configuration value item amount is the same as the “count” amount, return it directly.

Parameters:

count – Multi-Dut count
s – argument string

Returns:

The argument itself. if count is 1. The tuple of the parsed argument. if count is greater than 1.

Raises:

ValueError – when a configuration has multi values but the amount is different from the count amount.

pytest_embedded.plugin.part_tool(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.pexpect_proc(_pexpect_fr) → PexpectProcess: Pexpect process that run the expect functions on

pytest_embedded.plugin.port(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.port_app_cache() → Dict[str, str]: Session scoped port-app cache, for idf only

pytest_embedded.plugin.port_location(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.port_mac(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.port_target_cache(cache_dir) → Dict[str, str]: Session scoped port-target cache, for esp only

pytest_embedded.plugin.pytest_addoption(parser)

pytest_embedded.plugin.pytest_configure(config: Config) → None

pytest_embedded.plugin.pytest_unconfigure(config: Config) → None

pytest_embedded.plugin.qemu(_fixture_classes_and_options: ClassCliOptions, app) → Optional[Qemu]: A qemu subprocess that could read/redirect/write

pytest_embedded.plugin.qemu_cli_args(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.qemu_extra_args(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.qemu_image_path(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.qemu_prog_path(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.redirect(msg_queue: MessageQueue) → Callable[[...], redirect_stdout]

A context manager that could help duplicate all the sys.stdout to msg_queue.

Examples

>>> with redirect():
>>>    print('this should be logged and sent to pexpect_proc')

pytest_embedded.plugin.serial(_fixture_classes_and_options, msg_queue, app) → Optional[Union[Serial, LinuxSerial]]: A serial subprocess that could read/redirect/write

pytest_embedded.plugin.session_root_logdir(request: FixtureRequest) → str: Session scoped log dir for pytest-embedded

pytest_embedded.plugin.session_tempdir(request: FixtureRequest, session_root_logdir: str) → str: Session scoped temp dir for pytest-embedded

pytest_embedded.plugin.skip_autoflash(request: FixtureRequest) → Optional[bool]: Enable parametrization for the same cli option

pytest_embedded.plugin.skip_check_coredump(request: FixtureRequest) → Optional[bool]: Enable parametrization for the same cli option

pytest_embedded.plugin.skip_regenerate_image(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.target(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.test_case_name(request: FixtureRequest) → str: Current test case function name

pytest_embedded.plugin.test_case_tempdir(test_case_name: str, session_tempdir: str) → str: Function scoped temp dir for pytest-embedded

pytest_embedded.plugin.test_file_path(request: FixtureRequest) → str: Current test script file path

pytest_embedded.plugin.unity_tester(dut: Union[IdfDut, Tuple[IdfDut]]) → Optional[CaseTester]

pytest_embedded.plugin.with_timestamp(request: FixtureRequest) → bool: Enable parametrization for the same cli option

pytest_embedded.plugin.wokwi(_fixture_classes_and_options: ClassCliOptions, app) → Optional[WokwiCLI]: A wokwi subprocess that could read/redirect/write

pytest_embedded.plugin.wokwi_cli_path(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.wokwi_diagram(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.wokwi_scenario(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

pytest_embedded.plugin.wokwi_timeout(request: FixtureRequest) → Optional[str]: Enable parametrization for the same cli option

class pytest_embedded.unity.JunitMerger(main_junit: Optional[str], unity_test_report_mode: Optional[str] = None)

Bases: object

SUB_JUNIT_FILENAME = 'dut.xml'

property junit: ElementTree

merge(junit_files: List[str])

class pytest_embedded.unity.TestCase(name: str, result: str, **kwargs)

Bases: object

to_xml() → Element

class pytest_embedded.unity.TestFormat(value)

Bases: Enum

An enumeration.

BASIC = 0

FIXTURE = 1

class pytest_embedded.unity.TestSuite(name: Optional[str] = None, **kwargs)

Bases: object

add_unity_test_cases(s: AnyStr, additional_attrs: Optional[Dict[str, Any]] = None) → None

dump(path: str) → None

property failed_cases: List[TestCase]

to_xml() → Element

class pytest_embedded.unity.UnityTestReportMode(value)

Bases: str, Enum

An enumeration.

MERGE = 'merge'

REPLACE = 'replace'

pytest_embedded.unity.escape_dict_value(d: Dict[str, Any]) → Dict[str, str]

pytest_embedded.unity.escape_illegal_xml_chars(s: str) → str

class pytest_embedded.utils.ClassCliOptions(classes: Dict[str, type], mixins: Dict[str, List[type]], kwargs: Dict[str, Dict[str, Any]])

Bases: object

classes: Dict[str, type]

kwargs: Dict[str, Dict[str, Any]]

mixins: Dict[str, List[type]]

class pytest_embedded.utils.Meta(logdir: str, port_target_cache: Dict[str, str], port_app_cache: Dict[str, str], logfile_extension: str = '.log')

Bases: object

Meta info for testing, session scope

drop_port_app_cache(port: str) → None

drop_port_target_cache(port: str) → None

hit_port_app_cache(port: str, app: App) → bool

hit_port_target_cache(port: str, target: str) → bool

logdir: str

logfile_extension: str = '.log'

port_app_cache: Dict[str, str]

port_target_cache: Dict[str, str]

set_port_app_cache(port: str, app: App) → None

set_port_target_cache(port: str, target: str) → None

exception pytest_embedded.utils.PackageNotInstalledError(service: str): Bases: SystemExit

exception pytest_embedded.utils.RequireServiceError(func_name: str, services: Union[str, List[str]]): Bases: SystemExit

exception pytest_embedded.utils.UnknownServiceError(service: str): Bases: SystemExit

exception pytest_embedded.utils.UserHint: Bases: Warning

pytest_embedded.utils.find_by_suffix(suffix: str, path: str) → List[str]

pytest_embedded.utils.lazy_load(base_module: module, name_obj_dict: Dict[str, Any], obj_module_dict: Dict[str, str]) → Callable[[str], Any]

use __getattr__ in the __init__.py file to lazy load some objects

Parameters:

base_module (ModuleType) – base package module
name_obj_dict (dict[str, any]) – name, real object dict, used to store real objects, no need to add lazy-load objects
obj_module_dict (dict[str, str]) – dict of object name and module name

Returns:

__getattr__ function

Example

__getattr__ = lazy_load(
    importlib.import_module(__name__),
    {
        'IdfApp': IdfApp,
        'LinuxDut': LinuxDut,
        'LinuxSerial': LinuxSerial,
        'CaseTester': CaseTester,
    },
    {
        'IdfSerial': '.serial',
        'IdfDut': '.dut',
    },
)

pytest_embedded.utils.remove_asci_color_code(s: AnyStr) → str

pytest_embedded.utils.to_bytes(bytes_str: AnyStr, ending: Optional[AnyStr] = None) → bytes

Turn bytes or str to bytes

Parameters:

bytes_str – bytes or str
ending – bytes or str, will add to the end of the result. Only works when the bytes_str is str

Returns:

utf8-encoded bytes

pytest_embedded.utils.to_list(s: _T) → List[_T]

Parameters:

s – Anything

Returns:

List (list[_T])

list(s) (List. If s is a tuple or a set.
itself. If s is a list.
[s]. If s is other types.

pytest_embedded.utils.to_str(bytes_str: AnyStr) → str

Turn bytes or str to str

Parameters:: bytes_str – bytes or str
Returns:: utf8-decoded string

pytest_embedded.utils.utcnow_str() → str