From 58626c670ae45b9df3236ae9f51b3535fbc2b4ac Mon Sep 17 00:00:00 2001 From: Igor Udot Date: Mon, 13 Nov 2023 18:32:00 +0800 Subject: [PATCH] ref: create dut-factory --- pytest-embedded-idf/tests/test_idf.py | 49 ++ pytest-embedded-serial/tests/test_serial.py | 58 ++ pytest-embedded/pytest_embedded/__init__.py | 3 +- .../pytest_embedded/dut_factory.py | 682 ++++++++++++++++++ pytest-embedded/pytest_embedded/plugin.py | 431 ++--------- pytest-embedded/pytest_embedded/utils.py | 9 + 6 files changed, 853 insertions(+), 379 deletions(-) create mode 100644 pytest-embedded/pytest_embedded/dut_factory.py diff --git a/pytest-embedded-idf/tests/test_idf.py b/pytest-embedded-idf/tests/test_idf.py index 3bec38ca..8e910006 100644 --- a/pytest-embedded-idf/tests/test_idf.py +++ b/pytest-embedded-idf/tests/test_idf.py @@ -130,6 +130,55 @@ def test_no_matching_word_pass_rest(dut): result.assert_outcomes(passed=2, failed=2) +def test_custom_idf_device_dut(testdir): + p = os.path.join(testdir.tmpdir, 'hello_world_esp32') + p_c3 = os.path.join(testdir.tmpdir, 'hello_world_esp32c3') + unity_test_path = os.path.join(testdir.tmpdir, 'unit_test_app_esp32') + unity_test_path_c3 = os.path.join(testdir.tmpdir, 'unit_test_app_esp32c3') + testdir.makepyfile(f""" + import pytest + + def test_idf_custom_dev(): + from pytest_embedded.dut_factory import DutFactory + dut = DutFactory.create(embedded_services='esp,idf', app_path=r'{p}') + dut.expect("Hello") + + def test_idf_mixed(dut): + from pytest_embedded.dut_factory import DutFactory + dutc = DutFactory.create(embedded_services='esp,idf', app_path=r'{p_c3}') + dutc.expect("Hello") + dut.expect("Hello") + assert dutc.serial.port!=dut.serial.port + + def test_idf_unity_tester(): + from pytest_embedded.dut_factory import DutFactory + dut1 = DutFactory.create(embedded_services='esp,idf', app_path=r'{unity_test_path}') + dut2 = DutFactory.create(embedded_services='esp,idf', app_path=r'{unity_test_path_c3}') + tester = DutFactory.unity_tester(dut1, dut2) + tester.run_all_cases() + + def test_idf_run_all_single_board_cases(): + from pytest_embedded.dut_factory import DutFactory + dut1 = DutFactory.create(embedded_services='esp,idf', app_path=r'{unity_test_path}') + dut1.run_all_single_board_cases() + """) + + result = testdir.runpytest( + '-s', + '--app-path', p, + '--embedded-services', 'esp,idf', + '--junitxml', 'report.xml', + ) + result.assert_outcomes(passed=4, errors=0) + + junit_report = ET.parse('report.xml').getroot()[0] + + assert junit_report.attrib['errors'] == '0' + assert junit_report.attrib['failures'] == '2' + assert junit_report.attrib['skipped'] == '0' + assert junit_report.attrib['tests'] == '7' + + def test_idf_serial_flash_with_erase_nvs(testdir): testdir.makepyfile(""" import pexpect diff --git a/pytest-embedded-serial/tests/test_serial.py b/pytest-embedded-serial/tests/test_serial.py index b262f84d..49b12ecb 100644 --- a/pytest-embedded-serial/tests/test_serial.py +++ b/pytest-embedded-serial/tests/test_serial.py @@ -3,6 +3,64 @@ import pytest +def test_custom_serial_device(testdir): + testdir.makepyfile(r""" + import pytest + + def test_serial_mixed(dut): + from pytest_embedded.dut_factory import DutFactory + assert len(dut)==2 + another_dut = DutFactory.create() + st = set( + ( + dut[0].serial.port, + dut[1].serial.port, + another_dut.serial.port + ) + ) + assert len(st) == 3 + + def test_custom_dut(): + from pytest_embedded.dut_factory import DutFactory + another_dut = DutFactory.create(embedded_services='esp,serial') + """) + + result = testdir.runpytest( + '-s', + '--embedded-services', 'esp,serial', + '--count', 2, + ) + result.assert_outcomes(passed=2, errors=0) + + +def test_custom_serial_device_dut_count_1(testdir): + testdir.makepyfile(r""" + import pytest + + def test_serial_device_created_dut_count_1(dut): + from pytest_embedded.dut_factory import DutFactory + another_dut = DutFactory.create() + another_dut2 = DutFactory.create() + st = set( + ( + dut.serial.port, + another_dut.serial.port, + another_dut2.serial.port + ) + ) + assert len(st) == 3 + + + """) + + result = testdir.runpytest( + '-s', + '--embedded-services', 'esp,serial', + '--count', 1, + ) + result.assert_outcomes(passed=1, errors=0) + + @pytest.mark.skipif(sys.platform == 'win32', reason='No socat support on windows') @pytest.mark.flaky(reruns=3, reruns_delay=2) def test_serial_port(testdir): diff --git a/pytest-embedded/pytest_embedded/__init__.py b/pytest-embedded/pytest_embedded/__init__.py index bf5ac5b3..c00116a9 100644 --- a/pytest-embedded/pytest_embedded/__init__.py +++ b/pytest-embedded/pytest_embedded/__init__.py @@ -2,7 +2,8 @@ from .app import App from .dut import Dut +from .dut_factory import DutFactory -__all__ = ['App', 'Dut'] +__all__ = ['App', 'Dut', 'DutFactory'] __version__ = '1.8.3' diff --git a/pytest-embedded/pytest_embedded/dut_factory.py b/pytest-embedded/pytest_embedded/dut_factory.py new file mode 100644 index 00000000..610e7ad5 --- /dev/null +++ b/pytest-embedded/pytest_embedded/dut_factory.py @@ -0,0 +1,682 @@ +import datetime +import gc +import io +import logging +import multiprocessing +import os +import subprocess +import sys +import typing as t +from collections import defaultdict +from pathlib import Path + +from pytest_embedded_idf import CaseTester +from pytest_embedded_idf.dut import IdfDut + +if t.TYPE_CHECKING: + from pytest_embedded_idf import LinuxSerial + from pytest_embedded_jtag import Gdb, OpenOcd + from pytest_embedded_qemu import Qemu + from pytest_embedded_serial import Serial + from pytest_embedded_wokwi import WokwiCLI + +from pytest_embedded import App, Dut + +from .log import MessageQueue, PexpectProcess +from .utils import FIXTURES_SERVICES, ClassCliOptions, to_str + + +def _drop_none_kwargs(kwargs: t.Dict[t.Any, t.Any]): + return {k: v for k, v in kwargs.items() if v is not None} + + +_ctx = multiprocessing.get_context() +_stdout = sys.__stdout__ +dut_global_index = 0 + + +def msg_queue_gn() -> MessageQueue: + return MessageQueue(ctx=_ctx) + + +def _listen(q: MessageQueue, filepath: str, with_timestamp: bool = True, count: int = 1, total: int = 1) -> None: + _added_prefix = False + while True: + msg = q.get() + if not msg: + continue + + with open(filepath, 'ab') as fw: + fw.write(msg) + fw.flush() + + _s = to_str(msg) + if not _s: + continue + + prefix = '' + if total > 1: + source = f'dut-{count}' + else: + source = None + + if source: + prefix = f'[{source}] ' + prefix + + if with_timestamp: + prefix = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') + ' ' + prefix + + if not _added_prefix: + _s = prefix + _s + _added_prefix = True + _s = _s.replace('\r\n', '\n') # remove extra \r. since multi-dut \r would mess up the log + _s = _s.replace('\n', '\n' + prefix) + if prefix and _s.endswith(prefix): + _s = _s.rsplit(prefix, maxsplit=1)[0] + _added_prefix = False + + _stdout.write(_s) + _stdout.flush() + + +def _listener_gn(msg_queue, _pexpect_logfile, with_timestamp, dut_index, dut_total) -> multiprocessing.Process: + os.makedirs(os.path.dirname(_pexpect_logfile), exist_ok=True) + kwargs = { + 'with_timestamp': with_timestamp, + 'count': dut_index, + 'total': dut_total, + } + + return _ctx.Process( + target=_listen, + args=( + msg_queue, + _pexpect_logfile, + ), + kwargs=_drop_none_kwargs(kwargs), + ) + + +def _pexpect_fr_gn(_pexpect_logfile, _listener) -> t.BinaryIO: + Path(_pexpect_logfile).touch() + _listener.start() + return open(_pexpect_logfile, 'rb') + + +def pexpect_proc_fn(_pexpect_fr) -> PexpectProcess: + return PexpectProcess(_pexpect_fr) + + +def _fixture_classes_and_options_fn( + _services, + # parametrize fixtures + app_path, + build_dir, + port, + port_location, + port_mac, + target, + beta_target, + baud, + 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, + skip_regenerate_image, + encrypt, + keyfile, + # pre-initialized fixtures + dut_index, + _pexpect_logfile, + test_case_name, + pexpect_proc, + msg_queue, + _meta, +) -> ClassCliOptions: + classes: t.Dict[str, type] = {} + mixins: t.Dict[str, t.List[type]] = defaultdict(list) + kwargs: t.Dict[str, t.Dict[str, t.Any]] = defaultdict(dict) + + for fixture in FIXTURES_SERVICES.keys(): + if fixture == 'app': + kwargs['app'] = {'app_path': app_path, 'build_dir': build_dir} + if 'idf' in _services: + if 'qemu' in _services: + from pytest_embedded_qemu import DEFAULT_IMAGE_FN, QemuApp + + classes[fixture] = QemuApp + kwargs[fixture].update({ + 'msg_queue': msg_queue, + 'part_tool': part_tool, + 'qemu_image_path': qemu_image_path, + 'skip_regenerate_image': skip_regenerate_image, + 'encrypt': encrypt, + 'keyfile': keyfile, + 'qemu_prog_path': qemu_prog_path, + }) + else: + from pytest_embedded_idf import IdfApp + + classes[fixture] = IdfApp + kwargs[fixture].update({ + 'part_tool': part_tool, + }) + elif 'arduino' in _services: + from pytest_embedded_arduino import ArduinoApp + + classes[fixture] = ArduinoApp + else: + from .app import App + + classes[fixture] = App + elif fixture == 'serial': + if 'esp' in _services: + from pytest_embedded_serial_esp import EspSerial + + kwargs[fixture] = { + 'pexpect_proc': pexpect_proc, + 'msg_queue': msg_queue, + 'target': target, + 'beta_target': beta_target, + 'port': os.getenv('ESPPORT') or port, + 'port_location': port_location, + 'port_mac': port_mac, + 'baud': int(baud or EspSerial.DEFAULT_BAUDRATE), + 'esptool_baud': int(os.getenv('ESPBAUD') or esptool_baud or EspSerial.ESPTOOL_DEFAULT_BAUDRATE), + 'esp_flash_force': esp_flash_force, + 'skip_autoflash': skip_autoflash, + 'erase_all': erase_all, + 'meta': _meta, + } + if 'idf' in _services: + from pytest_embedded_idf import IdfSerial + + classes[fixture] = IdfSerial + kwargs[fixture].update({ + 'app': None, + 'confirm_target_elf_sha256': confirm_target_elf_sha256, + 'erase_nvs': erase_nvs, + }) + elif 'arduino' in _services: + from pytest_embedded_arduino import ArduinoSerial + + classes[fixture] = ArduinoSerial + kwargs[fixture].update({ + 'app': None, + }) + else: + from pytest_embedded_serial_esp import EspSerial + + classes[fixture] = EspSerial + elif 'serial' in _services or 'jtag' in _services: + from pytest_embedded_serial.serial import Serial + + classes[fixture] = Serial + kwargs[fixture] = { + 'msg_queue': msg_queue, + 'port': port, + 'port_location': port_location, + 'baud': int(baud or Serial.DEFAULT_BAUDRATE), + 'meta': _meta, + } + elif fixture in ['openocd', 'gdb']: + if 'jtag' in _services: + if fixture == 'openocd': + from pytest_embedded_jtag import OpenOcd + + classes[fixture] = OpenOcd + kwargs[fixture] = { + 'msg_queue': msg_queue, + 'app': None, + 'openocd_prog_path': openocd_prog_path, + 'openocd_cli_args': openocd_cli_args, + 'port_offset': dut_index, + 'meta': _meta, + } + elif not no_gdb: + from pytest_embedded_jtag import Gdb + + classes[fixture] = Gdb + kwargs[fixture] = { + 'msg_queue': msg_queue, + 'gdb_prog_path': gdb_prog_path, + 'gdb_cli_args': gdb_cli_args, + 'meta': _meta, + } + elif fixture == 'qemu': + if 'qemu' in _services: + from pytest_embedded_qemu import ( + DEFAULT_IMAGE_FN, + ENCRYPTED_IMAGE_FN, + Qemu, + ) + + classes[fixture] = Qemu + kwargs[fixture] = { + 'msg_queue': msg_queue, + 'qemu_image_path': qemu_image_path + or os.path.join( + app_path or '', build_dir or 'build', ENCRYPTED_IMAGE_FN if encrypt else DEFAULT_IMAGE_FN + ), + 'qemu_prog_path': qemu_prog_path, + 'qemu_cli_args': qemu_cli_args, + 'qemu_extra_args': qemu_extra_args, + 'app': None, + 'meta': _meta, + 'dut_index': dut_index, + } + elif fixture == 'wokwi': + if 'wokwi' in _services: + from pytest_embedded_wokwi import WokwiCLI + + classes[fixture] = WokwiCLI + kwargs[fixture].update({ + 'wokwi_cli_path': wokwi_cli_path, + 'wokwi_timeout': wokwi_timeout, + 'msg_queue': msg_queue, + 'app': None, + 'meta': _meta, + }) + elif fixture == 'dut': + classes[fixture] = Dut + kwargs[fixture] = { + 'pexpect_proc': pexpect_proc, + 'msg_queue': msg_queue, + 'app': None, + 'pexpect_logfile': _pexpect_logfile, + 'test_case_name': test_case_name, + 'meta': _meta, + } + if 'idf' in _services and 'esp' not in _services: + # esp,idf will use IdfDut, which based on IdfUnityDutMixin already + from pytest_embedded_idf.unity_tester import IdfUnityDutMixin + + mixins[fixture].append(IdfUnityDutMixin) + + if 'wokwi' in _services: + from pytest_embedded_wokwi import WokwiDut + + classes[fixture] = WokwiDut + kwargs[fixture].update({ + 'wokwi': None, + }) + + if 'idf' in _services: + from pytest_embedded_wokwi.idf import IDFFirmwareResolver + + kwargs['wokwi'].update({'firmware_resolver': IDFFirmwareResolver()}) + else: + raise SystemExit('wokwi service should be used together with idf service') + elif 'qemu' in _services: + from pytest_embedded_qemu import QemuDut + + classes[fixture] = QemuDut + kwargs[fixture].update({ + 'qemu': None, + }) + elif 'jtag' in _services: + if 'idf' in _services: + from pytest_embedded_idf import IdfDut + + classes[fixture] = IdfDut + else: + from pytest_embedded_serial import SerialDut + + classes[fixture] = SerialDut + + kwargs[fixture].update({ + 'serial': None, + 'openocd': None, + 'gdb': None, + }) + elif 'serial' in _services or 'esp' in _services: + if 'esp' in _services and 'idf' in _services: + from pytest_embedded_idf import IdfDut + + classes[fixture] = IdfDut + kwargs[fixture].update({ + 'skip_check_coredump': skip_check_coredump, + 'panic_output_decode_script': panic_output_decode_script, + }) + else: + from pytest_embedded_serial import SerialDut + + classes[fixture] = SerialDut + + kwargs[fixture].update({ + 'serial': None, + }) + + return ClassCliOptions(classes, mixins, kwargs) + + +def app_fn(_fixture_classes_and_options: ClassCliOptions) -> App: + cls = _fixture_classes_and_options.classes['app'] + kwargs = _fixture_classes_and_options.kwargs['app'] + return cls(**_drop_none_kwargs(kwargs)) + + +def serial_gn(_fixture_classes_and_options, msg_queue, app) -> t.Optional[t.Union['Serial', 'LinuxSerial']]: + if hasattr(app, 'target') and app.target == 'linux': + from pytest_embedded_idf import LinuxSerial + + cls = LinuxSerial + kwargs = { + 'app': app, + 'msg_queue': msg_queue, + } + return cls(**kwargs) + + if 'serial' not in _fixture_classes_and_options.classes: + return None + + cls = _fixture_classes_and_options.classes['serial'] + kwargs = _fixture_classes_and_options.kwargs['serial'] + if 'app' in kwargs and kwargs['app'] is None: + kwargs['app'] = app + return cls(**_drop_none_kwargs(kwargs)) + + +def openocd_gn(_fixture_classes_and_options: ClassCliOptions) -> t.Optional['OpenOcd']: + if 'openocd' not in _fixture_classes_and_options.classes: + return None + + cls = _fixture_classes_and_options.classes['openocd'] + kwargs = _fixture_classes_and_options.kwargs['openocd'] + return cls(**_drop_none_kwargs(kwargs)) + + +def gdb_gn(_fixture_classes_and_options: ClassCliOptions) -> t.Optional['Gdb']: + if 'gdb' not in _fixture_classes_and_options.classes: + return None + + cls = _fixture_classes_and_options.classes['gdb'] + kwargs = _fixture_classes_and_options.kwargs['gdb'] + return cls(**_drop_none_kwargs(kwargs)) + + +def qemu_gn(_fixture_classes_and_options: ClassCliOptions, app) -> t.Optional['Qemu']: + if 'qemu' not in _fixture_classes_and_options.classes: + return None + + cls = _fixture_classes_and_options.classes['qemu'] + kwargs = _fixture_classes_and_options.kwargs['qemu'] + + if 'app' in kwargs and kwargs['app'] is None: + kwargs['app'] = app + + return cls(**_drop_none_kwargs(kwargs)) + + +def wokwi_gn(_fixture_classes_and_options: ClassCliOptions, app) -> t.Optional['WokwiCLI']: + """A wokwi subprocess that could read/redirect/write""" + if 'wokwi' not in _fixture_classes_and_options.classes: + return None + + cls = _fixture_classes_and_options.classes['wokwi'] + kwargs = _fixture_classes_and_options.kwargs['wokwi'] + + if 'app' in kwargs and kwargs['app'] is None: + kwargs['app'] = app + return cls(**_drop_none_kwargs(kwargs)) + + +def dut_gn( + _fixture_classes_and_options: ClassCliOptions, + openocd: t.Optional['OpenOcd'], + gdb: t.Optional['Gdb'], + app: App, + serial: t.Optional[t.Union['Serial', 'LinuxSerial']], + qemu: t.Optional['Qemu'], + wokwi: t.Optional['WokwiCLI'], +) -> t.Union[Dut, t.List[Dut]]: + global dut_global_index + dut_global_index += 1 + + kwargs = _fixture_classes_and_options.kwargs['dut'] + mixins = _fixture_classes_and_options.mixins['dut'] + + # since there's no way to know the target before setup finished + # we have to use the `app.target` to determine the dut class here + if hasattr(app, 'target') and app.target == 'linux': + from pytest_embedded_idf import LinuxDut + + cls = LinuxDut + kwargs['serial'] = None # replace it later with LinuxSerial + else: + cls = _fixture_classes_and_options.classes['dut'] + + for k, v in kwargs.items(): + if v is None: + if k == 'app': + kwargs[k] = app + elif k == 'serial': + kwargs[k] = serial + elif k == 'openocd': + kwargs[k] = openocd + elif k == 'gdb': + kwargs[k] = gdb + elif k == 'qemu': + kwargs[k] = qemu + elif k == 'wokwi': + kwargs[k] = wokwi + return cls(**_drop_none_kwargs(kwargs), mixins=mixins) + + +buff_parametrized_fixtures = {} + + +def set_buff_parametrized_fixtures(values: t.Dict): + global buff_parametrized_fixtures + buff_parametrized_fixtures = values.copy() + + +def _close_or_terminate(obj): + if obj is None: + del obj + return + + try: + if isinstance(obj, (subprocess.Popen, multiprocessing.process.BaseProcess)): + obj.terminate() + obj.kill() + elif isinstance(obj, io.IOBase): + try: + obj.close() + except Exception as e: + logging.debug('file %s closed failed with error: %s', obj, str(e)) + else: + try: + obj.close() + except AttributeError: + try: + obj.terminate() + except AttributeError: + pass + except Exception as e: + logging.debug('Not properly caught object %s: %s', obj, str(e)) + except Exception as e: + logging.debug('%s: %s', obj, str(e)) + return # swallow up all error + finally: + referrers = gc.get_referrers(obj) + for _referrer in referrers: + if isinstance(_referrer, list): + for _i, val in enumerate(_referrer): + if val is obj: + _referrer[_i] = None + elif isinstance(_referrer, dict): + for key, value in _referrer.items(): + if value is obj: + _referrer[key] = None + del obj + + +class DutFactory: + obj_stack: t.ClassVar[t.List] = [] + + @classmethod + def close(cls): + global dut_global_index + dut_global_index = 0 + if hasattr(cls, 'obj_stack'): + while cls.obj_stack: + layout = DutFactory.obj_stack.pop() + while layout: + obj = layout.pop() + _close_or_terminate(obj) + del layout + del DutFactory.obj_stack + cls.obj_stack = [] + + @classmethod + def unity_tester(cls, *args: 'IdfDut'): + return CaseTester(args) + + @classmethod + def create( + cls, + *, + embedded_services='', + app_path='', + build_dir='build', + port=None, + port_location=None, + port_mac=None, + target=None, + beta_target=None, + baud=None, + skip_autoflash=None, + erase_all=None, + esptool_baud=None, + esp_flash_force=False, + part_tool=None, + confirm_target_elf_sha256=None, + erase_nvs=None, + skip_check_coredump=None, + panic_output_decode_script=None, + openocd_prog_path=None, + openocd_cli_args=None, + gdb_prog_path=None, + gdb_cli_args=None, + no_gdb=None, + qemu_image_path=None, + qemu_prog_path=None, + qemu_cli_args=None, + qemu_extra_args=None, + wokwi_cli_path=None, + wokwi_timeout=0, + skip_regenerate_image=None, + encrypt=None, + keyfile=None, + ): + layout = [] + try: + global buff_parametrized_fixtures + msg_queue = msg_queue_gn() + layout.append(msg_queue) + + _pexpect_logfile = os.path.join( + buff_parametrized_fixtures['_meta'].logdir, f'custom-dut-{dut_global_index}.txt' + ) + logging.warning('You can get your custom DUT log file at the following path: %s.', _pexpect_logfile) + + _listener = _listener_gn(msg_queue, _pexpect_logfile, True, dut_global_index, dut_global_index + 1) + layout.append(_listener) + + _pexpect_fr = _pexpect_fr_gn(_pexpect_logfile, _listener) + layout.append(_pexpect_fr) + + pexpect_proc = pexpect_proc_fn(_pexpect_fr) + + _kwargs = { + '_services': embedded_services or buff_parametrized_fixtures['_services'], + # parametrize fixtures + 'app_path': app_path, + 'build_dir': build_dir, + 'port': port, + 'port_location': port_location, + 'port_mac': port_mac, + 'target': target, + 'beta_target': beta_target, + 'baud': baud, + 'skip_autoflash': skip_autoflash, + 'erase_all': erase_all, + 'esptool_baud': esptool_baud, + 'esp_flash_force': esp_flash_force, + 'part_tool': part_tool, + 'confirm_target_elf_sha256': confirm_target_elf_sha256, + 'erase_nvs': erase_nvs, + 'skip_check_coredump': skip_check_coredump, + 'panic_output_decode_script': panic_output_decode_script, + 'openocd_prog_path': openocd_prog_path, + 'openocd_cli_args': openocd_cli_args, + 'gdb_prog_path': gdb_prog_path, + 'gdb_cli_args': gdb_cli_args, + 'no_gdb': no_gdb, + 'qemu_image_path': qemu_image_path, + 'qemu_prog_path': qemu_prog_path, + 'qemu_cli_args': qemu_cli_args, + 'qemu_extra_args': qemu_extra_args, + 'wokwi_cli_path': wokwi_cli_path, + 'wokwi_timeout': wokwi_timeout, + 'skip_regenerate_image': skip_regenerate_image, + 'encrypt': encrypt, + 'keyfile': keyfile, + # common + 'test_case_name': buff_parametrized_fixtures['test_case_name'], + '_meta': buff_parametrized_fixtures['_meta'], + # pre-initialized fixtures + 'dut_index': dut_global_index, + '_pexpect_logfile': _pexpect_logfile, + 'pexpect_proc': pexpect_proc, + 'msg_queue': msg_queue, + } + + _fixture_classes_and_options = _fixture_classes_and_options_fn(**_kwargs) + + app = app_fn(_fixture_classes_and_options) + + openocd = openocd_gn(_fixture_classes_and_options) + layout.append(openocd) + + gdb = gdb_gn(_fixture_classes_and_options) + layout.append(gdb) + + serial = serial_gn(_fixture_classes_and_options, msg_queue, app) + layout.append(serial) + + qemu = qemu_gn(_fixture_classes_and_options, app) + layout.append(qemu) + + wokwi = wokwi_gn(_fixture_classes_and_options, app) + layout.append(wokwi) + + dut = dut_gn(_fixture_classes_and_options, openocd, gdb, app, serial, qemu, wokwi) + layout.append(dut) + + cls.obj_stack.append(layout) + return dut + + except Exception as e: + while layout: + obj = layout.pop() + _close_or_terminate(obj) + del layout + raise e diff --git a/pytest-embedded/pytest_embedded/plugin.py b/pytest-embedded/pytest_embedded/plugin.py index 07f4e622..e9767018 100644 --- a/pytest-embedded/pytest_embedded/plugin.py +++ b/pytest-embedded/pytest_embedded/plugin.py @@ -14,10 +14,8 @@ import tempfile import typing as t import xml.dom.minidom -from collections import Counter, defaultdict -from dataclasses import dataclass +from collections import Counter from operator import itemgetter -from pathlib import Path import pytest from _pytest.config import Config @@ -34,17 +32,32 @@ from .app import App from .dut import Dut +from .dut_factory import ( + DutFactory, + _fixture_classes_and_options_fn, + _listener_gn, + _pexpect_fr_gn, + app_fn, + dut_gn, + gdb_gn, + msg_queue_gn, + openocd_gn, + pexpect_proc_fn, + qemu_gn, + serial_gn, + set_buff_parametrized_fixtures, + wokwi_gn, +) from .log import MessageQueue, PexpectProcess from .unity import JunitMerger, escape_illegal_xml_chars from .utils import ( - FIXTURES_SERVICES, SERVICE_LIB_NAMES, + ClassCliOptions, Meta, PackageNotInstalledError, UnknownServiceError, find_by_suffix, to_list, - to_str, ) if t.TYPE_CHECKING: @@ -197,7 +210,7 @@ def pytest_addoption(parser): idf_group.addoption( '--panic-output-decode-script', help='Panic output decode script that is used in conjunction with the check-panic-coredump option ' - 'to parse panic output. (Default: use gdb_panic_server.py from package esp_idf_panic_decoder)', + 'to parse panic output. (Default: $IDF_PATH/tools/gdb_panic_server.py)', ) jtag_group = parser.getgroup('embedded-jtag') @@ -293,10 +306,6 @@ def _str_bool(v: str) -> t.Union[bool, str, None]: return v -def _drop_none_kwargs(kwargs: t.Dict[t.Any, t.Any]): - return {k: v for k, v in kwargs.items() if v is not None} - - def _prettify_xml(file_path: str): dom = xml.dom.minidom.parse(file_path) pretty_xml_as_string = dom.toprettyxml() @@ -636,14 +645,11 @@ def _pexpect_logfile(test_case_tempdir, logfile_extension, dut_index, dut_total) if sys.platform == 'darwin': multiprocessing.set_start_method('fork') -_ctx = multiprocessing.get_context() -_stdout = sys.__stdout__ - @pytest.fixture @multi_dut_generator_fixture def msg_queue() -> MessageQueue: # kwargs passed by `multi_dut_generator_fixture()` - return MessageQueue(ctx=_ctx) + return msg_queue_gn() @pytest.fixture @@ -653,46 +659,6 @@ def with_timestamp(request: FixtureRequest) -> bool: return _request_param_or_config_option_or_default(request, 'with_timestamp', None) -def _listen(q: MessageQueue, filepath: str, with_timestamp: bool = True, count: int = 1, total: int = 1) -> None: - _added_prefix = False - while True: - msg = q.get() - if not msg: - continue - - with open(filepath, 'ab') as fw: - fw.write(msg) - fw.flush() - - _s = to_str(msg) - if not _s: - continue - - prefix = '' - if total > 1: - source = f'dut-{count}' - else: - source = None - - if source: - prefix = f'[{source}] ' + prefix - - if with_timestamp: - prefix = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') + ' ' + prefix - - if not _added_prefix: - _s = prefix + _s - _added_prefix = True - _s = _s.replace('\r\n', '\n') # remove extra \r. since multi-dut \r would mess up the log - _s = _s.replace('\n', '\n' + prefix) - if prefix and _s.endswith(prefix): - _s = _s.rsplit(prefix, maxsplit=1)[0] - _added_prefix = False - - _stdout.write(_s) - _stdout.flush() - - @pytest.fixture @multi_dut_generator_fixture def _listener(msg_queue, _pexpect_logfile, with_timestamp, dut_index, dut_total) -> multiprocessing.Process: @@ -703,29 +669,13 @@ def _listener(msg_queue, _pexpect_logfile, with_timestamp, dut_index, dut_total) 1. print the string to `sys.stdout` 2. write the string to `_pexpect_logfile` """ - os.makedirs(os.path.dirname(_pexpect_logfile), exist_ok=True) - kwargs = { - 'with_timestamp': with_timestamp, - 'count': dut_index, - 'total': dut_total, - } - - return _ctx.Process( - target=_listen, - args=( - msg_queue, - _pexpect_logfile, - ), - kwargs=_drop_none_kwargs(kwargs), - ) + return _listener_gn(**locals()) @pytest.fixture @multi_dut_generator_fixture def _pexpect_fr(_pexpect_logfile, _listener) -> t.BinaryIO: - Path(_pexpect_logfile).touch() - _listener.start() - return open(_pexpect_logfile, 'rb') + return _pexpect_fr_gn(**locals()) @pytest.fixture @@ -735,7 +685,7 @@ def _pexpect_fr(_pexpect_logfile, _listener) -> t.BinaryIO: @multi_dut_fixture def pexpect_proc(_pexpect_fr) -> PexpectProcess: """Pexpect process that run the expect functions on""" - return PexpectProcess(_pexpect_fr) + return pexpect_proc_fn(**locals()) @pytest.fixture @@ -1025,16 +975,9 @@ def _services(embedded_services: t.Optional[str]) -> t.List[str]: return ['base', *services] -@dataclass -class ClassCliOptions: - classes: t.Dict[str, type] - mixins: t.Dict[str, t.List[type]] - kwargs: t.Dict[str, t.Dict[str, t.Any]] - - -@pytest.fixture +@pytest.fixture(autouse=True) @multi_dut_fixture -def _fixture_classes_and_options( +def parametrize_fixtures( _services, # parametrize fixtures app_path, @@ -1068,13 +1011,29 @@ def _fixture_classes_and_options( skip_regenerate_image, encrypt, keyfile, + # common fixtures + test_case_name, + _meta, +): + set_buff_parametrized_fixtures(locals()) + return locals() + + +@pytest.fixture(autouse=True) +def close_factory_duts(): + yield + DutFactory.close() + + +@pytest.fixture +@multi_dut_fixture +def _fixture_classes_and_options( + parametrize_fixtures, # pre-initialized fixtures dut_index, _pexpect_logfile, - test_case_name, pexpect_proc, msg_queue, - _meta, ) -> ClassCliOptions: """ classes: the class that the fixture should instantiate @@ -1094,221 +1053,10 @@ def _fixture_classes_and_options( ... } """ - classes: t.Dict[str, type] = {} - mixins: t.Dict[str, t.List[type]] = defaultdict(list) - kwargs: t.Dict[str, t.Dict[str, t.Any]] = defaultdict(dict) - - for fixture in FIXTURES_SERVICES.keys(): - if fixture == 'app': - kwargs['app'] = {'app_path': app_path, 'build_dir': build_dir} - if 'idf' in _services: - if 'qemu' in _services: - from pytest_embedded_qemu import DEFAULT_IMAGE_FN, QemuApp - - classes[fixture] = QemuApp - kwargs[fixture].update({ - 'msg_queue': msg_queue, - 'part_tool': part_tool, - 'qemu_image_path': qemu_image_path, - 'skip_regenerate_image': skip_regenerate_image, - 'encrypt': encrypt, - 'keyfile': keyfile, - 'qemu_prog_path': qemu_prog_path, - }) - else: - from pytest_embedded_idf import IdfApp - - classes[fixture] = IdfApp - kwargs[fixture].update({ - 'part_tool': part_tool, - }) - elif 'arduino' in _services: - from pytest_embedded_arduino import ArduinoApp - - classes[fixture] = ArduinoApp - else: - from .app import App - - classes[fixture] = App - elif fixture == 'serial': - if 'esp' in _services: - from pytest_embedded_serial_esp import EspSerial - - kwargs[fixture] = { - 'pexpect_proc': pexpect_proc, - 'msg_queue': msg_queue, - 'target': target, - 'beta_target': beta_target, - 'port': os.getenv('ESPPORT') or port, - 'port_location': port_location, - 'port_mac': port_mac, - 'baud': int(baud or EspSerial.DEFAULT_BAUDRATE), - 'esptool_baud': int(os.getenv('ESPBAUD') or esptool_baud or EspSerial.ESPTOOL_DEFAULT_BAUDRATE), - 'esp_flash_force': esp_flash_force, - 'skip_autoflash': skip_autoflash, - 'erase_all': erase_all, - 'meta': _meta, - } - if 'idf' in _services: - from pytest_embedded_idf import IdfSerial - - classes[fixture] = IdfSerial - kwargs[fixture].update({ - 'app': None, - 'confirm_target_elf_sha256': confirm_target_elf_sha256, - 'erase_nvs': erase_nvs, - }) - elif 'arduino' in _services: - from pytest_embedded_arduino import ArduinoSerial - - classes[fixture] = ArduinoSerial - kwargs[fixture].update({ - 'app': None, - }) - else: - from pytest_embedded_serial_esp import EspSerial - - classes[fixture] = EspSerial - elif 'serial' in _services or 'jtag' in _services: - from pytest_embedded_serial.serial import Serial - - classes[fixture] = Serial - kwargs[fixture] = { - 'msg_queue': msg_queue, - 'port': port, - 'port_location': port_location, - 'baud': int(baud or Serial.DEFAULT_BAUDRATE), - 'meta': _meta, - } - elif fixture in ['openocd', 'gdb']: - if 'jtag' in _services: - if fixture == 'openocd': - from pytest_embedded_jtag import OpenOcd - - classes[fixture] = OpenOcd - kwargs[fixture] = { - 'msg_queue': msg_queue, - 'app': None, - 'openocd_prog_path': openocd_prog_path, - 'openocd_cli_args': openocd_cli_args, - 'port_offset': dut_index, - 'meta': _meta, - } - elif not no_gdb: - from pytest_embedded_jtag import Gdb - - classes[fixture] = Gdb - kwargs[fixture] = { - 'msg_queue': msg_queue, - 'gdb_prog_path': gdb_prog_path, - 'gdb_cli_args': gdb_cli_args, - 'meta': _meta, - } - elif fixture == 'qemu': - if 'qemu' in _services: - from pytest_embedded_qemu import ( - DEFAULT_IMAGE_FN, - ENCRYPTED_IMAGE_FN, - Qemu, - ) - - classes[fixture] = Qemu - kwargs[fixture] = { - 'msg_queue': msg_queue, - 'qemu_image_path': qemu_image_path - or os.path.join( - app_path or '', build_dir or 'build', ENCRYPTED_IMAGE_FN if encrypt else DEFAULT_IMAGE_FN - ), - 'qemu_prog_path': qemu_prog_path, - 'qemu_cli_args': qemu_cli_args, - 'qemu_extra_args': qemu_extra_args, - 'app': None, - 'meta': _meta, - 'dut_index': dut_index, - } - elif fixture == 'wokwi': - if 'wokwi' in _services: - from pytest_embedded_wokwi import WokwiCLI - - classes[fixture] = WokwiCLI - kwargs[fixture].update({ - 'wokwi_cli_path': wokwi_cli_path, - 'wokwi_timeout': wokwi_timeout, - 'msg_queue': msg_queue, - 'app': None, - 'meta': _meta, - }) - elif fixture == 'dut': - classes[fixture] = Dut - kwargs[fixture] = { - 'pexpect_proc': pexpect_proc, - 'msg_queue': msg_queue, - 'app': None, - 'pexpect_logfile': _pexpect_logfile, - 'test_case_name': test_case_name, - 'meta': _meta, - } - if 'idf' in _services and 'esp' not in _services: - # esp,idf will use IdfDut, which based on IdfUnityDutMixin already - from pytest_embedded_idf.unity_tester import IdfUnityDutMixin - - mixins[fixture].append(IdfUnityDutMixin) - - if 'wokwi' in _services: - from pytest_embedded_wokwi import WokwiDut - - classes[fixture] = WokwiDut - kwargs[fixture].update({ - 'wokwi': None, - }) - - if 'idf' in _services: - from pytest_embedded_wokwi.idf import IDFFirmwareResolver - - kwargs['wokwi'].update({'firmware_resolver': IDFFirmwareResolver()}) - else: - raise SystemExit('wokwi service should be used together with idf service') - elif 'qemu' in _services: - from pytest_embedded_qemu import QemuDut - - classes[fixture] = QemuDut - kwargs[fixture].update({ - 'qemu': None, - }) - elif 'jtag' in _services: - if 'idf' in _services: - from pytest_embedded_idf import IdfDut - - classes[fixture] = IdfDut - else: - from pytest_embedded_serial import SerialDut - - classes[fixture] = SerialDut - - kwargs[fixture].update({ - 'serial': None, - 'openocd': None, - 'gdb': None, - }) - elif 'serial' in _services or 'esp' in _services: - if 'esp' in _services and 'idf' in _services: - from pytest_embedded_idf import IdfDut - - classes[fixture] = IdfDut - kwargs[fixture].update({ - 'skip_check_coredump': skip_check_coredump, - 'panic_output_decode_script': panic_output_decode_script, - }) - else: - from pytest_embedded_serial import SerialDut - - classes[fixture] = SerialDut + kwargs = locals() + kwargs.update(kwargs.pop('parametrize_fixtures')) - kwargs[fixture].update({ - 'serial': None, - }) - - return ClassCliOptions(classes, mixins, kwargs) + return _fixture_classes_and_options_fn(**kwargs) #################### @@ -1318,88 +1066,42 @@ def _fixture_classes_and_options( @multi_dut_fixture def app(_fixture_classes_and_options: ClassCliOptions) -> App: """A pytest fixture to gather information from the specified built binary folder""" - cls = _fixture_classes_and_options.classes['app'] - kwargs = _fixture_classes_and_options.kwargs['app'] - return cls(**_drop_none_kwargs(kwargs)) + return app_fn(**locals()) @pytest.fixture @multi_dut_generator_fixture def serial(_fixture_classes_and_options, msg_queue, app) -> t.Optional[t.Union['Serial', 'LinuxSerial']]: """A serial subprocess that could read/redirect/write""" - if hasattr(app, 'target') and app.target == 'linux': - from pytest_embedded_idf import LinuxSerial - - cls = LinuxSerial - kwargs = { - 'app': app, - 'msg_queue': msg_queue, - } - return cls(**kwargs) - - if 'serial' not in _fixture_classes_and_options.classes: - return None - - cls = _fixture_classes_and_options.classes['serial'] - kwargs = _fixture_classes_and_options.kwargs['serial'] - if 'app' in kwargs and kwargs['app'] is None: - kwargs['app'] = app - return cls(**_drop_none_kwargs(kwargs)) + return serial_gn(**locals()) @pytest.fixture @multi_dut_generator_fixture def openocd(_fixture_classes_and_options: ClassCliOptions) -> t.Optional['OpenOcd']: """An openocd subprocess that could read/redirect/write""" - if 'openocd' not in _fixture_classes_and_options.classes: - return None - - cls = _fixture_classes_and_options.classes['openocd'] - kwargs = _fixture_classes_and_options.kwargs['openocd'] - return cls(**_drop_none_kwargs(kwargs)) + return openocd_gn(**locals()) @pytest.fixture @multi_dut_generator_fixture def gdb(_fixture_classes_and_options: ClassCliOptions) -> t.Optional['Gdb']: """A gdb subprocess that could read/redirect/write""" - if 'gdb' not in _fixture_classes_and_options.classes: - return None - - cls = _fixture_classes_and_options.classes['gdb'] - kwargs = _fixture_classes_and_options.kwargs['gdb'] - return cls(**_drop_none_kwargs(kwargs)) + return gdb_gn(**locals()) @pytest.fixture @multi_dut_generator_fixture def qemu(_fixture_classes_and_options: ClassCliOptions, app) -> t.Optional['Qemu']: """A qemu subprocess that could read/redirect/write""" - if 'qemu' not in _fixture_classes_and_options.classes: - return None - - cls = _fixture_classes_and_options.classes['qemu'] - kwargs = _fixture_classes_and_options.kwargs['qemu'] - - if 'app' in kwargs and kwargs['app'] is None: - kwargs['app'] = app - - return cls(**_drop_none_kwargs(kwargs)) + return qemu_gn(**locals()) @pytest.fixture @multi_dut_generator_fixture def wokwi(_fixture_classes_and_options: ClassCliOptions, app) -> t.Optional['WokwiCLI']: """A wokwi subprocess that could read/redirect/write""" - if 'wokwi' not in _fixture_classes_and_options.classes: - return None - - cls = _fixture_classes_and_options.classes['wokwi'] - kwargs = _fixture_classes_and_options.kwargs['wokwi'] - - if 'app' in kwargs and kwargs['app'] is None: - kwargs['app'] = app - return cls(**_drop_none_kwargs(kwargs)) + return wokwi_gn(**locals()) @pytest.fixture @@ -1417,34 +1119,7 @@ def 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. """ - kwargs = _fixture_classes_and_options.kwargs['dut'] - mixins = _fixture_classes_and_options.mixins['dut'] - - # since there's no way to know the target before setup finished - # we have to use the `app.target` to determine the dut class here - if hasattr(app, 'target') and app.target == 'linux': - from pytest_embedded_idf import LinuxDut - - cls = LinuxDut - kwargs['serial'] = None # replace it later with LinuxSerial - else: - cls = _fixture_classes_and_options.classes['dut'] - - for k, v in kwargs.items(): - if v is None: - if k == 'app': - kwargs[k] = app - elif k == 'serial': - kwargs[k] = serial - elif k == 'openocd': - kwargs[k] = openocd - elif k == 'gdb': - kwargs[k] = gdb - elif k == 'qemu': - kwargs[k] = qemu - elif k == 'wokwi': - kwargs[k] = wokwi - return cls(**_drop_none_kwargs(kwargs), mixins=mixins) + return dut_gn(**locals()) @pytest.fixture diff --git a/pytest-embedded/pytest_embedded/utils.py b/pytest-embedded/pytest_embedded/utils.py index 9b486dbb..3e7173c1 100644 --- a/pytest-embedded/pytest_embedded/utils.py +++ b/pytest-embedded/pytest_embedded/utils.py @@ -5,6 +5,7 @@ import os import re import typing as t +from dataclasses import dataclass if t.TYPE_CHECKING: from . import App @@ -34,6 +35,14 @@ 'dut': ['base', 'serial', 'jtag', 'qemu', 'idf', 'wokwi'], } + +@dataclass +class ClassCliOptions: + classes: t.Dict[str, type] + mixins: t.Dict[str, t.List[type]] + kwargs: t.Dict[str, t.Dict[str, t.Any]] + + _T = t.TypeVar('_T') _MIXIN_REQUIRED_SERVICES = {