Repository: espressif/pytest-embedded (language: python)

Issue:
[qemu,idf] `QemuDut` based on `Dut` even when service `idf` is enabled (RDT-319)
- put some IDF non-real-target related functions as a mixin
- if `idf` service is enabled, inject the mixin into the `QemuDut` class


Discussion / hints:

Fix patch:
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 878075e7..2a7cb9f6 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -27,3 +27,4 @@ repos:
     rev: 'v0.0.255'
     hooks:
       - id: ruff
+        args: ['--fix']
diff --git a/docs/requirements.txt b/docs/requirements.txt
index 1a8ad18a..fefa247a 100644
--- a/docs/requirements.txt
+++ b/docs/requirements.txt
@@ -3,3 +3,4 @@ mkdocs
 mkdocstrings[python]
 mkdocs-material
 plantuml-markdown
+griffe<0.27.4


Test patch:
diff --git a/.github/workflows/test-build-docs.yml b/.github/workflows/test-build-docs.yml
index 2399eb6c..67fd848d 100644
--- a/.github/workflows/test-build-docs.yml
+++ b/.github/workflows/test-build-docs.yml
@@ -21,8 +21,7 @@ jobs:
           python -m pip install --upgrade pip
           pip install -r requirements.txt
           bash foreach.sh install
-          cd docs
-          pip install -r requirements.txt
+          pip install -r docs/requirements.txt
       - name: Test build docs
         run: |
           mkdocs build
diff --git a/pytest-embedded-idf/pytest_embedded_idf/app.py b/pytest-embedded-idf/pytest_embedded_idf/app.py
index 55d5abf7..6368f53e 100644
--- a/pytest-embedded-idf/pytest_embedded_idf/app.py
+++ b/pytest-embedded-idf/pytest_embedded_idf/app.py
@@ -26,6 +26,9 @@ class IdfApp(App):
         flash_settings (dict[str, Any]): dict of flash settings
     """
 
+    XTENSA_TARGETS = ['esp32', 'esp32s2', 'esp32s3']
+    RISCV32_TARGETS = ['esp32c3', 'esp32h2', 'esp32c2', 'esp32c6']
+
     FLASH_ARGS_FILENAME = 'flash_args'
     FLASH_PROJECT_ARGS_FILENAME = 'flash_project_args'
     FLASH_ARGS_JSON_FILENAME = 'flasher_args.json'
@@ -110,6 +113,26 @@ def target(self) -> str:
         else:
             return self.flash_args.get('extra_esptool_args', {}).get('chip', 'esp32')
 
+    @property
+    def is_riscv32(self):
+        if self.sdkconfig.get('IDF_TARGET_ARCH_RISCV'):
+            return True
+
+        if self.target in self.RISCV32_TARGETS:
+            return True
+
+        return False
+
+    @property
+    def is_xtensa(self):
+        if self.sdkconfig.get('IDF_TARGET_ARCH_XTENSA'):
+            return True
+
+        if self.target in self.XTENSA_TARGETS:
+            return True
+
+        return False
+
     @property
     def partition_table(self) -> Dict[str, Any]:
         """
diff --git a/pytest-embedded-idf/pytest_embedded_idf/dut.py b/pytest-embedded-idf/pytest_embedded_idf/dut.py
index f7741334..f4da758c 100644
--- a/pytest-embedded-idf/pytest_embedded_idf/dut.py
+++ b/pytest-embedded-idf/pytest_embedded_idf/dut.py
@@ -1,29 +1,22 @@
-import functools
 import logging
 import os
 import re
 import subprocess
 import sys
 import tempfile
-import time
 import typing as t
 import warnings
 from contextlib import redirect_stdout
 
-from pytest_embedded.unity import (
-    UNITY_BASIC_REGEX,
-    UNITY_FIXTURE_REGEX,
-    UNITY_SUMMARY_LINE_REGEX,
-    TestCase,
-)
-from pytest_embedded.utils import UserHint, remove_asci_color_code, to_str
+from pytest_embedded.utils import UserHint, to_str
 from pytest_embedded_serial.dut import SerialDut
+from pytest_embedded_serial_esp import EspSerial
 
 from .app import IdfApp
-from .unity_tester import READY_PATTERN_LIST, UnittestMenuCase
+from .unity_tester import IdfUnityDutMixin
 
 
-class IdfDut(SerialDut):
+class IdfDut(IdfUnityDutMixin, SerialDut):
     """
     Dut class for serial ports connect to Espressif boards which are flashed with ESP-IDF apps
 
@@ -32,8 +25,8 @@ class IdfDut(SerialDut):
         skip_check_coredump (bool): skip check core dumped or not while dut teardown if set to True
     """
 
-    XTENSA_TARGETS = ['esp32', 'esp32s2', 'esp32s3']
-    RISCV32_TARGETS = ['esp32c3', 'esp32h2', 'esp32c2']
+    XTENSA_TARGETS = IdfApp.XTENSA_TARGETS
+    RISCV32_TARGETS = IdfApp.RISCV32_TARGETS
 
     COREDUMP_UART_START = b'================= CORE DUMP START ================='
     COREDUMP_UART_END = b'================= CORE DUMP END ================='
@@ -44,6 +37,7 @@ class IdfDut(SerialDut):
     PANIC_END = b'ELF file SHA256:'
 
     app: IdfApp
