feat: support 'skip_decode_panic', split panic decode output and coredump output

Author: hfudev

pytest-embedded-idf/pytest_embedded_idf/app.py

@@ -35,6 +35,8 @@ class IdfApp(App):
         'esp32p4',
         'esp32h2',
         'esp32c61',
+        'esp32h21',
+        'esp32h4',
     ]
 
     FLASH_ARGS_FILENAME = 'flash_args'

pytest-embedded-idf/pytest_embedded_idf/dut.py

@@ -26,6 +26,7 @@ class IdfDut(IdfUnityDutMixin, SerialDut):
     Attributes:
         target (str): target chip type
         skip_check_coredump (bool): skip check core dumped or not while dut teardown if set to True
+        skip_decode_panic (bool): skip decode panic output or not while dut teardown if set to True
     """
 
     XTENSA_TARGETS = IdfApp.XTENSA_TARGETS
@@ -46,11 +47,12 @@ def __init__(
         self,
         app: IdfApp,
         skip_check_coredump: bool = False,
+        skip_decode_panic: bool = False,
         **kwargs,
     ) -> None:
         self.target = app.target
         self.skip_check_coredump = skip_check_coredump
-
+        self.skip_decode_panic = skip_decode_panic
         super().__init__(app=app, **kwargs)
 
         self._hard_reset_func = self.serial.hard_reset
@@ -76,7 +78,11 @@ def _get_prefix_map_path(self) -> str:
             return primary
         return fallback
 
-    def _check_panic_decode_trigger(self):  # type: () -> None
+    def _decode_panic(self):  # type: () -> None
+        if self.target not in self.RISCV32_TARGETS:
+            logging.debug('panic decode only supported for riscv32 targets')
+            return
+
         if not self.app.elf_file:
             logging.warning('No elf file found. Skipping decode panic output...')
             return
@@ -109,9 +115,11 @@ def _check_panic_decode_trigger(self):  # type: () -> None
             'bt',
         ]
         try:
-            output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
-            logging.info('\n\nBacktrace:\n')
-            logging.info(output.decode())
+            output = subprocess.check_output(cmd, stderr=subprocess.STDOUT).decode('utf-8')
+            logging.info(f'Backtrace:\n{output}')
+            with open(self.logfile.replace('dut', 'panic_decoded', 1), 'w') as fw:
+                fw.write(output)
+                logging.info(f'Please check decoded panic output file at: {fw.name}')
         except subprocess.CalledProcessError as e:
             logging.error(f'Failed to decode panic output: {e.output}. Command was: \n{" ".join(cmd)}')
         finally:
@@ -137,8 +145,6 @@ def _check_coredump(self) -> None:
         Returns:
             None
         """
-        if self.target in self.RISCV32_TARGETS:
-            self._check_panic_decode_trigger()  # need IDF_PATH
         if self.app.sdkconfig.get('ESP_COREDUMP_ENABLE_TO_UART', False):
             self._dump_b64_coredumps()
         elif self.app.sdkconfig.get('ESP_COREDUMP_ENABLE_TO_FLASH', False):
@@ -166,12 +172,12 @@ def _dump_b64_coredumps(self) -> None:
                     coredump = CoreDump(
                         chip=self.target,
                         core=coredump_file.name,
-                        core_format='b64',
                         prog=self.app.elf_file,
                     )
-                    with open(os.path.join(self._meta.logdir, f'coredump_output_{i}'), 'w') as fw:
+                    with open(self.logfile.replace('dut', 'coredump', 1), 'w') as fw:
                         with redirect_stdout(fw):
                             coredump.info_corefile()
+                        logging.info(f'Please check coredump output file at: {fw.name}')
                 except Exception as e:
                     logging.error(f'Error dumping b64 coredump {i} for target: {self.target}: {e}')
                 finally:
@@ -185,25 +191,25 @@ def _dump_flash_coredump(self) -> None:
 
         from esp_coredump import CoreDump  # need IDF_PATH
 
-        if self.app.sdkconfig['ESP_COREDUMP_DATA_FORMAT_ELF']:
-            core_format = 'elf'
-        elif self.app.sdkconfig['ESP_COREDUMP_DATA_FORMAT_BIN']:
-            core_format = 'raw'
-        else:
-            raise ValueError('Invalid coredump format. Use _parse_b64_coredump for UART')
-
-        with self.serial.disable_redirect_thread():
+        self.serial.close()
+        with redirect_stdout(self._q):
             coredump = CoreDump(
                 chip=self.target,
-                core_format=core_format,
                 port=self.serial.port,
                 prog=self.app.elf_file,
             )
-            with open(os.path.join(self._meta.logdir, 'coredump_output'), 'w') as fw:
+            with open(self.logfile.replace('dut', 'coredump', 1), 'w') as fw:
                 with redirect_stdout(fw):
                     coredump.info_corefile()
+                logging.info(f'Please check coredump output file at: {fw.name}')
 
     def close(self) -> None:
+        if not self.skip_decode_panic:
+            try:
+                self._decode_panic()
+            except Exception as e:
+                logging.error(f'Error decoding panic output for target: {self.target}: {e}')
+
         if not self.skip_check_coredump:
             try:
                 self._check_coredump()

pytest-embedded-idf/tests/test_idf.py

@@ -562,13 +562,46 @@ def test_unity_tester_with_linux(dut):
 
 
 @toolchain_required
