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_queueMessageQueue 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])

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
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_strbytes or str

  • endingbytes 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_strbytes or str

Returns:

utf8-decoded string

pytest_embedded.utils.utcnow_str() str