+    serial: EspSerial
 
     def __init__(
         self,
@@ -55,19 +49,20 @@ def __init__(
         self.target = app.target
         self.skip_check_coredump = skip_check_coredump
         self._panic_output_decode_script = panic_output_decode_script
-        self._test_menu: t.List[UnittestMenuCase] = None  # type: ignore
 
         super().__init__(app=app, **kwargs)
 
+        self._hard_reset_func = self.serial.hard_reset
+
     @property
     def toolchain_prefix(self) -> str:
         """
         Returns:
             Toolchain prefix according to the `self.target`
         """
-        if self.target in self.XTENSA_TARGETS:
+        if self.app.is_xtensa:
             return f'xtensa-{self.target}-elf-'
-        elif self.target in self.RISCV32_TARGETS:
+        elif self.app.is_riscv32:
             return 'riscv32-esp-elf-'
         else:
             raise ValueError(f'Unknown target: {self.target}')
@@ -228,305 +223,6 @@ def close(self) -> None:
                 logging.debug(e)
         super().close()
 
-    #####################
-    # IDF-unity related #
-    #####################
-    def _parse_test_menu(
-        self,
-        ready_line: str = 'Press ENTER to see the list of tests',
-        pattern="Here's the test menu, pick your combo:(.+)Enter test for running.",
-        trigger: str = '',
-    ) -> t.List[UnittestMenuCase]:
-        """
-        Get test case list from test menu via UART print.
-
-        Args:
-            ready_line: Prompt to indicate that device is ready to print test menu.
-            pattern: Pattern to match the output from device, menu block should be in the first group.
-                     This will be directly passed to `pexpect.expect()`.
-            trigger: Keys to trigger device to print test menu by UART.
-
-        Returns:
-            A `list` of `UnittestMenuCase`, which includes info for each test case.
-        """
-        self.expect_exact(ready_line)
-        self.write(trigger)
-        menu_block = self.expect(pattern).group(1)
-        s = str(menu_block, encoding='UTF-8')
-        return self._parse_unity_menu_from_str(s)
-
-    def parse_test_menu(
-        self,
-        ready_line: str = 'Press ENTER to see the list of tests',
-        pattern="Here's the test menu, pick your combo:(.+)Enter test for running.",
-        trigger: str = '',
-    ) -> t.List[UnittestMenuCase]:
-        warnings.warn(
-            'Please use `dut.test_menu` property directly, '
-            'will rename this function to `_parse_test_menu` in release 2.0.0',
-            DeprecationWarning,
-        )
-
-        return self._parse_test_menu(ready_line, pattern, trigger)
-
-    @staticmethod
-    def parse_unity_menu_from_str(s: str) -> t.List[UnittestMenuCase]:
-        warnings.warn(
-            'Please use `dut.test_menu` property directly, '
-            'will rename this function to `_parse_unity_menu_from_str` in release 2.0.0',
-            DeprecationWarning,
-        )
-
-        return IdfDut._parse_unity_menu_from_str(s)
-
-    @staticmethod
-    def _parse_unity_menu_from_str(s: str) -> t.List[UnittestMenuCase]:
-        """
-        Parse test case menu from string to list of `UnittestMenuCase`.
-
-        Args:
-            s: string include test case menu.
-
-        Returns:
-            A `list` of `UnittestMenuCase`, which includes info for each test case.
-        """
-        cases = s.splitlines()
-
-        case_regex = re.compile(r'\((\d+)\)\s\"(.+)\"\s(\[.+\])+')
-        subcase_regex = re.compile(r'\t\((\d+)\)\s\"(.+)\"')
-
-        test_menu = []
-        for case in cases:
-            case_match = case_regex.match(case)
-            if case_match is not None:
-                index, name, tag_block = case_match.groups()
-                tags = re.findall(r'\[(.+?)\]', tag_block)
-
-                if 'multi_stage' in tags:
-                    _type = 'multi_stage'
-                    tags.remove('multi_stage')
-                elif 'multi_device' in tags:
-                    _type = 'multi_device'
-                    tags.remove('multi_device')
-                else:
-                    _type = 'normal'
-
-                keyword = []
-                if 'ignore' in tags:
-                    keyword.append('ignore')
-                    tags.remove('ignore')
-                elif 'disable' in tags:
-                    keyword = 'disable'
-                    tags.remove('disable')
-
-                attributes = {}
-                group = []
-                for tag in tags:
-                    if '=' in tag:
-                        k, v = tag.replace(' ', '').split('=')
-                        attributes[k] = v
-                    else:
-                        group.append(tag)
-
-                test_menu.append(
-                    UnittestMenuCase(
-                        index=int(index),
-                        name=name,
-                        type=_type,
-                        keywords=keyword,
-                        groups=group,
-                        attributes=attributes,
-                        subcases=[],
-                    )
-                )
-                continue
-            subcase_match = subcase_regex.match(case)
-            if subcase_match is not None:
-                index, name = subcase_match.groups()
-                test_menu[-1].subcases.append({'index': int(index), 'name': name})
-                continue
-
-            if case != '':
-                raise NotImplementedError('Unrecognized test case:', case)
-
-        return test_menu
-
-    @property
-    def test_menu(self) -> t.List[UnittestMenuCase]:
-        if self._test_menu is None:
-            self._test_menu = self._parse_test_menu()
-            logging.debug('Successfully parsed unity test menu')
-            self.serial.hard_reset()
-
-        return self._test_menu
-
-    def _record_single_unity_test_case(func):
-        """
-        The first argument of the function that is using this decorator must be `case`. passing with args.
-
-        Notes:
-            This function is better than `dut.expect_unity_output()` since it will record the test case even it core
-                dumped during running the test case or other reasons that cause the final result block is uncaught.
-        """
-
-        @functools.wraps(func)
-        def wrapper(self, *args, **kwargs):
-            _start_at = time.perf_counter()  # declare here in case hard reset failed
-            _timeout = kwargs.get('timeout', 30)
-            _case = args[0]
-
-            try:
-                # do it here since the first hard reset before test case shouldn't be counted in duration time
-                if 'reset' in kwargs:
-                    if kwargs.pop('reset'):
-                        self.serial.hard_reset()
-
-                _start_at = time.perf_counter()
-                func(self, *args, **kwargs)
-            finally:
-                _timestamp = time.perf_counter()
-                _log = ''
-                try:
-                    _timeout = _timeout - _timestamp + _start_at
-                    if _timeout < 0:  # pexpect process would expect 30s if < 0
-                        _timeout = 0
-                    self.expect(UNITY_SUMMARY_LINE_REGEX, timeout=_timeout)
-                except Exception:  # result block missing # noqa
-                    pass
-                else:  # result block exists
-                    _log = remove_asci_color_code(self.pexpect_proc.before)
-                finally:
-                    _end_at = time.perf_counter()
-                    self._add_single_unity_test_case(
-                        _case, _log, additional_attrs={'time': round(_end_at - _start_at, 3)}
-                    )
-
-        return wrapper
-
-    def _add_single_unity_test_case(
-        self, case: UnittestMenuCase, log: t.Optional[t.AnyStr], additional_attrs: t.Optional[t.Dict[str, t.Any]] = None
-    ):
-        if log:
-            # check format
-            check = UNITY_FIXTURE_REGEX.search(log)
-            if check:
-                regex = UNITY_FIXTURE_REGEX
-            else:
-                regex = UNITY_BASIC_REGEX
-
-            res = list(regex.finditer(log))
-        else:
-            res = []
-
-        # real parsing
-        if len(res) == 0:
-            logging.warning(f'unity test case not found, use case {case.name} instead')
-            attrs = {'name': case.name, 'result': 'FAIL', 'message': self.pexpect_proc.buffer_debug_str}
-        elif len(res) == 1:
-            attrs = {k: v for k, v in res[0].groupdict().items() if v is not None}
-        else:
-            warnings.warn('This function is for recording single unity test case only. Use the last matched one')
-            attrs = {k: v for k, v in res[-1].groupdict().items() if v is not None}
-
-        if additional_attrs:
-            attrs.update(additional_attrs)
-
-        if log:
-            attrs.update({'stdout': log})
-
-        testcase = TestCase(**attrs)
-        self.testsuite.testcases.append(testcase)
-        if testcase.result == 'FAIL':
-            self.testsuite.attrs['failures'] += 1
-        elif testcase.result == 'IGNORE':
-            self.testsuite.attrs['skipped'] += 1
-        else:
-            self.testsuite.attrs['tests'] += 1
-
-    @_record_single_unity_test_case
-    def _run_normal_case(
-        self,
-        case: UnittestMenuCase,
-        reset: bool = False,
-        timeout: float = 30,
-    ) -> None:
-        """
-        Run a specific normal case
-
-        Notes:
-            Will skip with a warning if the case type is not "normal"
-
-        Args:
-            case: the specific case that parsed in test menu
-            reset: whether to perform a hardware reset before running a case
-            timeout: timeout. (Default: 30 seconds)
-        """
-        if case.type != 'normal':
-            logging.warning('case %s is not a normal case', case.name)
-            return
-
-        self.expect_exact(READY_PATTERN_LIST, timeout=timeout)
-        self.write(str(case.index))
-        self.expect_exact(f'Running {case.name}...', timeout=1)
-
-    @_record_single_unity_test_case
-    def _run_multi_stage_case(
-        self,
-        case: UnittestMenuCase,
-        reset: bool = False,
-        timeout: float = 30,
-    ) -> None:
-        """
-        Run a specific multi_stage case
-
-        Notes:
-            Will skip with a warning if the case type is not "multi_stage"
-
-        Args:
-            case: the specific case that parsed in test menu
-            reset: whether to perform a hardware reset before running a case
-            timeout: timeout. (Default: 30 seconds)
-        """
-        if case.type != 'multi_stage':
-            logging.warning('case %s is not a multi stage case', case.name)
-            return
-
-        _start_at = time.perf_counter()
-        _timestamp = _start_at
-        for sub_case in case.subcases:
-            _timeout = timeout - _timestamp + _start_at
-            if _timeout < 0:  # pexpect process would expect 30s if < 0
-                _timeout = 0
-            self.expect_exact(READY_PATTERN_LIST, timeout=_timeout)
-            self.write(str(case.index))
-            self.expect_exact(case.name, timeout=1)
-            self.write(str(sub_case['index']))
-            _timestamp = time.perf_counter()
-
-    def run_all_single_board_cases(
-        self,
-        group: t.Optional[str] = None,
-        reset: bool = False,
-        timeout: float = 30,
-        run_ignore_cases: bool = False,
-    ):
-        """
-        Run all multi_stage cases
-
-        Args:
-            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
-        """
-        for case in self.test_menu:
-            if not group or group in case.groups:
-                if not case.is_ignored or run_ignore_cases:
-                    if case.type == 'normal':
-                        self._run_normal_case(case, reset=reset, timeout=timeout)
-                    elif case.type == 'multi_stage':
-                        self._run_multi_stage_case(case, reset=reset, timeout=timeout)
-
     def write(self, data: t.AnyStr) -> None:
         data_str = to_str(data).strip('\n') or ''
         if data_str == '*':
diff --git a/pytest-embedded-idf/pytest_embedded_idf/unity_tester.py b/pytest-embedded-idf/pytest_embedded_idf/unity_tester.py
index db07b96a..07b1a70f 100644
--- a/pytest-embedded-idf/pytest_embedded_idf/unity_tester.py
+++ b/pytest-embedded-idf/pytest_embedded_idf/unity_tester.py
@@ -1,14 +1,24 @@
 # SPDX-FileCopyrightText: 2022 Espressif Systems (Shanghai) CO LTD
 # SPDX-License-Identifier: Apache-2.0
+import functools
 import logging
+import re
 import time
 import typing as t
+import warnings
 from collections.abc import Iterable
 from dataclasses import dataclass
 from threading import Semaphore, Thread
 
+import pexpect
 from pexpect.exceptions import TIMEOUT
-from pytest_embedded import unity, utils
+from pytest_embedded.unity import (
+    UNITY_BASIC_REGEX,
+    UNITY_FIXTURE_REGEX,
+    UNITY_SUMMARY_LINE_REGEX,
+    TestCase,
+)
+from pytest_embedded.utils import remove_asci_color_code
 
 if t.TYPE_CHECKING:
     from .dut import IdfDut
@@ -53,6 +63,349 @@ def is_ignored(self):
         return 'ignore' in self.keywords or 'disable' in self.keywords
 
 
+class IdfUnityDutMixin:
+    """
+    This mixin class provide ESP-IDF modified unity test framework related functions.
+    """
+
+    def __init__(self, *args, **kwargs):
+        self._test_menu: t.List[UnittestMenuCase] = None  # type: ignore
+
+        self._hard_reset_func: t.Optional[t.Callable] = None
+
+        super().__init__(*args, **kwargs)
+
+    def confirm_write(self, write_str: t.Any, expect_str: str, timeout: int = 1, retry_times: int = 3):
+        err = None
+        for _ in range(retry_times):
+            try:
+                self.write(str(write_str))
+                res = self.expect(expect_str, timeout=timeout)
+            except pexpect.TIMEOUT as e:
+                err = e
+            else:
+                break
+        else:
+            raise err
+
+        return res
+
+    def _parse_test_menu(
+        self,
+        ready_line: str = 'Press ENTER to see the list of tests',
+        pattern="Here's the test menu, pick your combo:(.+)Enter test for running.",
+        trigger: str = '',
+    ) -> t.List[UnittestMenuCase]:
+        """
+        Get test case list from test menu via UART print.
+
+        Args:
+            ready_line: Prompt to indicate that device is ready to print test menu.
+            pattern: Pattern to match the output from device, menu block should be in the first group.
+                     This will be directly passed to `pexpect.expect()`.
+            trigger: Keys to trigger device to print test menu by UART.
+
+        Returns:
+            A `list` of `UnittestMenuCase`, which includes info for each test case.
+        """
+        self.expect_exact(ready_line)
+        res = self.confirm_write(trigger, pattern)
+        return self._parse_unity_menu_from_str(res.group(1).decode('utf8'))
+
+    def parse_test_menu(
+        self,
+        ready_line: str = 'Press ENTER to see the list of tests',
+        pattern="Here's the test menu, pick your combo:(.+)Enter test for running.",
+        trigger: str = '',
+    ) -> t.List[UnittestMenuCase]:
+        warnings.warn(
+            'Please use `dut.test_menu` property directly, '
+            'will rename this function to `_parse_test_menu` in release 2.0.0',
+            DeprecationWarning,
+        )
+
+        return self._parse_test_menu(ready_line, pattern, trigger)
+
+    @staticmethod
+    def parse_unity_menu_from_str(s: str) -> t.List[UnittestMenuCase]:
+        warnings.warn(
+            'Please use `dut.test_menu` property directly, '
+            'will rename this function to `_parse_unity_menu_from_str` in release 2.0.0',
+            DeprecationWarning,
+        )
+
+        return IdfUnityDutMixin._parse_unity_menu_from_str(s)
+
+    @staticmethod
+    def _parse_unity_menu_from_str(s: str) -> t.List[UnittestMenuCase]:
+        """
+        Parse test case menu from string to list of `UnittestMenuCase`.
+
+        Args:
+            s: string include test case menu.
+
+        Returns:
+            A `list` of `UnittestMenuCase`, which includes info for each test case.
+        """
+        cases = s.splitlines()
+
+        case_regex = re.compile(r'\((\d+)\)\s\"(.+)\"\s(\[.+\])+')
+        subcase_regex = re.compile(r'\t\((\d+)\)\s\"(.+)\"')
+
+        test_menu = []
+        for case in cases:
+            case_match = case_regex.match(case)
+            if case_match is not None:
+                index, name, tag_block = case_match.groups()
+                tags = re.findall(r'\[(.+?)\]', tag_block)
+
+                if 'multi_stage' in tags:
+                    _type = 'multi_stage'
+                    tags.remove('multi_stage')
+                elif 'multi_device' in tags:
+                    _type = 'multi_device'
+                    tags.remove('multi_device')
+                else:
+                    _type = 'normal'
+
+                keyword = []
+                if 'ignore' in tags:
+                    keyword.append('ignore')
+                    tags.remove('ignore')
+                elif 'disable' in tags:
+                    keyword = 'disable'
+                    tags.remove('disable')
+
+                attributes = {}
+                group = []
+                for tag in tags:
+                    if '=' in tag:
+                        k, v = tag.replace(' ', '').split('=')
+                        attributes[k] = v
+                    else:
+                        group.append(tag)
+
+                test_menu.append(
+                    UnittestMenuCase(
+                        index=int(index),
+                        name=name,
+                        type=_type,
+                        keywords=keyword,
+                        groups=group,
+                        attributes=attributes,
+                        subcases=[],
+                    )
+                )
+                continue
+            subcase_match = subcase_regex.match(case)
+            if subcase_match is not None:
+                index, name = subcase_match.groups()
+                test_menu[-1].subcases.append({'index': int(index), 'name': name})
+                continue
+
+            if case != '':
+                raise NotImplementedError('Unrecognized test case:', case)
+
+        return test_menu
+
+    @property
+    def test_menu(self) -> t.List[UnittestMenuCase]:
+        if self._test_menu is None:
+            self._test_menu = self._parse_test_menu()
+            logging.debug('Successfully parsed unity test menu')
+            if self._hard_reset_func:
+                try:
+                    self._hard_reset_func()
+                except NotImplementedError:
+                    self.write('\n')  # print the menu
+
+        return self._test_menu
+
+    def _record_single_unity_test_case(func):
+        """
+        The first argument of the function that is using this decorator must be `case`. passing with args.
+
+        Notes:
+            This function is better than `dut.expect_unity_output()` since it will record the test case even it core
+                dumped during running the test case or other reasons that cause the final result block is uncaught.
+        """
+
+        @functools.wraps(func)
+        def wrapper(self, *args, **kwargs):
+            _start_at = time.perf_counter()  # declare here in case hard reset failed
+            _timeout = kwargs.get('timeout', 30)
+            _case = args[0]
+
+            try:
+                # do it here since the first hard reset before test case shouldn't be counted in duration time
+                if 'reset' in kwargs:
+                    if kwargs.pop('reset') and self._hard_reset_func:
+                        try:
+                            self._hard_reset_func()
+                        except NotImplementedError:
+                            self.write('\n')  # print the menu
+
+                _start_at = time.perf_counter()
+                func(self, *args, **kwargs)
+            finally:
+                _timestamp = time.perf_counter()
+                _log = ''
+                try:
+                    _timeout = _timeout - _timestamp + _start_at
+                    if _timeout < 0:  # pexpect process would expect 30s if < 0
+                        _timeout = 0
+                    self.expect(UNITY_SUMMARY_LINE_REGEX, timeout=_timeout)
+                except Exception:  # result block missing # noqa
+                    pass
+                else:  # result block exists
+                    _log = remove_asci_color_code(self.pexpect_proc.before)
+                finally:
+                    _end_at = time.perf_counter()
+                    self._add_single_unity_test_case(
+                        _case, _log, additional_attrs={'time': round(_end_at - _start_at, 3)}
+                    )
+
+        return wrapper
+
+    def _add_single_unity_test_case(
+        self, case: UnittestMenuCase, log: t.Optional[t.AnyStr], additional_attrs: t.Optional[t.Dict[str, t.Any]] = None
+    ):
+        if log:
+            # check format
+            check = UNITY_FIXTURE_REGEX.search(log)
+            if check:
+                regex = UNITY_FIXTURE_REGEX
+            else:
+                regex = UNITY_BASIC_REGEX
+
+            res = list(regex.finditer(log))
+        else:
+            res = []
+
+        # real parsing
+        if len(res) == 0:
+            logging.warning(f'unity test case not found, use case {case.name} instead')
+            attrs = {'name': case.name, 'result': 'FAIL', 'message': self.pexpect_proc.buffer_debug_str}
+        elif len(res) == 1:
+            attrs = {k: v for k, v in res[0].groupdict().items() if v is not None}
+        else:
+            warnings.warn('This function is for recording single unity test case only. Use the last matched one')
+            attrs = {k: v for k, v in res[-1].groupdict().items() if v is not None}
+
+        if additional_attrs:
+            attrs.update(additional_attrs)
+
+        if log:
+            attrs.update({'stdout': log})
+
+        testcase = TestCase(**attrs)
+        self.testsuite.testcases.append(testcase)
+        if testcase.result == 'FAIL':
+            self.testsuite.attrs['failures'] += 1
+        elif testcase.result == 'IGNORE':
+            self.testsuite.attrs['skipped'] += 1
+        else:
+            self.testsuite.attrs['tests'] += 1
+
+    @_record_single_unity_test_case
+    def _run_normal_case(
+        self,
+        case: UnittestMenuCase,
+        reset: bool = False,
+        timeout: float = 30,
+    ) -> None:
+        """
+        Run a specific normal case
+
+        Notes:
+            Will skip with a warning if the case type is not "normal"
+
+        Args:
+            case: the specific case that parsed in test menu
+            reset: whether to perform a hardware reset before running a case
+            timeout: timeout. (Default: 30 seconds)
+        """
+        if case.type != 'normal':
+            logging.warning('case %s is not a normal case', case.name)
+            return
+
+        self.expect_exact(READY_PATTERN_LIST, timeout=timeout)
+        self.confirm_write(case.index, f'Running {case.name}...')
+
+    @_record_single_unity_test_case
+    def _run_multi_stage_case(
+        self,
+        case: UnittestMenuCase,
+        reset: bool = False,
+        timeout: float = 30,
+    ) -> None:
+        """
+        Run a specific multi_stage case
+
+        Notes:
+            Will skip with a warning if the case type is not "multi_stage"
+
+        Args:
+            case: the specific case that parsed in test menu
+            reset: whether to perform a hardware reset before running a case
+            timeout: timeout. (Default: 30 seconds)
+        """
+        if case.type != 'multi_stage':
+            logging.warning('case %s is not a multi stage case', case.name)
+            return
+
+        _start_at = time.perf_counter()
+        _timestamp = _start_at
+        for sub_case in case.subcases:
+            _timeout = timeout - _timestamp + _start_at
+            if _timeout < 0:  # pexpect process would expect 30s if < 0
+                _timeout = 0
+
+            self.expect_exact(READY_PATTERN_LIST, timeout=_timeout)
+            self.confirm_write(case.index, f'Running {case.name}...')
+
+            # here we can't use confirm_write becuase the sub cases won't print anything
+            time.sleep(1)
+            self.write(str(sub_case['index']))
+
+            _timestamp = time.perf_counter()
+
+    def run_single_board_case(self, name: str, reset: bool = False, timeout: float = 30) -> None:
+        for case in self.test_menu:
+            if case.name == name and case.type == 'normal':
+                self._run_normal_case(case, reset=reset, timeout=timeout)
+                break
+            elif case.name == name and case.type == 'multi_stage':
+                self._run_multi_stage_case(case, reset=reset, timeout=timeout)
+                break
+        else:
+            raise ValueError(f'single-board test case {name} not found')
+
+    def run_all_single_board_cases(
+        self,
+        group: t.Optional[str] = None,
+        reset: bool = False,
+        timeout: float = 30,
+        run_ignore_cases: bool = False,
+    ) -> None:
+        """
+        Run all multi_stage cases
+
+        Args:
+            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
+        """
+        for case in self.test_menu:
+            if not group or group in case.groups:
+                if not case.is_ignored or run_ignore_cases:
+                    if case.type == 'normal':
+                        self._run_normal_case(case, reset=reset, timeout=timeout)
+                    elif case.type == 'multi_stage':
+                        self._run_multi_stage_case(case, reset=reset, timeout=timeout)
+
+
 class MultiDevResource:
     """
     Resources of multi_dev dut
@@ -161,7 +514,7 @@ def _run(self, **kwargs) -> None:  # type: ignore
         signal_pattern_list = [
             self.UNITY_SEND_SIGNAL_REGEX,  # The dut send a signal
             self.UNITY_WAIT_SIGNAL_REGEX,  # The dut is blocked and waiting for a signal
-            unity.UNITY_SUMMARY_LINE_REGEX,  # Means the case finished
+            UNITY_SUMMARY_LINE_REGEX,  # Means the case finished
         ]
         dut = kwargs['dut']
         dev_res = kwargs['dev_res']
@@ -213,7 +566,7 @@ def _run(self, **kwargs) -> None:  # type: ignore
                     case_end_time = time.perf_counter()
                     case_duration = case_end_time - start_time
                     additional_attrs = {'time': round(case_duration, 3)}
-                    log = utils.remove_asci_color_code(dut.pexpect_proc.before)
+                    log = remove_asci_color_code(dut.pexpect_proc.before)
                     dut.testsuite.add_unity_test_cases(log, additional_attrs=additional_attrs)
                     break
 
diff --git a/pytest-embedded-qemu/pytest_embedded_qemu/app.py b/pytest-embedded-qemu/pytest_embedded_qemu/app.py
index 61bad7f8..55670e92 100644
--- a/pytest-embedded-qemu/pytest_embedded_qemu/app.py
+++ b/pytest-embedded-qemu/pytest_embedded_qemu/app.py
@@ -90,9 +90,6 @@ def __init__(
         if self.encrypt:
             self.encrypted_image_path = os.path.join(self.binary_path, ENCRYPTED_IMAGE_FN)
 
-        if self.target != 'esp32':
-            raise ValueError('For now on QEMU we only support ESP32')
-
         self.create_image()
 
     def create_image(self) -> None:
diff --git a/pytest-embedded-qemu/pytest_embedded_qemu/dut.py b/pytest-embedded-qemu/pytest_embedded_qemu/dut.py
index 1ea243b2..0da2488e 100644
--- a/pytest-embedded-qemu/pytest_embedded_qemu/dut.py
+++ b/pytest-embedded-qemu/pytest_embedded_qemu/dut.py
@@ -15,9 +15,11 @@ def __init__(
         qemu: Qemu,
         **kwargs,
     ) -> None:
+        self.qemu = qemu
+
         super().__init__(**kwargs)
 
-        self.qemu = qemu
+        self._hard_reset_func = self.qemu._hard_reset
 
     def write(self, s: AnyStr) -> None:
         self.qemu.write(s)
diff --git a/pytest-embedded-qemu/pytest_embedded_qemu/qemu.py b/pytest-embedded-qemu/pytest_embedded_qemu/qemu.py
index 4f0be253..e84e1f3d 100644
--- a/pytest-embedded-qemu/pytest_embedded_qemu/qemu.py
+++ b/pytest-embedded-qemu/pytest_embedded_qemu/qemu.py
@@ -4,7 +4,7 @@
 
 from pytest_embedded.log import DuplicateStdoutPopen
 
-from . import DEFAULT_IMAGE_FN
+from . import DEFAULT_IMAGE_FN, QemuApp
 
 
 class Qemu(DuplicateStdoutPopen):
@@ -15,7 +15,10 @@ class Qemu(DuplicateStdoutPopen):
     SOURCE = 'QEMU'
 
     QEMU_PROG_PATH = 'qemu-system-xtensa'
+    QEMU_PROG_FMT = 'qemu-system-{}'
+
     QEMU_DEFAULT_ARGS = '-nographic -machine esp32'
+    QEMU_DEFAULT_FMT = '-nographic -machine {}'
 
     QEMU_STRAP_MODE_FMT = '-global driver=esp32.gpio,property=strap_mode,value={}'
     QEMU_SERIAL_TCP_FMT = '-serial tcp::{},server,nowait'
@@ -26,6 +29,7 @@ def __init__(
         qemu_prog_path: Optional[str] = None,
         qemu_cli_args: Optional[str] = None,
         qemu_extra_args: Optional[str] = None,
+        app: Optional[QemuApp] = None,
         **kwargs,
     ):
         """
@@ -35,15 +39,40 @@ def __init__(
             qemu_cli_args: QEMU CLI arguments
             qemu_extra_args: QEMU CLI extra arguments, will be appended to `qemu_cli_args`
         """
+        self.app = app
+
         image_path = qemu_image_path or DEFAULT_IMAGE_FN
         if not os.path.exists(image_path):
             raise ValueError(f'QEMU image path doesn\'t exist: {image_path}')
 
-        qemu_prog_path = qemu_prog_path or self.QEMU_PROG_PATH
-        qemu_cli_args = shlex.split(qemu_cli_args or self.QEMU_DEFAULT_ARGS)
+        qemu_prog_path = qemu_prog_path or self.qemu_prog_name
+        qemu_cli_args = shlex.split(qemu_cli_args or self.qemu_default_args)
         qemu_extra_args = shlex.split(qemu_extra_args or '')
 
         super().__init__(
             cmd=[qemu_prog_path, *qemu_cli_args, *qemu_extra_args] + ['-drive', f'file={image_path},if=mtd,format=raw'],
             **kwargs,
         )
+
+    @property
+    def qemu_prog_name(self):
+        if self.app:
+            return self.QEMU_PROG_FMT.format('xtensa' if self.app.is_xtensa else 'riscv32')
+
+        return self.QEMU_PROG_PATH
+
+    @property
+    def qemu_default_args(self):
+        if self.app:
+            return self.QEMU_DEFAULT_FMT.format(self.app.target)
+
+        return self.QEMU_DEFAULT_ARGS
+
+    def _hard_reset(self):
+        """
+        This is a fake hard_reset. Keep this API to keep the consistency.
+        """
+        # TODO: implement with QMP
+        #  https://gitlab.com/qemu-project/python-qemu-qmp/-/issues/6
+        #  for now got so many unexpected exceptions while __del__
+        raise NotImplementedError
diff --git a/pytest-embedded-qemu/tests/test_qemu.py b/pytest-embedded-qemu/tests/test_qemu.py
index aa75fe48..1037d895 100644
--- a/pytest-embedded-qemu/tests/test_qemu.py
+++ b/pytest-embedded-qemu/tests/test_qemu.py
@@ -1,10 +1,13 @@
 import os
+import shutil
+import xml.etree.ElementTree as ET
 
 import pytest
 
 qemu_bin_required = pytest.mark.skipif(
-    os.getenv('DONT_SKIP_QEMU_TESTS', False) is False,
-    reason='Build QEMU for ESP32 locally and then ' 'use "DONT_SKIP_QEMU_TESTS" to run this test.',
+    shutil.which('qemu-system-xtensa') is None,
+    reason='Please make sure that `qemu-system-xtensa` is in your PATH env var. Build QEMU for ESP32 locally and then '
+           'run `pytest` again'
 )
 
 
@@ -80,3 +83,29 @@ def test_pexpect_by_qemu(dut):
     )
 
     result.assert_outcomes(passed=1)
+
+@qemu_bin_required
+def test_qemu_use_idf_mixin_methods(testdir):
+    testdir.makepyfile("""
+        import pexpect
+        import pytest
+
+        def test_qemu_use_idf_mixin_methods(dut):
+            dut.run_all_single_board_cases()
+    """)
+
+    result = testdir.runpytest(
+        '-s',
+        '--embedded-services', 'idf,qemu',
+        '--app-path', f'{os.path.join(testdir.tmpdir, "unit_test_app_esp32")}',
+        '--junitxml', 'report.xml',
+    )
+
+    result.assert_outcomes(failed=1)
+
+    junit_report = ET.parse('report.xml').getroot()[0]
+
+    assert junit_report.attrib['errors'] == '0'
+    assert junit_report.attrib['failures'] == '1'
+    assert junit_report.attrib['skipped'] == '0'
+    assert junit_report.attrib['tests'] == '2'
diff --git a/pytest-embedded/pytest_embedded/dut.py b/pytest-embedded/pytest_embedded/dut.py
index 96b40994..3b65e457 100644
--- a/pytest-embedded/pytest_embedded/dut.py
+++ b/pytest-embedded/pytest_embedded/dut.py
@@ -9,10 +9,10 @@
 from .app import App
 from .log import PexpectProcess
 from .unity import UNITY_SUMMARY_LINE_REGEX, TestSuite
-from .utils import Meta, remove_asci_color_code, to_bytes, to_list
+from .utils import Meta, _InjectMixinCls, remove_asci_color_code, to_bytes, to_list
 
 
-class Dut:
+class Dut(_InjectMixinCls):
     """
     Device under test (DUT) base class
 
@@ -160,7 +160,7 @@ def expect_unity_test_output(
             - Would raise TIMEOUT exception at the end of the test if any unity test case execution took longer
                 than timeout value
 
-        Warnings:
+        Warning:
             - All unity test cases record would be missed if the final report block is uncaught.
         """
         self.expect(UNITY_SUMMARY_LINE_REGEX, timeout=timeout)
@@ -174,3 +174,25 @@ def expect_unity_test_output(
             log = remove_asci_color_code(log)
 
         self.testsuite.add_unity_test_cases(log)
+
+    @_InjectMixinCls.require_services('idf')
+    def run_all_single_board_cases(
+        self,
+        group: Optional[str] = None,
+        reset: bool = False,
+        timeout: float = 30,
+        run_ignore_cases: bool = False,
+    ):
+        """
+        Run all multi_stage cases
+
+        Args:
+            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`
+        """
+        pass
diff --git a/pytest-embedded/pytest_embedded/plugin.py b/pytest-embedded/pytest_embedded/plugin.py
index feb91855..6c2db4ac 100644
--- a/pytest-embedded/pytest_embedded/plugin.py
+++ b/pytest-embedded/pytest_embedded/plugin.py
@@ -947,6 +947,7 @@ def _services(embedded_services: t.Optional[str]) -> t.List[str]:
 @dataclass
 class ClassCliOptions:
     classes: t.Dict[str, type]
+    mixins: t.Dict[str, t.List[type]]
     kwargs: t.Dict[str, t.Dict[str, t.Any]]
 
 
@@ -1010,6 +1011,7 @@ 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():
@@ -1142,6 +1144,7 @@ def _fixture_classes_and_options(
                     'qemu_prog_path': qemu_prog_path,
                     'qemu_cli_args': qemu_cli_args,
                     'qemu_extra_args': qemu_extra_args,
+                    'app': None,
                     'meta': _meta,
                 }
         elif fixture == 'dut':
@@ -1163,6 +1166,11 @@ def _fixture_classes_and_options(
                         'qemu': None,
                     }
                 )
+
+                if 'idf' in _services:
+                    from pytest_embedded_idf.unity_tester import IdfUnityDutMixin
+
+                    mixins[fixture].append(IdfUnityDutMixin)
             elif 'jtag' in _services:
                 if 'idf' in _services:
                     from pytest_embedded_idf import IdfDut
@@ -1202,7 +1210,7 @@ def _fixture_classes_and_options(
                     }
                 )
 
-    return ClassCliOptions(classes, kwargs)
+    return ClassCliOptions(classes, mixins, kwargs)
 
 
 ####################
@@ -1301,6 +1309,7 @@ def dut(
         return cls(**kwargs)
 
     cls = _fixture_classes_and_options.classes['dut']
+    mixins = _fixture_classes_and_options.mixins['dut']
     kwargs = _fixture_classes_and_options.kwargs['dut']
 
     for k, v in kwargs.items():
@@ -1316,7 +1325,7 @@ def dut(
             elif k == 'qemu':
                 kwargs[k] = qemu
 
-    return cls(**_drop_none_kwargs(kwargs))
+    return cls(**_drop_none_kwargs(kwargs), mixins=mixins)
 
 
 @pytest.fixture
diff --git a/pytest-embedded/pytest_embedded/utils.py b/pytest-embedded/pytest_embedded/utils.py
index 9480c6d2..e34bd56c 100644
--- a/pytest-embedded/pytest_embedded/utils.py
+++ b/pytest-embedded/pytest_embedded/utils.py
@@ -33,6 +33,12 @@
 
 _T = t.TypeVar('_T')
 
+_MIXIN_REQUIRED_SERVICES = {
+    'IdfUnityMixin': ['idf'],
+}
+
+_MIXIN_REQUIRED_SERVICES_KEY = '_based_on_services'
+
 
 #######################
 # Errors and Warnings #
@@ -54,6 +60,16 @@ def __init__(self, service: str) -> None:
         )
 
 
+class RequireServiceError(SystemExit):
+    def __init__(self, func_name: str, services: t.Union[str, t.List[str]]) -> None:
+        services_str = ','.join(to_list(services))
+        super().__init__(
+            f'function {func_name} requires enabling one of the service(s) {services_str}. '
+            f'Please enable by passing CLI options "--embedded-services {services_str}". '
+            f'For more details, please refer to "pytest --help".'
+        )
+
+
 #####################
 # Utility Functions #
 #####################
@@ -245,3 +261,80 @@ def __getattr__(object_name):
             raise AttributeError('Attribute %s not found in module %s', object_name, base_module.__name__)
 
     return __getattr__
+
+
+class _InjectMixinMeta(type):
+    def __call__(cls, *args, **kwargs):
+        try:
+            mixins = kwargs.pop('mixins', None)
+            if mixins:
+                mixins = to_list(mixins)
+                name = cls.__name__ + 'With' + 'And'.join([m.__name__ for m in mixins])
+                cls = type(name, tuple([*mixins, cls]), {})
+
+                # users should only know concept "services", not mixins
+                _based_on_services = set()
+                for m in mixins:
+                    if m.__name__ not in _MIXIN_REQUIRED_SERVICES:
+                        continue
+
+                    _based_on_services.update(to_list(_MIXIN_REQUIRED_SERVICES[m.__name__]))
+                kwargs[_MIXIN_REQUIRED_SERVICES_KEY] = sorted(_based_on_services)
+        except KeyError:
+            pass
+        return type.__call__(cls, *args, **kwargs)
+
+
+class _InjectMixinCls(metaclass=_InjectMixinMeta):
+    """
+    This class provide a check function `require_services()` to check if the function is injected by
+    enabling one of the required services.
+
+    The benefits are:
+    - provide the autocompletion for the functions provided by the mixins
+    - check the requirement at runtime
+
+    Examples:
+
+        class IdfUnityMixin:
+            def foo(self):
+                print('foo from MixinOne')
+
+
+        class Test(_InjectMixinCls):
+            def __init__(self, **kwargs):
+                for k, v in kwargs.items():
+                    setattr(self, k, v)
+
+            @_InjectMixinCls.require_services('idf')
+            def foo(self):
+                pass
+
+
+        # mro(): TestWithIdfUnityMixin, IdfUnityMixin, Test, object
+        s1 = Test(mixins=IdfUnityMixin)
+        # mro(): Test, object
+        s2 = Test()
+        s1.foo()  # foo from IdfUnityMixin
+        s2.foo()  # function foo requires enabling one of the service(s) idf.
+    """
+
+    def require_services(*services):
+        def decorator(func):
+            def wrapped(self, *args, **kwargs):
+                based = False
+                for service in services:
+                    if hasattr(self, _MIXIN_REQUIRED_SERVICES_KEY) and service in getattr(
+                        self, _MIXIN_REQUIRED_SERVICES_KEY
+                    ):
+                        based = True
+                        break
+
+                if based:
+                    return func(self, *args, **kwargs)
+                else:
+                    raise RequireServiceError(func.__name__, services)
+
+            return wrapped
+
+        return decorator