-def test_check_coredump(testdir, caplog, first_index_of_messages):
+@pytest.mark.parametrize(
+    'coredump_type',
+    ['flash', 'uart'],
+)
+def test_check_coredump(testdir, coredump_type, caplog, first_index_of_messages):
+    testdir.makepyfile(rf"""
+        import pexpect
+        import pytest
+        def test_check_coredump_{coredump_type}(dut):
+            dut.expect(pexpect.TIMEOUT, timeout=3)
+    """)
+
+    result = testdir.runpytest(
+        '-s',
+        '--embedded-services',
+        'esp,idf',
+        '--app-path',
+        f'{os.path.join(testdir.tmpdir, f"hello_world_esp32_coredump_{coredump_type}")}',
+        '--target',
+        'esp32',
+        '--log-cli-level',
+        'INFO',
+        '-p',
+        'no:idf-ci',
+    )
+    first_index_of_messages(
+        re.compile('Please check coredump output file at:.+', re.MULTILINE),
+        caplog.messages,
+    )
+    result.assert_outcomes(passed=1)
+
+
+@toolchain_required
+def test_check_panic(testdir, caplog, first_index_of_messages):
     testdir.makepyfile(r"""
         import pexpect
         import pytest
 
-        def test_check_coredump(dut):
-            dut.expect(pexpect.TIMEOUT, timeout=10)
+        def test_check_panic(dut):
+            dut.expect(pexpect.TIMEOUT, timeout=3)
     """)
 
     result = testdir.runpytest(
@@ -581,9 +614,15 @@ def test_check_coredump(dut):
         'esp32c3',
         '--log-cli-level',
         'INFO',
+        '-p',
+        'no:idf-ci',
+    )
+    first_index_of_messages(
+        re.compile(r'Backtrace:\napp_main \(\) at /COMPONENT_MAIN_DIR/hello_world_main.c:17', re.MULTILINE),
+        caplog.messages,
     )
     first_index_of_messages(
-        re.compile(r'app_main \(\) at /COMPONENT_MAIN_DIR/hello_world_main.c:17', re.MULTILINE),
+        re.compile(r'Please check decoded panic output file at:.+', re.MULTILINE),
         caplog.messages,
     )
 

pytest-embedded-serial/pytest_embedded_serial/serial.py

@@ -144,18 +144,22 @@ def close(self):
             logging.debug(f'released {port}')
 
     @contextlib.contextmanager
-    def disable_redirect_thread(self) -> bool:
+    def disable_redirect_thread(self, kill_port: bool = False) -> bool:
         """
         kill the redirect thread, and start a new one after got yield back
 
         Yields:
             True if redirect serial thread has been terminated
         """
         killed = self.stop_redirect_thread()
+        if kill_port:
+            self.proc.close()
 
         yield killed
 
         if killed:
+            if kill_port:
+                self._start()
             self.start_redirect_thread()
 
 

pytest-embedded/pytest_embedded/dut_factory.py

@@ -128,6 +128,7 @@ def _fixture_classes_and_options_fn(
     confirm_target_elf_sha256,
     erase_nvs,
     skip_check_coredump,
+    skip_decode_panic,
     openocd_prog_path,
     openocd_cli_args,
     gdb_prog_path,
@@ -403,6 +404,7 @@ def _fixture_classes_and_options_fn(
                     kwargs[fixture].update(
                         {
                             'skip_check_coredump': skip_check_coredump,
+                            'skip_decode_panic': skip_decode_panic,
                         }
                     )
                 elif 'esp' in _services and 'nuttx' in _services:
@@ -665,6 +667,7 @@ def create(
         confirm_target_elf_sha256: bool | None = None,
         erase_nvs: bool | None = None,
         skip_check_coredump: bool | None = None,
+        skip_decode_panic: bool | None = None,
         openocd_prog_path: str | None = None,
         openocd_cli_args: str | None = None,
         gdb_prog_path: str | None = None,
@@ -711,6 +714,7 @@ def create(
             confirm_target_elf_sha256: Confirm target ELF SHA256.
             erase_nvs: Erase NVS flag.
             skip_check_coredump: Skip coredump check flag.
+            skip_decode_panic: TODO
             openocd_prog_path: OpenOCD program path.
             openocd_cli_args: OpenOCD CLI arguments.
             gdb_prog_path: GDB program path.
@@ -778,6 +782,7 @@ def create(
                 'confirm_target_elf_sha256': confirm_target_elf_sha256,
                 'erase_nvs': erase_nvs,
                 'skip_check_coredump': skip_check_coredump,
+                'skip_decode_panic': skip_decode_panic,
                 'openocd_prog_path': openocd_prog_path,
                 'openocd_cli_args': openocd_cli_args,
                 'gdb_prog_path': gdb_prog_path,

pytest-embedded/pytest_embedded/plugin.py

@@ -940,6 +940,13 @@ def skip_check_coredump(request: FixtureRequest) -> bool | None:
     return _request_param_or_config_option_or_default(request, 'skip_check_coredump', None)
 
 
+@pytest.fixture
+@multi_dut_argument
+def skip_decode_panic(request: FixtureRequest) -> bool | None:
+    """Enable parametrization for the same cli option"""
+    return _request_param_or_config_option_or_default(request, 'skip_decode_panic', None)
+
+
 ########
 # jtag #
 ########
@@ -1092,6 +1099,7 @@ def parametrize_fixtures(
     confirm_target_elf_sha256,
     erase_nvs,
     skip_check_coredump,
+    skip_decode_panic,
     openocd_prog_path,
     openocd_cli_args,
     gdb_prog_path,