You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nuttx.apache.org by xi...@apache.org on 2026/08/04 14:34:10 UTC
(nuttx-ntfc) branch main updated: add advanced debugging tools
This is an automated email from the ASF dual-hosted git repository.
xiaoxiang781216 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nuttx-ntfc.git
The following commit(s) were added to refs/heads/main by this push:
new 7344b43 add advanced debugging tools
7344b43 is described below
commit 7344b43edea10448ca2a4d9b6c3e645c2b026724
Author: p-szafonimateusz <p-...@xiaomi.com>
AuthorDate: Mon Jul 6 19:32:30 2026 +0200
add advanced debugging tools
Add advanced debug features: coredump collection and GDB integration.
Coredump backends: GDB, fastboot, ymodem, local file, syslog. GDB
supports target-remote and simulator PID-attach, plus auto-planted
crash breakpoints for automatic coredump generation.
Signed-off-by: p-szafonimateusz <p-...@xiaomi.com>
Co-authored-by: wangmingrong1 <wa...@xiaomi.com>
Co-authored-by: v-zhanhaixiang <v-...@xiaomi.com>
Co-authored-by: nijie3 <ni...@xiaomi.com>
Co-authored-by: chenjiahua1 <ch...@xiaomi.com>
Co-authored-by: yangao1 <ya...@xiaomi.com>
Co-authored-by: v-zouweiwei <v-...@xiaomi.com>
Co-authored-by: v-yangjing26 <v-...@xiaomi.com>
Co-authored-by: v-wuguohui1 <v-...@xiaomi.com>
---
Documentation/config-yaml.rst | 3 +
Documentation/config.yaml | 35 ++
Documentation/debug/coredump.rst | 93 ++++
Documentation/debug/gdb.rst | 135 +++++
Documentation/debug/index.rst | 77 +++
Documentation/index.rst | 1 +
pyproject.toml | 1 +
src/ntfc/debug/config.py | 413 ++++++++++++++
src/ntfc/debug/coredump/base.py | 96 ++++
src/ntfc/debug/coredump/fastboot_handler.py | 94 ++++
src/ntfc/debug/coredump/gdb_handler.py | 149 +++++
src/ntfc/debug/coredump/local_file_handler.py | 102 ++++
src/ntfc/debug/coredump/manager.py | 129 +++++
src/ntfc/debug/coredump/syslog_handler.py | 210 ++++++++
src/ntfc/debug/coredump/ymodem_handler.py | 157 ++++++
src/ntfc/debug/gdb/controller.py | 456 ++++++++++++++++
src/ntfc/debug/gdb/plugin/autobp.py | 509 ++++++++++++++++++
src/ntfc/debug/gdb/plugin/gdbprefix.py | 55 ++
src/ntfc/debug/gdb/plugin/init.py | 57 ++
src/ntfc/debug/getdebug.py | 282 ++++++++++
src/ntfc/device/common.py | 35 ++
src/ntfc/device/host.py | 5 +
src/ntfc/lib/fastboot/controller.py | 124 +++++
src/ntfc/lib/ymodem/controller.py | 107 ++++
src/ntfc/productconfig.py | 9 +
src/ntfc/pytest/configure.py | 26 +-
src/ntfc/pytest/coredump_plugin.py | 137 +++++
src/ntfc/pytest/mypytest.py | 10 +-
tests/conftest.py | 2 +
tests/debug/conftest.py | 164 ++++++
tests/debug/test_config.py | 331 ++++++++++++
tests/debug/test_coredump_handlers.py | 628 ++++++++++++++++++++++
tests/debug/test_coredump_manager.py | 263 +++++++++
tests/debug/test_gdb_controller.py | 746 ++++++++++++++++++++++++++
tests/debug/test_gdb_plugin.py | 696 ++++++++++++++++++++++++
tests/debug/test_getdebug.py | 373 +++++++++++++
tests/device/test_common.py | 65 +++
tests/device/test_host.py | 14 +
tests/lib/test_fastboot_controller.py | 184 +++++++
tests/lib/test_ymodem_controller.py | 107 ++++
tests/pytest/test_configure.py | 10 +
tests/pytest/test_coredump_plugin.py | 212 ++++++++
tests/pytest/test_hook_ordering.py | 90 ++++
tests/pytest/test_mypytest.py | 80 +++
tests/test_productconfig.py | 51 ++
tox.ini | 2 +
typings/gdb/__init__.pyi | 62 +++
47 files changed, 7585 insertions(+), 2 deletions(-)
diff --git a/Documentation/config-yaml.rst b/Documentation/config-yaml.rst
index ce8ef9f..5b2ffb8 100644
--- a/Documentation/config-yaml.rst
+++ b/Documentation/config-yaml.rst
@@ -396,6 +396,9 @@ These fields are parsed by :class:`ntfc.productconfig.ProductConfig`.
* - ``ignored_cores``
- (Optional) List of core names to skip when collecting core topology
info via ``get_core_info``. Defaults to ``["dsp"]``
+ * - ``debug``
+ - (Optional) Advanced debug features: coredump collection and GDB
+ integration. See :doc:`debug/index`
Core Configuration Fields
=========================
diff --git a/Documentation/config.yaml b/Documentation/config.yaml
index a5531e6..208aa94 100644
--- a/Documentation/config.yaml
+++ b/Documentation/config.yaml
@@ -24,6 +24,41 @@ product: # many products can be supported in tests (pro
name: "product" # product name
platform: "amp" # Platform type: "amp" (default) or "smp"
ignored_cores: ["dsp"] # (optional) core names skipped in get_core_info. Defaults to ["dsp"]
+
+ debug: # (optional) advanced debug features, see Documentation/debug/index.rst
+ coredump: # coredump collection on test failure
+ enable: false # enable coredump collection (default: false)
+ type: 'auto' # backend: 'auto' (default), 'gdb', 'fastboot',
+ # 'ymodem', 'local_file' or 'syslog'
+ limit: 5 # max coredumps collected per session (default: 5)
+ gdb: # GDB-based debugging
+ enable: false # enable the GDB backend (default: false)
+ force_panic: false # force a panic on the device before coredump (default: false)
+ target: '' # 'target remote' address, eg 'localhost:1234' or '/tmp/gdb.socket'
+ gdb_path: 'gdb' # GDB executable, eg 'gdb-multiarch' for cross targets
+ plugin: false # load the NTFC in-GDB Python plugin (default: false)
+ setup_cmds: [] # raw GDB commands sent after attach, eg
+ # - 'ntfcautobp bp up_assert ntfcautogcore -d $RESULT_DIR gcore'
+ gcore_cmd: 'gcore' # command used to write corefiles
+ nx_plugin: '' # path to NuttX pynuttx gdbinit.py
+ osabi: '' # 'set osabi <value>' sent when non-empty
+ auto_breakpoints: false # plant standard crash/poweroff breakpoints
+ mmleak: false # add ntfcmmleak to the poweroff breakpoint
+ attach: false # PID-attach to sim process (excludes target)
+ use_sudo: false # prefix attach-mode GDB with sudo
+ local_file: # host-filesystem coredump backend
+ core_dir: '' # watched host directory (required to enable)
+ pattern: '*.core' # corefile glob pattern
+ syslog: # serial syslog coredump backend
+ enable: false # decode base64+lzf coredump from output
+ fastboot: # fastboot coredump transport
+ dev_sn: '' # fastboot device serial number (required to enable this backend)
+ mem_addr: '0x40000000' # memory dump start address
+ mem_size: '0x08000000' # memory dump size
+ ymodem: # ymodem coredump transport
+ serial_port: '' # host serial port, eg '/dev/ttyUSB0' (required to enable)
+ baud_rate: 921600 # serial baud rate for the transfer
+ sbrb_path: '' # path to the sbrb.py ymodem receiver script (required to enable)
# amp: Each core has its own device instance
# smp: All cores share the same device, automatic core switching
cores: # Cores configuration
diff --git a/Documentation/debug/coredump.rst b/Documentation/debug/coredump.rst
new file mode 100644
index 0000000..c415fa9
--- /dev/null
+++ b/Documentation/debug/coredump.rst
@@ -0,0 +1,93 @@
+===================
+Coredump Collection
+===================
+
+What it does
+============
+
+A coredump is a snapshot of the device's memory and registers at the
+moment it crashed. It's what you load into a debugger afterwards to
+see exactly what the code was doing - which function, which
+variables, which thread. Without one, a crash in an automated test run
+just leaves you with "it crashed" and a log.
+
+Enable with ``debug.coredump.enable: true``. When a test fails, ntfc
+picks a device that looks unhealthy (crashed, busy-looping, or dead)
+and asks a *backend* to pull a coredump from it. If nothing looks
+unhealthy - e.g. a plain assertion in the test itself - it collects
+from every product that has coredump collection enabled.
+
+Backends
+========
+
+There are five ways to actually get the coredump off the device.
+Which one you use depends on what your hardware/setup supports:
+
+* **gdb** - ntfc keeps a GDB session attached to the device for the
+ whole test run and tells it to ``gcore`` on failure. Works for QEMU
+ and any real hardware with a debug probe (JTAG/SWD) or a running GDB
+ stub. See :doc:`gdb` for how this is set up.
+
+* **local_file** - for setups where the coredump ends up as a plain
+ file on the host (e.g. QEMU writing straight to the host
+ filesystem), ntfc just watches a directory and grabs the newest one.
+
+* **fastboot** - pulls a memory region over USB via the Android
+ fastboot protocol, for devices whose bootloader exposes it after a
+ crash.
+
+* **syslog** - decodes a coredump that NuttX itself printed to the
+ serial console as base64 text (optionally LZF-compressed). Useful
+ when there's no debug probe and no fastboot, only a serial port.
+
+* **ymodem** - receives the coredump as a file transferred over the
+ serial port using the Ymodem protocol, once NuttX has written it to
+ local storage.
+
+============== ======== ==========================================
+Backend Priority Requirements
+============== ======== ==========================================
+``gdb`` 10 ``gdb.enable`` + ``elf_path`` + ``target``/``attach``
+``local_file`` 15 ``local_file.core_dir``
+``fastboot`` 20 ``fastboot.dev_sn``
+``syslog`` 25 ``syslog.enable``
+``ymodem`` 30 ``ymodem.serial_port`` + ``sbrb_path``
+============== ======== ==========================================
+
+With ``type: auto`` (default) the enabled backend with the lowest
+priority value that reports itself available is used - so if you
+configure both GDB and fastboot, GDB is tried first. Set ``type`` to a
+specific backend name to force just that one.
+
+``limit`` caps the number of coredumps collected in one session
+(default: 5) so a device that keeps crashing doesn't fill the disk.
+
+Coredump files are written to::
+
+ <result_dir>/<product-name>/<test-name>.core
+
+If two failing tests share the same terminal name, the later ones get
+a numeric suffix (``<test-name>.2.core``, ...) instead of overwriting
+the first.
+
+.. note::
+
+ On multi-core (AMP) products, coredump collection only looks at
+ core 0 today.
+
+.. _post-mortem-analysis:
+
+Inspecting a coredump
+=====================
+
+Once you have a ``.core`` file, load it with the same ELF the session
+used::
+
+ gdb-multiarch <elf_path> result/<session>/<product>/<test>.core
+
+ (gdb) bt full
+ (gdb) info threads
+ (gdb) print g_my_variable
+
+This is the normal way to actually debug the crash - there's no time
+pressure and it doesn't disturb a running test session.
diff --git a/Documentation/debug/gdb.rst b/Documentation/debug/gdb.rst
new file mode 100644
index 0000000..c567b13
--- /dev/null
+++ b/Documentation/debug/gdb.rst
@@ -0,0 +1,135 @@
+===============
+GDB Integration
+===============
+
+What it does
+============
+
+ntfc can keep a GDB session attached to your device for the whole test
+run, so that when a test fails it can pull a coredump on the spot
+(the ``gdb`` backend in :doc:`coredump`), and optionally react to
+crashes automatically instead of waiting for a test to notice.
+
+Enable with ``debug.gdb.enable: true``. ntfc starts one GDB process per
+product after the device is up::
+
+ <gdb_path> -q <elf_path> -ex "set pagination off" -ex "set confirm off"
+
+and connects to it one of two ways:
+
+* **Remote target** (``debug.gdb.target``) - for QEMU or real hardware
+ with a debug probe. Set it to the ``host:port`` (or Unix socket path)
+ of a GDB stub. For QEMU, add ``-s`` to ``exec_args`` to get a stub on
+ ``localhost:1234``.
+
+* **Simulator PID-attach** (``debug.gdb.attach: true``) - for the
+ host-based NuttX simulator, where there's no remote stub, just a
+ running process. ntfc attaches GDB to it by PID instead. Mutually
+ exclusive with ``target``.
+
+ .. note::
+
+ Same-user attach requires ``kernel.yama.ptrace_scope=0`` (or
+ ``CAP_SYS_PTRACE``). If your system has it locked down, either
+ relax it or set ``gdb.use_sudo: true`` (needs passwordless sudo).
+
+ .. note::
+
+ On ``sim`` a crash is a real OS ``SIGSEGV``, not a NuttX assert.
+ Without GDB already attached it kills the process outright (no
+ coredump; ntfc reports ``notalive``, not ``crash``). ``attach``
+ must be enabled *before* the crash: ``ptrace`` lets an attached
+ GDB catch the signal and freeze the process for ``gcore`` instead.
+
+GDB stays attached for the whole session; a plain ``gcore`` writes the
+coredump when a test fails.
+
+Automatic coredumps on crash (push model)
+=========================================
+
+Normally ntfc only asks GDB for a coredump *after* a test fails -
+by then the crash may already be a while in the past. Setting
+``debug.gdb.plugin: true`` and ``debug.gdb.auto_breakpoints: true``
+flips this around: ntfc plants breakpoints on NuttX's crash and
+power-off handlers right after attaching, so GDB generates the
+coredump itself the instant the device crashes, before ntfc even
+notices. When the failing test runs its own collection, it just picks
+up that coredump instead of asking GDB to generate a fresh one.
+
+.. code-block:: yaml
+
+ debug:
+ gdb:
+ enable: true
+ target: "localhost:1234"
+ plugin: true
+ auto_breakpoints: true
+
+Set ``debug.gdb.mmleak: true`` (requires ``auto_breakpoints``) to also
+dump NuttX's memory-leak report on power-off.
+
+The breakpoint's job is to freeze the process for ``gcore``, not to
+give you a live session. Inspect the resulting coredump with your own
+GDB afterward (:ref:`post-mortem-analysis` in :doc:`coredump`) for the
+same backtrace, variables, and threads -- minus the ability to step
+forward past the crash.
+
+Other GDB options
+=================
+
+* ``gdb.gcore_cmd`` (default ``'gcore'``) - override the corefile
+ command, for images that need something different.
+* ``gdb.osabi`` - sends ``set osabi <value>`` after attach; ARM NuttX
+ targets typically need ``'none'``.
+* ``gdb.nx_plugin`` - path to NuttX's own GDB helper
+ (``tools/pynuttx/gdbinit.py`` in the NuttX source tree), sourced
+ alongside ntfc's plugin.
+* ``gdb.setup_cmds`` - raw GDB commands sent right after attach, for
+ anything not covered above:
+
+ .. code-block:: yaml
+
+ setup_cmds:
+ - 'ntfcautobp bp dump_assert_info ntfcautogcore -d $RESULT_DIR gcore'
+ - 'continue'
+
+ Commands can be plain GDB (e.g. ``print my_var``) too, but a plain
+ breakpoint does **not** auto-resume - the target stays halted, which
+ will make the test time out. Use ``ntfcautobp`` (below) for anything
+ meant to run unattended.
+
+The in-GDB plugin
+=================
+
+With ``debug.gdb.plugin: true``, ntfc sources a small Python plugin
+into the GDB process that provides:
+
+* ``ntfcautobp <bp|wp> <spec> <cmd1>[;<cmd2>...]`` - set a breakpoint
+ or watchpoint that runs the given commands when hit, then resumes
+ automatically. This is what makes unattended crash handling possible.
+* ``ntfcautogcore -d <dir|$RESULT_DIR> [-n <name>] <gcore-cmd>`` -
+ generate and verify a coredump.
+* ``ntfcmmleak`` - dump NuttX's ``mm leak`` report alongside a
+ coredump.
+* ``ntfcgdbprefix <command>`` - run a command and quit GDB if it
+ fails, so a bad ``setup_cmds`` entry fails fast instead of hanging.
+
+To add your own commands, drop a Python file into
+``src/ntfc/debug/gdb/plugin/`` - everything there is imported when
+GDB starts.
+
+Manual debugging
+================
+
+ntfc owns the GDB process's stdio, and a stub/PID-attach only allows
+one client. Disable ``debug.gdb`` and connect your own::
+
+ gdb <elf_path> -ex "target remote localhost:1234" # QEMU
+ gdb <elf_path> -ex "target remote /path/to/socket" # real hardware
+
+Works for stepping through normal execution. **Does not work for
+catching a crash**: on ``sim`` the process is already dead by the time
+you attach (nothing caught the signal); on QEMU/hardware the board
+resets within seconds of a panic, wiping the crashed state first.
+Use ``auto_breakpoints`` to catch a crash instead, then inspect the
+coredump it produces (:ref:`post-mortem-analysis`).
diff --git a/Documentation/debug/index.rst b/Documentation/debug/index.rst
new file mode 100644
index 0000000..01e8657
--- /dev/null
+++ b/Documentation/debug/index.rst
@@ -0,0 +1,77 @@
+=======================
+Advanced Debug Features
+=======================
+
+.. toctree::
+ :maxdepth: 1
+
+ coredump
+ gdb
+
+Two optional features, both disabled by default and enabled per
+product via the ``debug`` section of the product configuration:
+
+* :doc:`coredump` - when a test fails, grab a snapshot of the
+ device's memory so you can see what it was doing when it crashed.
+* :doc:`gdb` - keep GDB attached to the device throughout the test
+ run, so a coredump can be pulled on demand, or generated
+ automatically the instant the device crashes.
+
+Device reboot-on-crash (a product going crashed, busy-looping, or
+dead is rebooted with retry and back-off) is core ntfc behavior, not
+a debug feature -- see the top-level ``config.recovery`` section in
+:doc:`../config-yaml`.
+
+Configuration Reference
+=======================
+
+.. code-block:: yaml
+
+ product:
+ name: "example"
+ debug:
+ coredump:
+ enable: false # enable coredump collection on test failure
+ type: 'auto' # 'auto', 'gdb', 'fastboot', 'ymodem',
+ # 'local_file' or 'syslog'
+ limit: 5 # max coredumps collected per session
+ gdb:
+ enable: false # enable the GDB coredump backend
+ force_panic: false # force a panic before coredump collection
+ target: '' # 'target remote' address,
+ # e.g. 'localhost:1234' or '/tmp/gdb.socket'
+ gdb_path: 'gdb' # GDB executable, e.g. 'gdb-multiarch'
+ plugin: false # load the NTFC in-GDB Python plugin
+ setup_cmds: [] # raw GDB commands sent after attach
+ gcore_cmd: 'gcore' # corefile command; NuttX images with
+ # the pynuttx plugin may override this
+ nx_plugin: '' # path to nuttx/tools/pynuttx/gdbinit.py
+ osabi: '' # NuttX arm targets typically need 'none'
+ auto_breakpoints: false # plant crash/poweroff breakpoints + continue
+ mmleak: false # mmleak dump on poweroff breakpoint
+ attach: false # PID-attach to the simulator process
+ use_sudo: false # sudo for attach (locked-down ptrace)
+ local_file:
+ core_dir: '' # host dir watched for corefiles
+ pattern: '*.core'
+ syslog:
+ enable: false # decode coredump from serial syslog
+ fastboot:
+ dev_sn: '' # fastboot device serial number
+ mem_addr: '0x40000000' # memory dump start address
+ mem_size: '0x08000000' # memory dump size
+ ymodem:
+ serial_port: '' # host serial port, e.g. '/dev/ttyUSB0'
+ baud_rate: 921600 # baud rate for the transfer
+ sbrb_path: '' # path to the sbrb.py Ymodem receiver
+
+Notes:
+
+* A coredump backend is only registered when its prerequisites are
+ configured: GDB needs ``gdb.enable``, a core ``elf_path`` and either
+ ``gdb.target`` (remote attach) or ``gdb.attach`` (simulator PID
+ attach); ``local_file`` needs ``local_file.core_dir``; fastboot needs
+ ``fastboot.dev_sn``; ``syslog`` needs ``syslog.enable``; Ymodem needs
+ both ``ymodem.serial_port`` and ``ymodem.sbrb_path``.
+
+See :doc:`coredump` and :doc:`gdb` for details on each feature.
diff --git a/Documentation/index.rst b/Documentation/index.rst
index 5383f43..9e79f5b 100644
--- a/Documentation/index.rst
+++ b/Documentation/index.rst
@@ -15,6 +15,7 @@ for NuttX.
config-yaml
multi-session
session-json
+ debug/index
report
api
diff --git a/pyproject.toml b/pyproject.toml
index 58ceba7..f4fe325 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -34,6 +34,7 @@ dependencies = [
"pexpect",
"psutil",
"pyserial",
+ "lzf",
"pytest",
"pytest-dependency",
"pytest-html",
diff --git a/src/ntfc/debug/config.py b/src/ntfc/debug/config.py
new file mode 100644
index 0000000..af50ccb
--- /dev/null
+++ b/src/ntfc/debug/config.py
@@ -0,0 +1,413 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+"""Debug configuration handlers."""
+
+from typing import Any, Dict, List
+
+###############################################################################
+# Class: CoredumpConfig
+###############################################################################
+
+
+class CoredumpConfig:
+ """Coredump debug configuration.
+
+ Parses the ``debug.coredump`` section of a product YAML configuration.
+ All fields are optional and default to safe/disabled values.
+ """
+
+ VALID_TYPES = ("auto", "gdb", "fastboot", "ymodem", "local_file", "syslog")
+ _DEFAULT_LIMIT = 5
+
+ def __init__(self, cfg: Dict[str, Any]) -> None:
+ """Initialize coredump configuration.
+
+ :param cfg: Raw ``debug.coredump`` dictionary from product YAML.
+ :raises ValueError: If ``type`` is not one of the valid values.
+ """
+ self._enable: bool = bool(cfg.get("enable", False))
+ self._limit: int = int(cfg.get("limit", self._DEFAULT_LIMIT))
+
+ raw_type: str = str(cfg.get("type", "auto"))
+ if raw_type not in self.VALID_TYPES:
+ raise ValueError(
+ f"Invalid coredump type '{raw_type}'. "
+ f"Must be one of: {', '.join(self.VALID_TYPES)}"
+ )
+ self._type: str = raw_type
+
+ @property
+ def enable(self) -> bool:
+ """Return whether coredump collection is enabled."""
+ return self._enable
+
+ @property
+ def collection_type(self) -> str:
+ """Return the coredump collection method.
+
+ :return: One of ``auto``, ``gdb``, ``fastboot``, ``ymodem``.
+ """
+ return self._type
+
+ @property
+ def limit(self) -> int:
+ """Return the maximum number of coredumps to collect per session."""
+ return self._limit
+
+
+###############################################################################
+# Class: GdbConfig
+###############################################################################
+
+
+class GdbConfig:
+ """GDB debug configuration.
+
+ Parses the ``debug.gdb`` section of a product YAML configuration.
+ """
+
+ _DEFAULT_GDB_PATH = "gdb"
+
+ def __init__(self, cfg: Dict[str, Any]) -> None:
+ """Initialize GDB configuration.
+
+ :param cfg: Raw ``debug.gdb`` dictionary from product YAML.
+ """
+ self._enable: bool = bool(cfg.get("enable", False))
+ self._force_panic: bool = bool(cfg.get("force_panic", False))
+ self._target: str = str(cfg.get("target", ""))
+ self._gdb_path: str = str(cfg.get("gdb_path", self._DEFAULT_GDB_PATH))
+ self._plugin: bool = bool(cfg.get("plugin", False))
+ self._setup_cmds: List[str] = [
+ str(cmd) for cmd in cfg.get("setup_cmds", [])
+ ]
+ self._gcore_cmd: str = str(cfg.get("gcore_cmd", "gcore"))
+ self._nx_plugin: str = str(cfg.get("nx_plugin", ""))
+ self._osabi: str = str(cfg.get("osabi", ""))
+ self._auto_breakpoints: bool = bool(cfg.get("auto_breakpoints", False))
+ self._mmleak: bool = bool(cfg.get("mmleak", False))
+ self._attach: bool = bool(cfg.get("attach", False))
+ self._use_sudo: bool = bool(cfg.get("use_sudo", False))
+
+ if self._auto_breakpoints and not self._plugin:
+ raise ValueError("gdb.auto_breakpoints requires gdb.plugin")
+ if self._mmleak and not self._auto_breakpoints:
+ raise ValueError("gdb.mmleak requires gdb.auto_breakpoints")
+ if self._attach and self._target:
+ raise ValueError(
+ "gdb.attach and gdb.target are mutually exclusive"
+ )
+
+ @property
+ def enable(self) -> bool:
+ """Return whether GDB integration is enabled."""
+ return self._enable
+
+ @property
+ def force_panic(self) -> bool:
+ """Return whether to force a panic on the device before coredump."""
+ return self._force_panic
+
+ @property
+ def target(self) -> str:
+ """Return the GDB remote target address or socket path.
+
+ The value is passed directly to ``target remote <target>`` in GDB.
+ Examples: ``"localhost:1234"``, ``"/tmp/gdb.socket"``.
+ An empty string means no ``target remote`` command is sent.
+
+ :return: Target string (may be empty).
+ """
+ return self._target
+
+ @property
+ def gdb_path(self) -> str:
+ """Return the GDB executable used to debug the target.
+
+ Cross-compiled targets typically require ``gdb-multiarch`` or a
+ toolchain-specific GDB (e.g. ``arm-none-eabi-gdb``).
+
+ :return: GDB executable name or path.
+ """
+ return self._gdb_path
+
+ @property
+ def plugin(self) -> bool:
+ """Return whether to load the NTFC in-GDB Python plugin."""
+ return self._plugin
+
+ @property
+ def setup_cmds(self) -> List[str]:
+ """Return raw GDB commands sent after the debugger is attached.
+
+ :return: List of GDB command strings (may be empty).
+ """
+ return list(self._setup_cmds)
+
+ @property
+ def gcore_cmd(self) -> str:
+ """Return the command used to write corefiles."""
+ return self._gcore_cmd
+
+ @property
+ def nx_plugin(self) -> str:
+ """Return the path to NuttX pynuttx gdbinit.py."""
+ return self._nx_plugin
+
+ @property
+ def osabi(self) -> str:
+ """Return the OSABI value sent via ``set osabi`` (empty to skip)."""
+ return self._osabi
+
+ @property
+ def auto_breakpoints(self) -> bool:
+ """Return whether to plant standard crash/poweroff breakpoints."""
+ return self._auto_breakpoints
+
+ @property
+ def mmleak(self) -> bool:
+ """Return whether to add ntfcmmleak to the poweroff breakpoint."""
+ return self._mmleak
+
+ @property
+ def attach(self) -> bool:
+ """Return whether to PID-attach to simulator process."""
+ return self._attach
+
+ @property
+ def use_sudo(self) -> bool:
+ """Return whether to prefix attach-mode GDB with sudo."""
+ return self._use_sudo
+
+
+###############################################################################
+# Class: FastbootConfig
+###############################################################################
+
+
+class FastbootConfig:
+ """Fastboot coredump configuration.
+
+ Parses the ``debug.fastboot`` section of a product YAML configuration.
+ """
+
+ _DEFAULT_MEM_ADDR = "0x40000000"
+ _DEFAULT_MEM_SIZE = "0x08000000"
+
+ def __init__(self, cfg: Dict[str, Any]) -> None:
+ """Initialize fastboot configuration.
+
+ :param cfg: Raw ``debug.fastboot`` dictionary from product YAML.
+ """
+ self._dev_sn: str = str(cfg.get("dev_sn", ""))
+ self._mem_addr: str = str(cfg.get("mem_addr", self._DEFAULT_MEM_ADDR))
+ self._mem_size: str = str(cfg.get("mem_size", self._DEFAULT_MEM_SIZE))
+
+ @property
+ def dev_sn(self) -> str:
+ """Return the fastboot device serial number."""
+ return self._dev_sn
+
+ @property
+ def mem_addr(self) -> str:
+ """Return the memory start address for the dump (hex string)."""
+ return self._mem_addr
+
+ @property
+ def mem_size(self) -> str:
+ """Return the memory region size to dump (hex string)."""
+ return self._mem_size
+
+
+###############################################################################
+# Class: YmodemConfig
+###############################################################################
+
+
+class YmodemConfig:
+ """Ymodem coredump configuration.
+
+ Parses the ``debug.ymodem`` section of a product YAML configuration.
+ """
+
+ _DEFAULT_BAUD_RATE = 921600
+
+ def __init__(self, cfg: Dict[str, Any]) -> None:
+ """Initialize Ymodem configuration.
+
+ :param cfg: Raw ``debug.ymodem`` dictionary from product YAML.
+ """
+ self._serial_port: str = str(cfg.get("serial_port", ""))
+ self._baud_rate: int = int(
+ cfg.get("baud_rate", self._DEFAULT_BAUD_RATE)
+ )
+ self._sbrb_path: str = str(cfg.get("sbrb_path", ""))
+
+ @property
+ def serial_port(self) -> str:
+ """Return the host serial port path (e.g. ``/dev/ttyUSB0``)."""
+ return self._serial_port
+
+ @property
+ def baud_rate(self) -> int:
+ """Return the baud rate used for the Ymodem transfer."""
+ return self._baud_rate
+
+ @property
+ def sbrb_path(self) -> str:
+ """Return the path to the ``sbrb.py`` Ymodem receiver script."""
+ return self._sbrb_path
+
+
+###############################################################################
+# Class: LocalFileConfig
+###############################################################################
+
+
+class LocalFileConfig:
+ """Local-file coredump configuration.
+
+ Parses the ``debug.local_file`` section of a product YAML
+ configuration.
+ """
+
+ _DEFAULT_PATTERN = "*.core"
+
+ def __init__(self, cfg: Dict[str, Any]) -> None:
+ """Initialize local-file configuration.
+
+ :param cfg: Raw ``debug.local_file`` dictionary from product YAML.
+ """
+ self._core_dir: str = str(cfg.get("core_dir", ""))
+ self._pattern: str = str(cfg.get("pattern", self._DEFAULT_PATTERN))
+
+ @property
+ def core_dir(self) -> str:
+ """Return the host directory watched for corefiles."""
+ return self._core_dir
+
+ @property
+ def pattern(self) -> str:
+ """Return the corefile glob pattern."""
+ return self._pattern
+
+
+###############################################################################
+# Class: SyslogConfig
+###############################################################################
+
+
+class SyslogConfig:
+ """Syslog coredump configuration.
+
+ Parses the ``debug.syslog`` section of a product YAML configuration.
+ """
+
+ def __init__(self, cfg: Dict[str, Any]) -> None:
+ """Initialize syslog configuration.
+
+ :param cfg: Raw ``debug.syslog`` dictionary from product YAML.
+ """
+ self._enable: bool = bool(cfg.get("enable", False))
+
+ @property
+ def enable(self) -> bool:
+ """Return whether syslog coredump decoding is enabled."""
+ return self._enable
+
+
+###############################################################################
+# Class: DebugConfig
+###############################################################################
+
+
+class DebugConfig:
+ """Top-level debug configuration.
+
+ Parses the ``debug`` section of a product YAML configuration::
+
+ product:
+ debug:
+ coredump:
+ enable: true
+ type: auto
+ limit: 5
+ gdb:
+ enable: true
+ force_panic: true
+ fastboot:
+ dev_sn: "ABC123"
+ ymodem:
+ serial_port: "/dev/ttyUSB0"
+ sbrb_path: "/path/to/sbrb.py"
+ local_file:
+ core_dir: "/cores"
+ pattern: "*.core"
+ syslog:
+ enable: false
+
+ When the ``debug`` section is absent the configuration defaults to all
+ features disabled.
+ """
+
+ def __init__(self, cfg: Dict[str, Any]) -> None:
+ """Initialize debug configuration.
+
+ :param cfg: Raw ``debug`` dictionary from product YAML. May be
+ empty when the section is not present in the configuration file.
+ :raises ValueError: If a nested section contains an invalid value.
+ """
+ self._coredump = CoredumpConfig(cfg.get("coredump", {}))
+ self._gdb = GdbConfig(cfg.get("gdb", {}))
+ self._fastboot = FastbootConfig(cfg.get("fastboot", {}))
+ self._ymodem = YmodemConfig(cfg.get("ymodem", {}))
+ self._local_file = LocalFileConfig(cfg.get("local_file", {}))
+ self._syslog = SyslogConfig(cfg.get("syslog", {}))
+
+ @property
+ def coredump(self) -> CoredumpConfig:
+ """Return coredump configuration."""
+ return self._coredump
+
+ @property
+ def gdb(self) -> GdbConfig:
+ """Return GDB configuration."""
+ return self._gdb
+
+ @property
+ def fastboot(self) -> FastbootConfig:
+ """Return fastboot configuration."""
+ return self._fastboot
+
+ @property
+ def ymodem(self) -> YmodemConfig:
+ """Return Ymodem configuration."""
+ return self._ymodem
+
+ @property
+ def local_file(self) -> LocalFileConfig:
+ """Return local-file coredump configuration."""
+ return self._local_file
+
+ @property
+ def syslog(self) -> SyslogConfig:
+ """Return syslog coredump configuration."""
+ return self._syslog
diff --git a/src/ntfc/debug/coredump/base.py b/src/ntfc/debug/coredump/base.py
new file mode 100644
index 0000000..928e666
--- /dev/null
+++ b/src/ntfc/debug/coredump/base.py
@@ -0,0 +1,96 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+"""Abstract base class for coredump handlers."""
+
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+ from pathlib import Path
+
+###############################################################################
+# Class: CoredumpHandler
+###############################################################################
+
+
+class CoredumpHandler(ABC):
+ """Abstract base class for coredump collection handlers.
+
+ Concrete implementations provide a specific collection method such as
+ GDB, fastboot, or Y-Modem. Handlers can be toggled at runtime via
+ :meth:`enable` / :meth:`disable`.
+ """
+
+ def __init__(self) -> None:
+ """Initialize :class:`CoredumpHandler` with enabled state."""
+ self._enabled: bool = True
+
+ @property
+ @abstractmethod
+ def name(self) -> str:
+ """Return the handler identifier string.
+
+ :return: A short, unique name (e.g. ``"gdb"``, ``"fastboot"``).
+ """
+
+ @property
+ @abstractmethod
+ def priority(self) -> int:
+ """Return the handler selection priority.
+
+ Lower values are preferred when multiple handlers are available.
+
+ :return: Non-negative integer priority.
+ """
+
+ def is_available(self) -> bool:
+ """Return ``True`` if the handler can currently collect a coredump.
+
+ Default ``True``; override for handlers with a real runtime
+ precondition (e.g. a connected device or running process).
+
+ :return: Availability flag.
+ """
+ return True
+
+ @abstractmethod
+ def collect(self, output_dir: "Path", prefix: str) -> bool:
+ """Collect a coredump and write it under *output_dir*.
+
+ :param output_dir: Directory where the coredump file is written.
+ :param prefix: Filename prefix for the output file.
+ :return: ``True`` on success, ``False`` on failure.
+ """
+
+ def enable(self) -> None:
+ """Enable this handler for coredump collection."""
+ self._enabled = True
+
+ def disable(self) -> None:
+ """Disable this handler so it is skipped during collection."""
+ self._enabled = False
+
+ def is_enabled(self) -> bool:
+ """Return ``True`` if this handler is currently enabled.
+
+ :return: Enabled flag.
+ """
+ return self._enabled
diff --git a/src/ntfc/debug/coredump/fastboot_handler.py b/src/ntfc/debug/coredump/fastboot_handler.py
new file mode 100644
index 0000000..0e4493e
--- /dev/null
+++ b/src/ntfc/debug/coredump/fastboot_handler.py
@@ -0,0 +1,94 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+"""Fastboot-based coredump handler."""
+
+from typing import TYPE_CHECKING
+
+from ntfc.debug.coredump.base import CoredumpHandler
+
+if TYPE_CHECKING:
+ from pathlib import Path
+
+ from ntfc.lib.fastboot.controller import FastbootController
+
+
+class FastbootHandler(CoredumpHandler):
+ """Coredump handler that collects via fastboot memory dump.
+
+ Delegates to
+ :class:`~ntfc.lib.fastboot.controller.FastbootController` which
+ runs ``fastboot oem memdump`` then ``fastboot get_staged``.
+
+ :param controller: Fastboot controller managing the download.
+ :param mem_addr: Memory start address for the dump (hex string).
+ :param mem_size: Memory region size to dump (hex string).
+ """
+
+ def __init__(
+ self,
+ controller: "FastbootController",
+ mem_addr: str,
+ mem_size: str,
+ ) -> None:
+ """Initialize :class:`FastbootHandler`.
+
+ :param controller: Fastboot controller managing the download.
+ :param mem_addr: Memory start address (hex string).
+ :param mem_size: Memory region size (hex string).
+ """
+ super().__init__()
+ self._controller = controller
+ self._mem_addr = mem_addr
+ self._mem_size = mem_size
+
+ @property
+ def name(self) -> str:
+ """Return handler name ``"fastboot"``."""
+ return "fastboot"
+
+ @property
+ def priority(self) -> int:
+ """Return selection priority ``20``."""
+ return 20
+
+ def is_available(self) -> bool:
+ """Return ``True`` if the device is visible in fastboot mode.
+
+ :return: Delegates to
+ :meth:`~ntfc.lib.fastboot.controller.FastbootController\
+.is_connected`.
+ """
+ return self._controller.is_connected()
+
+ def collect(self, output_dir: "Path", prefix: str) -> bool:
+ """Download a memory dump via fastboot.
+
+ :param output_dir: Directory where the dump file is written.
+ :param prefix: Filename stem for the output file (saved as
+ ``<prefix>.bin``).
+ :return: ``True`` on success, ``False`` on failure.
+ """
+ result = self._controller.memdump(
+ self._mem_addr,
+ self._mem_size,
+ output_dir / f"{prefix}.bin",
+ )
+ return result is not None
diff --git a/src/ntfc/debug/coredump/gdb_handler.py b/src/ntfc/debug/coredump/gdb_handler.py
new file mode 100644
index 0000000..c21e06e
--- /dev/null
+++ b/src/ntfc/debug/coredump/gdb_handler.py
@@ -0,0 +1,149 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+"""GDB-based coredump handler."""
+
+import shutil
+from typing import TYPE_CHECKING, Callable, Optional
+
+from ntfc.debug.coredump.base import CoredumpHandler
+from ntfc.log.logger import logger
+
+if TYPE_CHECKING:
+ from pathlib import Path
+
+ from ntfc.debug.gdb.controller import GdbController
+
+
+###############################################################################
+# Class: GdbHandler
+###############################################################################
+
+
+class GdbHandler(CoredumpHandler):
+ """Coredump handler that uses a running :class:`GdbController`.
+
+ :param controller: GDB controller instance used to generate coredumps.
+ :param force_panic: Optional callable that panics the device before
+ the coredump is generated (``debug.gdb.force_panic``), so the
+ dump captures crash state instead of a live system.
+ :param crash_check: Optional callable returning ``True`` when the
+ product has already crashed, in which case an in-GDB automatic
+ coredump is expected instead of a pulled one.
+ :param auto_dump: Whether automatic crash breakpoints are active in
+ the GDB session, so a crash coredump is written by GDB itself
+ and only needs to be harvested.
+ """
+
+ #: Seconds to wait for an in-GDB crash coredump to appear.
+ AUTO_DUMP_TIMEOUT = 600.0
+
+ def __init__(
+ self,
+ controller: "GdbController",
+ force_panic: "Optional[Callable[[], bool]]" = None,
+ crash_check: "Optional[Callable[[], bool]]" = None,
+ auto_dump: bool = False,
+ ) -> None:
+ """Initialize :class:`GdbHandler`.
+
+ :param controller: GDB controller used for coredump generation.
+ :param force_panic: Optional callable forcing a device panic
+ before collection.
+ :param crash_check: Optional callable reporting whether the
+ product has already crashed.
+ :param auto_dump: Whether automatic crash coredumps are active.
+ """
+ super().__init__()
+ self._controller = controller
+ self._force_panic = force_panic
+ self._crash_check = crash_check
+ self._auto_dump = auto_dump
+
+ @property
+ def name(self) -> str:
+ """Return the handler name ``"gdb"``.
+
+ :return: ``"gdb"``
+ """
+ return "gdb"
+
+ @property
+ def priority(self) -> int:
+ """Return the handler priority ``10``.
+
+ :return: ``10``
+ """
+ return 10
+
+ def is_available(self) -> bool:
+ """Return ``True`` when the GDB controller process is running.
+
+ :return: Availability flag.
+ """
+ return self._controller.is_running()
+
+ def collect(self, output_dir: "Path", prefix: str) -> bool:
+ """Generate or harvest a coredump via the GDB controller.
+
+ When the product crashed and automatic crash breakpoints are
+ active, the in-GDB plugin is already writing the dump: wait for
+ it and move it into place. Otherwise fall back to the pull path
+ (optional forced panic, then ``gcore``).
+
+ :param output_dir: Directory where the coredump file is written.
+ :param prefix: Filename prefix for the output file.
+ :return: ``True`` if a corefile was created, ``False`` otherwise.
+ """
+ if (
+ self._auto_dump
+ and self._crash_check is not None
+ and self._crash_check()
+ ):
+ return self._harvest_auto_dump(output_dir, prefix)
+
+ if self._force_panic is not None:
+ logger.debug("gdb: forcing device panic before coredump")
+ if not self._force_panic():
+ logger.warning("gdb: force panic failed")
+
+ result = self._controller.generate_coredump(output_dir, prefix)
+ return result is not None
+
+ def _harvest_auto_dump(self, output_dir: "Path", prefix: str) -> bool:
+ """Wait for the in-GDB auto-generated coredump and move it.
+
+ :param output_dir: Destination directory.
+ :param prefix: Destination filename prefix.
+ :return: ``True`` when a corefile arrived within the window.
+ """
+ logger.debug("gdb: waiting for in-GDB crash coredump")
+ corefile = self._controller.wait_corefile(
+ timeout=self.AUTO_DUMP_TIMEOUT
+ )
+ if corefile is None:
+ logger.warning("gdb: timed out waiting for crash coredump")
+ return False
+
+ output_dir.mkdir(parents=True, exist_ok=True)
+ dest = output_dir / f"{prefix}.core"
+ shutil.move(str(corefile), dest)
+ logger.debug(f"gdb: crash coredump harvested to {dest}")
+ return True
diff --git a/src/ntfc/debug/coredump/local_file_handler.py b/src/ntfc/debug/coredump/local_file_handler.py
new file mode 100644
index 0000000..a05cd7d
--- /dev/null
+++ b/src/ntfc/debug/coredump/local_file_handler.py
@@ -0,0 +1,102 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+"""Local-file coredump handler for host-filesystem corefiles."""
+
+import shutil
+from pathlib import Path
+from typing import TYPE_CHECKING, Optional, Set
+
+from ntfc.debug.coredump.base import CoredumpHandler
+from ntfc.log.logger import logger
+
+if TYPE_CHECKING:
+ from ntfc.debug.config import LocalFileConfig
+
+###############################################################################
+# Class: LocalFileHandler
+###############################################################################
+
+
+class LocalFileHandler(CoredumpHandler):
+ """Collect corefiles written directly to the host filesystem.
+
+ Useful for QEMU or simulator targets configured to write their
+ coredump into a host directory.
+
+ :param cfg: Local-file section of the product debug configuration.
+ """
+
+ def __init__(self, cfg: "LocalFileConfig") -> None:
+ """Initialize :class:`LocalFileHandler`.
+
+ :param cfg: Local-file configuration object.
+ """
+ super().__init__()
+ self._cfg = cfg
+ self._consumed: Set[Path] = set()
+
+ @property
+ def name(self) -> str:
+ """Return the handler name ``"local_file"``."""
+ return "local_file"
+
+ @property
+ def priority(self) -> int:
+ """Return the handler priority ``15``."""
+ return 15
+
+ def is_available(self) -> bool:
+ """Return ``True`` when the watched directory exists."""
+ return Path(self._cfg.core_dir).is_dir()
+
+ def collect(self, output_dir: "Path", prefix: str) -> bool:
+ """Move the newest unconsumed corefile into the result dir.
+
+ :param output_dir: Directory where the coredump file is written.
+ :param prefix: Filename prefix for the output file.
+ :return: ``True`` when a corefile was collected.
+ """
+ candidate = self._newest_corefile()
+ if candidate is None:
+ logger.debug("local_file: no new corefile found")
+ return False
+
+ output_dir.mkdir(parents=True, exist_ok=True)
+ dest = output_dir / f"{prefix}.core"
+ shutil.move(str(candidate), dest)
+ self._consumed.add(candidate)
+ logger.debug(f"local_file: collected {candidate} -> {dest}")
+ return True
+
+ def _newest_corefile(self) -> Optional[Path]:
+ """Return the newest matching, non-empty, unconsumed corefile.
+
+ :return: Corefile path or ``None``.
+ """
+ directory = Path(self._cfg.core_dir)
+ candidates = [
+ p
+ for p in directory.glob(self._cfg.pattern)
+ if p.is_file() and p not in self._consumed and p.stat().st_size > 0
+ ]
+ if not candidates:
+ return None
+ return max(candidates, key=lambda p: p.stat().st_mtime)
diff --git a/src/ntfc/debug/coredump/manager.py b/src/ntfc/debug/coredump/manager.py
new file mode 100644
index 0000000..d475232
--- /dev/null
+++ b/src/ntfc/debug/coredump/manager.py
@@ -0,0 +1,129 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+"""Coredump manager: selects and invokes the best available handler."""
+
+from typing import TYPE_CHECKING, List
+
+from ntfc.log.logger import logger
+
+if TYPE_CHECKING:
+ from pathlib import Path
+
+ from ntfc.debug.config import CoredumpConfig
+ from ntfc.debug.coredump.base import CoredumpHandler
+
+###############################################################################
+# Class: CoredumpManager
+###############################################################################
+
+
+class CoredumpManager:
+ """Orchestrates coredump collection across registered handlers.
+
+ Handlers are selected by priority (lowest value wins). When
+ :attr:`~CoredumpConfig.collection_type` is not ``"auto"`` only the
+ handler whose :attr:`~CoredumpHandler.name` matches that value is
+ considered.
+
+ :param cfg: Coredump section of the product debug configuration.
+ """
+
+ def __init__(self, cfg: "CoredumpConfig") -> None:
+ """Initialize :class:`CoredumpManager`.
+
+ :param cfg: Coredump configuration object.
+ """
+ self._cfg = cfg
+ self._handlers: "List[CoredumpHandler]" = []
+ self._count: int = 0
+
+ def register(self, handler: "CoredumpHandler") -> None:
+ """Register a handler with this manager.
+
+ :param handler: :class:`~ntfc.debug.coredump.base.CoredumpHandler`
+ instance to add.
+ """
+ self._handlers.append(handler)
+
+ def collect(self, output_dir: "Path", prefix: str) -> bool:
+ """Collect a coredump using the best available handler.
+
+ Handlers are tried in priority order until one succeeds, so a
+ transient failure of the preferred backend falls back to the
+ next configured one.
+
+ Returns ``False`` when:
+
+ * coredump collection is disabled in configuration,
+ * the per-session limit has been reached, or
+ * no enabled and available handler succeeds.
+
+ The collection count is incremented only on a successful collection.
+
+ :param output_dir: Directory where the coredump file is written.
+ :param prefix: Filename prefix passed through to the handler.
+ :return: ``True`` on success, ``False`` otherwise.
+ """
+ if not self._cfg.enable:
+ logger.debug("coredump collection disabled")
+ return False
+
+ if self._count >= self._cfg.limit:
+ logger.debug(
+ f"coredump limit reached ({self._count}/{self._cfg.limit})"
+ )
+ return False
+
+ candidates = [
+ h for h in self._handlers if h.is_enabled() and h.is_available()
+ ]
+
+ if self._cfg.collection_type != "auto":
+ candidates = [
+ h for h in candidates if h.name == self._cfg.collection_type
+ ]
+
+ candidates.sort(key=lambda h: h.priority)
+
+ if not candidates:
+ logger.debug("no available coredump handler")
+ return False
+
+ for handler in candidates:
+ logger.debug(f"collecting coredump via handler '{handler.name}'")
+ if handler.collect(output_dir, prefix):
+ self._count += 1
+ return True
+ logger.warning(f"coredump handler '{handler.name}' failed")
+
+ return False
+
+ def reset(self) -> None:
+ """Reset the collection counter between test runs."""
+ self._count = 0
+
+ @property
+ def collected_count(self) -> int:
+ """Return the number of coredumps collected this session.
+
+ :return: Non-negative integer count.
+ """
+ return self._count
diff --git a/src/ntfc/debug/coredump/syslog_handler.py b/src/ntfc/debug/coredump/syslog_handler.py
new file mode 100644
index 0000000..75cd73e
--- /dev/null
+++ b/src/ntfc/debug/coredump/syslog_handler.py
@@ -0,0 +1,210 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+"""Syslog coredump handler for NuttX BOARD_COREDUMP_SYSLOG output."""
+
+import base64
+import io
+import struct
+from typing import TYPE_CHECKING, Optional
+
+import lzf # type: ignore[import-untyped]
+
+from ntfc.debug.coredump.base import CoredumpHandler
+from ntfc.log.logger import logger
+
+if TYPE_CHECKING:
+ from pathlib import Path
+
+ from ntfc.device.common import DeviceCommon
+
+START_MARKER = "Start coredump"
+FINISH_MARKER = "Finish coredump"
+
+
+###############################################################################
+# Class: SyslogHandler
+###############################################################################
+
+
+class SyslogHandler(CoredumpHandler):
+ """Decode a coredump embedded in the device's syslog output.
+
+ NuttX with ``CONFIG_BOARD_COREDUMP_SYSLOG`` emits the coredump as a
+ base64 block between ``Start coredump`` and ``Finish coredump``
+ lines. The decoded payload is a stream of ``ZV`` lzf blocks.
+
+ :param device: Device whose output contains the coredump.
+ """
+
+ def __init__(self, device: "DeviceCommon") -> None:
+ """Initialize :class:`SyslogHandler`.
+
+ :param device: Device instance providing ``output_tail()``.
+ """
+ super().__init__()
+ self._device = device
+
+ @property
+ def name(self) -> str:
+ """Return the handler name ``"syslog"``.
+
+ :return: ``"syslog"``
+ """
+ return "syslog"
+
+ @property
+ def priority(self) -> int:
+ """Return the handler priority ``25``.
+
+ :return: ``25``
+ """
+ return 25
+
+ def collect(self, output_dir: "Path", prefix: str) -> bool:
+ """Extract, decode and write the coredump from device output.
+
+ The output tail is cleared as soon as it is read, regardless of
+ outcome, so a later, unrelated test can never pick up markers
+ left over from this (or an even earlier) collection attempt.
+
+ :param output_dir: Directory where the coredump file is written.
+ :param prefix: Filename prefix for the output file.
+ :return: ``True`` when a coredump was decoded and saved.
+ """
+ output = self._device.output_tail()
+ self._device.reset_output_tail()
+
+ payload = self._extract_payload(output)
+ if payload is None:
+ logger.warning("syslog: no coredump markers in device output")
+ return False
+
+ try:
+ raw = base64.b64decode(payload)
+ except ValueError as exc:
+ logger.warning(f"syslog: base64 decode failed: {exc}")
+ return False
+
+ data = self._decompress(raw)
+ if not data:
+ logger.warning("syslog: empty coredump payload")
+ return False
+
+ output_dir.mkdir(parents=True, exist_ok=True)
+ dest = output_dir / f"{prefix}.core"
+ dest.write_bytes(data)
+ logger.debug(f"syslog: saved coredump {dest} ({len(data)} bytes)")
+ return True
+
+ @staticmethod
+ def _extract_payload(output: str) -> Optional[str]:
+ """Return the base64 payload between the last complete pair.
+
+ Marker pairs are committed atomically: a ``Start coredump``
+ line only takes effect once it is matched by a following
+ ``Finish coredump`` line, and the last such complete pair
+ wins. A trailing, unterminated ``Start coredump`` line (with
+ no matching ``Finish coredump`` after it) is ignored and does
+ not discard a previously recorded valid pair.
+
+ Each payload line carries a syslog prefix; only the text after
+ the last space is coredump data.
+
+ :param output: Recent raw device output.
+ :return: Concatenated base64 text, or ``None``.
+ """
+ lines = output.splitlines()
+ pending: Optional[int] = None
+ start = finish = None
+ for i, line in enumerate(lines):
+ if START_MARKER in line:
+ pending = i
+ elif FINISH_MARKER in line and pending is not None:
+ start, finish = pending, i
+ pending = None
+ if start is None or finish is None:
+ return None
+
+ chars = []
+ for line in lines[start + 1 : finish]:
+ index = line.rfind(" ")
+ if index > 0:
+ line = line[index + 1 :]
+ chars.append(line.strip())
+ return "".join(chars)
+
+ @staticmethod
+ def _decompress(data: bytes) -> bytes:
+ """Decode a stream of ``ZV`` lzf blocks.
+
+ :param data: Base64-decoded payload.
+ :return: Decompressed coredump bytes (empty on failure).
+ """
+ stream = io.BytesIO(data)
+ output = bytearray()
+ while True:
+ if stream.read(2) != b"ZV":
+ break
+ typ = stream.read(1)
+ clen_raw = stream.read(2)
+ if len(clen_raw) < 2:
+ break
+ clen = struct.unpack(">H", clen_raw)[0]
+ if typ == b"\x00":
+ output.extend(stream.read(clen))
+ elif typ == b"\x01":
+ ulen_raw = stream.read(2)
+ if len(ulen_raw) < 2:
+ break
+ ulen = struct.unpack(">H", ulen_raw)[0]
+ block = SyslogHandler._lzf_decompress_block(
+ stream.read(clen), ulen
+ )
+ if block is None:
+ break
+ output.extend(block)
+ else:
+ break
+ return bytes(output)
+
+ @staticmethod
+ def _lzf_decompress_block(compressed: bytes, ulen: int) -> Optional[bytes]:
+ """Decompress a single liblzf-framed block.
+
+ The ``lzf`` package (a CFFI binding) exposes no module-level
+ ``decompress(data, length)`` helper; the C function is called
+ directly via ``lzf.lib.lzf_decompress``, which writes into a
+ pre-allocated output buffer and returns the number of bytes
+ written (``0`` on error), mirroring the package's own
+ ``_iter_decompress`` implementation.
+
+ :param compressed: Compressed block bytes.
+ :param ulen: Expected decompressed length.
+ :return: Decompressed bytes, or ``None`` on failure.
+ """
+ out_buf = lzf.ffi.new("char[]", max(ulen, 1))
+ written = lzf.lib.lzf_decompress(
+ compressed, len(compressed), out_buf, ulen
+ )
+ if written == 0 or written != ulen:
+ logger.warning("syslog: lzf block decompression failed")
+ return None
+ return bytes(lzf.ffi.unpack(out_buf, written))
diff --git a/src/ntfc/debug/coredump/ymodem_handler.py b/src/ntfc/debug/coredump/ymodem_handler.py
new file mode 100644
index 0000000..6e2ad98
--- /dev/null
+++ b/src/ntfc/debug/coredump/ymodem_handler.py
@@ -0,0 +1,157 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+"""Ymodem-based coredump handler."""
+
+import datetime
+import re
+from typing import TYPE_CHECKING, Optional
+
+from ntfc.debug.coredump.base import CoredumpHandler
+from ntfc.device.common import CmdStatus
+
+if TYPE_CHECKING:
+ from pathlib import Path
+
+ from ntfc.device.common import DeviceCommon
+ from ntfc.lib.ymodem.controller import YmodemController
+
+
+class YmodemHandler(CoredumpHandler):
+ """Coredump handler that collects via Ymodem serial protocol.
+
+ Locates the corefile on the device using shell commands, then
+ transfers it using
+ :class:`~ntfc.lib.ymodem.controller.YmodemController`.
+
+ :param controller: Ymodem controller managing the file transfer.
+ :param device: Device instance used to send shell commands.
+ """
+
+ _COREDIR = "/log/offlinelog"
+
+ def __init__(
+ self,
+ controller: "YmodemController",
+ device: "DeviceCommon",
+ ) -> None:
+ """Initialize :class:`YmodemHandler`.
+
+ :param controller: Ymodem controller managing the file transfer.
+ :param device: Device instance used to send shell commands.
+ """
+ super().__init__()
+ self._controller = controller
+ self._device = device
+
+ @property
+ def name(self) -> str:
+ """Return handler name ``"ymodem"``."""
+ return "ymodem"
+
+ @property
+ def priority(self) -> int:
+ """Return selection priority ``30``."""
+ return 30
+
+ def is_available(self) -> bool:
+ """Return ``True`` if the controller is configured and ready.
+
+ :return: Delegates to
+ :meth:`~ntfc.lib.ymodem.controller.YmodemController.is_ready`.
+ """
+ return self._controller.is_ready()
+
+ def collect(self, output_dir: "Path", prefix: str) -> bool:
+ """Locate and download a corefile via Ymodem.
+
+ Calls :meth:`_find_coredump` to rename the file on the device,
+ then
+ :meth:`~ntfc.lib.ymodem.controller.YmodemController.download`
+ to transfer it. The device is stopped for the duration of the
+ transfer so ``sbrb.py`` gets exclusive access to the serial
+ port, and restarted afterwards.
+
+ :param output_dir: Host directory where the file is saved.
+ :param prefix: Filename prefix used when renaming the corefile.
+ :return: ``True`` on success, ``False`` on failure.
+ """
+ corefile_name = self._find_coredump(prefix)
+ if corefile_name is None:
+ return False
+
+ # sbrb.py needs exclusive access to the serial port; the device
+ # must release it during the transfer.
+ self._device.stop()
+ try:
+ return self._controller.download(output_dir, corefile_name)
+ finally:
+ self._device.start()
+
+ def _find_coredump(self, prefix: str) -> Optional[str]:
+ """Locate and rename an unformatted core file on the device.
+
+ Searches :attr:`_COREDIR` for a ``.core`` file that has not yet
+ been timestamped. Renames it to
+ ``<prefix>.<YYYY.MM.DD_HH.MM.SS>.core`` in-place on the device.
+
+ :param prefix: Filename prefix for the renamed corefile.
+ :return: New corefile name on success, ``None`` if not found.
+ """
+ result = self._device.send_cmd_read_until_pattern(
+ b"ls -l " + self._COREDIR.encode() + b"/",
+ rb"\.core",
+ timeout=20,
+ )
+ if result.status != CmdStatus.SUCCESS:
+ return None
+
+ core_file = self._parse_corefile(result.output)
+ if core_file is None:
+ return None
+
+ ts = datetime.datetime.now().strftime("%Y.%m.%d_%H.%M.%S")
+ new_name = f"{prefix}.{ts}.core"
+
+ mv_cmd = (
+ f"mv {self._COREDIR}/{core_file}" f" {self._COREDIR}/{new_name}"
+ ).encode()
+ self._device.send_cmd_read_until_pattern(mv_cmd, b">", timeout=60)
+
+ return new_name
+
+ @staticmethod
+ def _parse_corefile(output: str) -> Optional[str]:
+ """Find the first ``.core`` file without a timestamp in its name.
+
+ A file is considered timestamped when its name contains a segment
+ matching ``YYYY.MM.DD_HH.MM.SS``.
+
+ :param output: Text output from ``ls`` on the device.
+ :return: Core filename, or ``None`` if none found.
+ """
+ stamped = re.compile(r"\d{4}\.\d{2}\.\d{2}_\d{2}\.\d{2}\.\d{2}")
+ for line in output.splitlines():
+ m = re.search(r"(\S+\.core)", line)
+ if m:
+ name = m.group(1)
+ if not stamped.search(name):
+ return name
+ return None
diff --git a/src/ntfc/debug/gdb/controller.py b/src/ntfc/debug/gdb/controller.py
new file mode 100644
index 0000000..e82177d
--- /dev/null
+++ b/src/ntfc/debug/gdb/controller.py
@@ -0,0 +1,456 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+"""GDB controller for automated coredump generation."""
+
+import subprocess
+import threading
+from pathlib import Path
+from typing import TYPE_CHECKING, Callable, List, Optional, Set
+
+from ntfc.log.logger import logger
+
+if TYPE_CHECKING:
+ from ntfc.debug.config import GdbConfig
+
+
+###############################################################################
+# Class: GdbController
+###############################################################################
+
+
+class GdbController:
+ """Controls a GDB subprocess for remote debugging and coredump generation.
+
+ Launches GDB in non-interactive (``-q``) mode, and either attaches
+ to a remote target specified by
+ :attr:`~ntfc.debug.config.GdbConfig.target` or PID-attaches to a
+ local simulator process when
+ :attr:`~ntfc.debug.config.GdbConfig.attach` is set, then exposes
+ :meth:`generate_coredump` to write a core file.
+
+ :param elf_path: Path to the ELF binary loaded by GDB.
+ :param cfg: GDB section of the product debug configuration.
+ :param pid_provider: Callable returning the simulator process PID
+ for attach mode, or ``None`` when unused.
+ """
+
+ # GDB prints no interactive prompt when its stdout is a pipe, so
+ # explicit markers are echoed after key command sequences: without
+ # them there is no way to tell a slow command from a failed one.
+ READY_MARKER = "NTFC_GDB_READY"
+ ATTACH_MARKER = "NTFC_GDB_ATTACH_DONE"
+ GCORE_MARKER = "NTFC_GDB_GCORE_DONE"
+
+ # Standard crash breakpoint specs for auto_breakpoints feature.
+ # dump_mini_info/dump_core_info are vendor-kernel symbols absent on
+ # upstream NuttX; _assert is the public assert.c entry point present
+ # on any build (with or without debug info), so auto_breakpoints
+ # still plants a working crash breakpoint on a stripped image.
+ CRASH_BP_SPECS = (
+ "dump_assert_info",
+ "dump_mini_info",
+ "dump_core_info",
+ "_assert",
+ )
+
+ def __init__(
+ self,
+ elf_path: Path,
+ cfg: "GdbConfig",
+ pid_provider: "Optional[Callable[[], Optional[int]]]" = None,
+ ) -> None:
+ """Initialize :class:`GdbController`.
+
+ :param elf_path: Path to the ELF binary.
+ :param cfg: GDB configuration object.
+ :param pid_provider: Callable returning the simulator process PID
+ for :attr:`~ntfc.debug.config.GdbConfig.attach` mode, or
+ ``None`` when the PID is unavailable.
+ """
+ self._elf_path = elf_path
+ self._cfg = cfg
+ self._pid_provider = pid_provider
+ self._process: Optional[subprocess.Popen[bytes]] = None
+ self._reader: Optional[threading.Thread] = None
+ self._ready = threading.Event()
+ self._attach_done = threading.Event()
+ self._attached = False
+ self._gcore_done = threading.Event()
+ self._last_corefile: Optional[Path] = None
+ self._lock = threading.Lock()
+ self._corefiles: List[Path] = []
+ self._consumed: "Set[Path]" = set()
+ self._corefile_cond = threading.Condition(self._lock)
+
+ def start(self, timeout: float = 30.0) -> bool:
+ """Start GDB and attach to the configured remote target or PID.
+
+ Spawns ``<cfg.gdb_path> -q -nx <elf_path>`` (optionally prefixed
+ with ``sudo`` in attach mode), optionally sourcing the NTFC
+ in-GDB plugin and any NuttX plugin, configuring the OS ABI if
+ specified, waits for the readiness marker, then either sends
+ ``attach <pid>`` (when :attr:`~ntfc.debug.config.GdbConfig.attach`
+ is set) or ``target remote <cfg.target>`` and verifies that GDB
+ reports a successful attach. User configuration files
+ (``.gdbinit``) are skipped for a deterministic environment.
+
+ :param timeout: Seconds to wait for GDB readiness and for the
+ attach to complete before giving up.
+ :return: ``True`` on success, ``False`` on launch failure,
+ timeout, process exit, or failed attach.
+ """
+ cmd = self._build_argv()
+
+ self._ready.clear()
+ self._attach_done.clear()
+ self._attached = False
+ try:
+ self._process = subprocess.Popen(
+ cmd,
+ stdin=subprocess.PIPE,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.STDOUT,
+ )
+ except OSError as exc:
+ logger.warning(
+ f"gdb: failed to launch {self._cfg.gdb_path}: {exc}"
+ )
+ return False
+ self._reader = threading.Thread(
+ target=self._read_stdout, name="GdbReader", daemon=True
+ )
+ self._reader.start()
+
+ if not self._ready.wait(timeout=timeout):
+ logger.warning("gdb: timed out waiting for prompt")
+ self._terminate()
+ return False
+
+ if self._process.poll() is not None:
+ logger.warning("gdb: process exited before prompt")
+ return False
+
+ if self._cfg.attach:
+ pid = self._pid_provider() if self._pid_provider else None
+ if pid is None:
+ logger.warning("gdb: attach requested but no PID available")
+ self._terminate()
+ return False
+ return self._verify_attach(f"attach {pid}", timeout)
+
+ if self._cfg.target:
+ return self._verify_attach(
+ f"target remote {self._cfg.target}", timeout
+ )
+
+ return True
+
+ def _build_argv(self) -> List[str]:
+ """Build the GDB command-line invocation.
+
+ Prefixes ``sudo`` in attach mode when
+ :attr:`~ntfc.debug.config.GdbConfig.use_sudo` is set, then adds
+ the NTFC in-GDB plugin, NuttX plugin, and OS ABI options as
+ configured.
+
+ :return: Full argv list to pass to :class:`subprocess.Popen`.
+ """
+ cmd: List[str] = []
+ if self._cfg.attach and self._cfg.use_sudo:
+ cmd.append("sudo")
+ cmd.extend(
+ [
+ self._cfg.gdb_path,
+ "-q",
+ "-nx",
+ str(self._elf_path),
+ "-ex",
+ "set pagination off",
+ "-ex",
+ "set confirm off",
+ ]
+ )
+
+ if self._cfg.plugin:
+ init = Path(__file__).parent / "plugin" / "init.py"
+ cmd.extend(["-ex", f"source {init}"])
+
+ if self._cfg.nx_plugin:
+ cmd.extend(["-ex", f"source {self._cfg.nx_plugin}"])
+
+ if self._cfg.osabi:
+ cmd.extend(["-ex", f"set osabi {self._cfg.osabi}"])
+
+ cmd.extend(["-ex", f"echo {self.READY_MARKER}\\n"])
+ return cmd
+
+ def _verify_attach(self, command: str, timeout: float) -> bool:
+ """Send an attach command and verify GDB reports success.
+
+ Sends the attach command and waits for the attach marker. On timeout,
+ logs a timeout warning. On failure (marker received but attach
+ unsuccessful), logs a mode-specific warning:
+
+ * For PID attach (``attach <pid>``), suggests checking
+ ``kernel.yama.ptrace_scope`` or setting ``use_sudo``.
+ * For remote attach (``target remote <target>``), logs a generic
+ attach failure message.
+
+ :param command: ``attach <pid>`` or ``target remote <target>``.
+ :param timeout: Seconds to wait for the attach marker.
+ :return: ``True`` when GDB confirmed the attach, ``False`` on
+ timeout or failed attach.
+ """
+ self._send(f"{command}\n")
+ self._send(f"echo {self.ATTACH_MARKER}\\n\n")
+
+ if not self._attach_done.wait(timeout=timeout):
+ logger.warning(f"gdb: timed out waiting for attach ({command})")
+ self._terminate()
+ return False
+
+ if not self._attached:
+ if self._cfg.attach:
+ logger.warning(
+ f"gdb: attach failed ({command}); check "
+ f"kernel.yama.ptrace_scope or set gdb.use_sudo "
+ f"(see Documentation/debug/gdb.rst)"
+ )
+ else:
+ logger.warning(f"gdb: failed to attach ({command})")
+ self._terminate()
+ return False
+
+ return True
+
+ def setup(self, result_dir: str) -> None:
+ """Send post-attach setup commands to GDB.
+
+ When the in-GDB plugin is enabled the coredump output directory
+ is configured via ``ntfcsetoutdir``. All commands from
+ :attr:`~ntfc.debug.config.GdbConfig.setup_cmds` are sent
+ afterwards. If :attr:`~ntfc.debug.config.GdbConfig.auto_breakpoints`
+ is enabled, standard crash breakpoints are planted and
+ ``continue`` is sent to resume execution.
+
+ :param result_dir: Session result directory used for coredumps.
+ """
+ if self._cfg.plugin:
+ self._send(f"ntfcsetoutdir {result_dir}\n")
+
+ for cmd in self._cfg.setup_cmds:
+ self._send(f"{cmd}\n")
+
+ if self._cfg.auto_breakpoints:
+ gcore = self._cfg.gcore_cmd
+ for spec in self.CRASH_BP_SPECS:
+ self._send(
+ f"ntfcautobp bp {spec} ntfcautogcore"
+ f" -d $RESULT_DIR -n crash {gcore};bt\n"
+ )
+ mmleak = (
+ f"ntfcmmleak -d $RESULT_DIR {gcore};"
+ if self._cfg.mmleak
+ else ""
+ )
+ self._send(
+ f"ntfcautobp bp reboot_notifier_call_chain"
+ f" {mmleak}ntfcautopoweroffcheck\n"
+ )
+ self._send("continue\n")
+
+ def stop(self) -> None:
+ """Quit GDB and clean up the subprocess and reader thread."""
+ self._send("quit\n")
+ if self._process is not None:
+ try:
+ self._process.wait(timeout=5.0)
+ except subprocess.TimeoutExpired:
+ self._terminate()
+ if self._reader is not None:
+ self._reader.join(timeout=5.0)
+
+ def generate_coredump(
+ self, output_dir: Path, prefix: str, timeout: float = 300.0
+ ) -> Optional[Path]:
+ """Send a gcore command (via cfg.gcore_cmd) and wait for writing.
+
+ The returned corefile is marked consumed so a later
+ :meth:`wait_corefile` call (e.g. for an unrelated crash-aware
+ harvest) can never hand out this same file.
+
+ :param output_dir: Directory where the corefile is placed.
+ :param prefix: Filename stem for the generated corefile.
+ :param timeout: Seconds to wait for the gcore command to finish.
+ :return: :class:`~pathlib.Path` to the corefile, or ``None`` on
+ failure or timeout.
+ """
+ with self._lock:
+ self._gcore_done.clear()
+ self._last_corefile = None
+
+ corefile = output_dir / f"{prefix}.core"
+ self._send(f"{self._cfg.gcore_cmd} {corefile}\n")
+ self._send(f"echo {self.GCORE_MARKER}\\n\n")
+
+ if not self._gcore_done.wait(timeout=timeout):
+ logger.warning("gdb: timed out waiting for gcore")
+ return None
+
+ with self._lock:
+ if self._last_corefile is None:
+ logger.warning("gdb: gcore failed to produce a corefile")
+ return None
+ self._consumed.add(self._last_corefile)
+ return self._last_corefile
+
+ def is_running(self) -> bool:
+ """Return ``True`` if the GDB process is alive.
+
+ :return: ``True`` when the process exists and polls as running.
+ """
+ return self._process is not None and self._process.poll() is None
+
+ def _next_unconsumed_corefile(self) -> Optional[Path]:
+ """Return the newest recorded corefile not yet handed out.
+
+ Must be called while holding :attr:`_lock` (or ``_corefile_cond``).
+
+ :return: Corefile path, or ``None`` if none are available.
+ """
+ for corefile in reversed(self._corefiles):
+ if corefile not in self._consumed:
+ return corefile
+ return None
+
+ def wait_corefile(self, timeout: float) -> Optional[Path]:
+ """Return the newest not-yet-consumed corefile, waiting if needed.
+
+ A corefile already recorded (but not yet handed out via this
+ method or :meth:`generate_coredump`) at call time is returned
+ immediately -- this synchronizes with coredumps the in-GDB
+ plugin can generate autonomously (``ntfcautogcore`` on a crash
+ breakpoint) *before* the Python side starts waiting for one.
+ Each corefile is handed out at most once.
+
+ :param timeout: Seconds to wait when none is immediately available.
+ :return: Corefile path, or ``None`` on timeout.
+ """
+ with self._corefile_cond:
+ if self._next_unconsumed_corefile() is None:
+ self._corefile_cond.wait_for(
+ lambda: self._next_unconsumed_corefile() is not None,
+ timeout=timeout,
+ )
+ corefile = self._next_unconsumed_corefile()
+ if corefile is not None:
+ self._consumed.add(corefile)
+ return corefile
+
+ @property
+ def saved_corefiles(self) -> "List[Path]":
+ """Return all corefiles recorded this session."""
+ with self._corefile_cond:
+ return list(self._corefiles)
+
+ def _send(self, command: str) -> None:
+ r"""Write *command* to GDB's stdin.
+
+ No-ops silently when the process is not running.
+
+ :param command: Raw GDB command string (should end with ``\n``).
+ """
+ if self._process is None or self._process.stdin is None:
+ return
+ try:
+ self._process.stdin.write(command.encode())
+ self._process.stdin.flush()
+ except OSError as exc:
+ logger.debug(f"gdb: stdin write failed: {exc}")
+
+ def _terminate(self) -> None:
+ """Forcibly terminate the GDB process and reap it."""
+ if self._process is not None: # pragma: no branch
+ self._process.terminate()
+ try:
+ self._process.wait(timeout=5.0)
+ except subprocess.TimeoutExpired:
+ self._process.kill()
+ self._process.wait()
+
+ def _read_stdout(self) -> None:
+ """Background thread: parse GDB stdout and set synchronisation events.
+
+ Recognises:
+
+ * :attr:`READY_MARKER` / ``(gdb)`` / ``Type "help"`` — sets
+ :attr:`_ready`
+ * ``Remote debugging using`` / ``Attaching to program`` —
+ records a successful attach
+ * :attr:`ATTACH_MARKER` — sets :attr:`_attach_done`
+ * ``Saved corefile`` — parses the corefile path
+ * :attr:`GCORE_MARKER` — sets :attr:`_gcore_done` (with no
+ corefile recorded this signals a failed ``gcore``)
+ * EOF — sets all events to unblock any waiters
+ """
+ if (
+ self._process is None or self._process.stdout is None
+ ): # pragma: no cover
+ self._ready.set()
+ self._attach_done.set()
+ self._gcore_done.set()
+ return
+
+ for raw_line in self._process.stdout:
+ line = raw_line.decode(errors="replace").rstrip()
+ logger.debug(f"gdb< {line}")
+
+ if (
+ self.READY_MARKER in line
+ or "(gdb)" in line
+ or 'Type "help"' in line
+ ):
+ self._ready.set()
+
+ if (
+ "Remote debugging using" in line
+ or "Attaching to program" in line
+ ):
+ self._attached = True
+
+ if self.ATTACH_MARKER in line:
+ self._attach_done.set()
+
+ if "Saved corefile" in line:
+ parts = line.split()
+ corefile_path = Path(parts[-1])
+ with self._corefile_cond:
+ self._last_corefile = corefile_path
+ self._corefiles.append(corefile_path)
+ self._corefile_cond.notify_all()
+
+ if self.GCORE_MARKER in line:
+ self._gcore_done.set()
+
+ # EOF — unblock any waiters so they can observe the failure
+ self._ready.set() # pragma: no cover
+ self._attach_done.set() # pragma: no cover
+ self._gcore_done.set() # pragma: no cover
diff --git a/src/ntfc/debug/gdb/plugin/autobp.py b/src/ntfc/debug/gdb/plugin/autobp.py
new file mode 100644
index 0000000..a0cdaf2
--- /dev/null
+++ b/src/ntfc/debug/gdb/plugin/autobp.py
@@ -0,0 +1,509 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+"""Automatic breakpoint handling loaded inside a GDB process.
+
+This module is sourced by GDB (see ``init.py``) and provides custom
+commands for automated coredump generation:
+
+* ``ntfcautobp`` -- set a breakpoint/watchpoint that executes commands,
+* ``ntfcsetoutdir`` -- set the output directory for coredump files,
+* ``ntfcautogcore`` -- generate a timestamped coredump file,
+* ``ntfcmmleak`` -- dump a coredump when NuttX ``mm leak`` reports leaks,
+* ``ntfcautopoweroffcheck`` -- print a backtrace on power-off checks.
+"""
+
+import datetime
+import os
+import re
+from typing import Any, Callable, Dict, List, Optional, Set, Tuple
+
+import gdb
+
+# File used to persist the output directory across GDB invocations.
+NTFC_DIR_FILE = "/tmp/.ntfc_gdb_output_dir"
+
+# Breakpoint specs treated as crash coredump breakpoints: when one of
+# them triggers, all of its siblings are deleted as well. Must stay in
+# sync with GdbController.CRASH_BP_SPECS (controller.py), which plants
+# the breakpoints these specs are matched against.
+CRASH_COREDUMP_SPECS = (
+ "dump_mini_info",
+ "dump_core_info",
+ "dump_assert_info",
+ "_assert",
+)
+
+# Placeholder arguments resolved to the configured output directory.
+DYNAMIC_DIR_ARGS = ("$RESULT_DIR", "__NTFC_DYNAMIC__")
+
+_output_dir: Optional[str] = None
+
+
+def _result_check(method: Callable[..., Any]) -> Callable[..., Any]:
+ """Decorate a command method to log its boolean result.
+
+ :param method: Method returning a truthy value on success.
+ :return: Wrapped method printing ``Success:``/``Failed:`` lines.
+ """
+
+ def wrapper(*args: Any, **kwargs: Any) -> Any:
+ result = method(*args, **kwargs)
+
+ name = type(args[0]).__name__
+ argstr = " ".join(map(str, args[1:-1])) if len(args) > 1 else ""
+
+ if result:
+ print(f"Success: {name} {argstr}")
+ else:
+ print(f"Failed: {name} {argstr}")
+
+ return result
+
+ return wrapper
+
+
+def _output_dir_get() -> Optional[str]:
+ """Return the configured output directory.
+
+ The global variable set by ``ntfcsetoutdir`` takes precedence; the
+ backup file is used as a fallback.
+
+ :return: Output directory path or ``None`` when not configured.
+ """
+ if _output_dir is not None:
+ return _output_dir
+
+ if os.path.exists(NTFC_DIR_FILE):
+ try:
+ with open(NTFC_DIR_FILE, encoding="utf-8") as f:
+ content = f.read().strip()
+ if content:
+ return content
+ except OSError as e:
+ print(f"Could not read backup file: {e}")
+
+ return None
+
+
+def _output_dir_ensure(output_dir: str) -> bool:
+ """Create the output directory when it does not exist.
+
+ :param output_dir: Directory path to verify or create.
+ :return: ``True`` when the directory exists or was created.
+ """
+ if os.path.isdir(output_dir):
+ return True
+
+ try:
+ os.makedirs(output_dir, exist_ok=True)
+ print(f"Created output directory: {output_dir}")
+ except OSError as e:
+ print(f"Failed to create directory {output_dir}: {e}")
+ return False
+
+ return True
+
+
+def _resolve_output_dir(argv: List[str]) -> Optional[str]:
+ """Resolve the ``-d <dir|$RESULT_DIR>`` output directory argument.
+
+ Shared by :class:`NtfcAutoGcore` and :class:`NtfcAutoMmleak`, whose
+ commands both accept the same ``-d`` flag.
+
+ :param argv: Parsed command arguments.
+ :return: Output directory or ``None`` on error.
+ """
+ if len(argv) < 2:
+ print("Error: No arguments provided")
+ return None
+
+ if argv[0] != "-d":
+ print(f"Error: First argument must be '-d', got '{argv[0]}'")
+ return None
+
+ if argv[1] in DYNAMIC_DIR_ARGS:
+ output_dir = _output_dir_get()
+ if not output_dir:
+ print(
+ "Error: Output directory not set. "
+ "Use 'ntfcsetoutdir <dir>' first."
+ )
+ return None
+ return output_dir
+
+ return argv[1]
+
+
+def _coredump_path(output_dir: str, name: str) -> str:
+ """Build a timestamped coredump file path.
+
+ :param output_dir: Directory where the coredump is written.
+ :param name: Coredump file name prefix.
+ :return: Full path of the coredump file.
+ """
+ stamp = datetime.datetime.now().strftime("%Y.%m.%d_%H.%M.%S")
+ return os.path.join(output_dir, f"{name}.{stamp}.core")
+
+
+class AutoBp(gdb.Breakpoint):
+ """Breakpoint that executes a list of commands when hit.
+
+ Crash coredump breakpoints (see :data:`CRASH_COREDUMP_SPECS`) with an
+ ``ntfcautogcore`` command are tracked together: when one triggers all
+ of its siblings are deleted, so a single crash produces one coredump.
+
+ :param spec: Breakpoint location or watchpoint expression.
+ :param commands: GDB commands to execute when the breakpoint is hit.
+ :param bp_type: ``"bp"`` for a breakpoint, ``"wp"`` for a watchpoint.
+ """
+
+ _pending_delete: "Set[AutoBp]" = set()
+ _crash_coredump_breakpoints: "Dict[str, AutoBp]" = {}
+
+ def __init__(
+ self, spec: str, commands: List[str], bp_type: str = "bp"
+ ) -> None:
+ """Initialize :class:`AutoBp`.
+
+ :param spec: Breakpoint location or watchpoint expression.
+ :param commands: GDB commands to execute on hit.
+ :param bp_type: ``"bp"`` (breakpoint) or ``"wp"`` (watchpoint).
+ """
+ if bp_type == "wp":
+ super().__init__(spec, gdb.BP_WATCHPOINT, internal=False)
+ else:
+ super().__init__(spec, internal=False)
+ self._cmds = commands
+ self._spec = spec
+
+ is_crash_coredump = spec in CRASH_COREDUMP_SPECS
+ has_autogcore = any("ntfcautogcore" in cmd for cmd in commands)
+
+ if is_crash_coredump and has_autogcore:
+ AutoBp._crash_coredump_breakpoints[spec] = self
+ print(f"Registered crash coredump breakpoint: {spec}")
+
+ def stop(self) -> bool:
+ """Execute the configured commands and mark cleanup.
+
+ :return: ``True`` to stop the inferior (resumed by the stop
+ event handler).
+ """
+ for cmd in self._cmds:
+ try:
+ print(f"Executing command: {cmd}")
+ gdb.execute(cmd)
+ except gdb.error as e:
+ print(f"Command failed: {cmd}, error: {e}")
+
+ if self._spec in AutoBp._crash_coredump_breakpoints:
+ AutoBp._pending_delete.add(self)
+
+ for name, bp in AutoBp._crash_coredump_breakpoints.items():
+ if name != self._spec and bp is not self:
+ print(
+ f"Crash coredump triggered at {self._spec}. "
+ f"Also deleting sibling breakpoint: {name}"
+ )
+ AutoBp._pending_delete.add(bp)
+
+ AutoBp._crash_coredump_breakpoints.clear()
+ else:
+ AutoBp._pending_delete.add(self)
+
+ return True
+
+
+class NtfcAutoBp(gdb.Command):
+ """Command ``ntfcautobp``: set an automatic breakpoint.
+
+ Usage: ``ntfcautobp <bp|wp> <spec> <cmd1>[;<cmd2>...]``
+ """
+
+ def __init__(self) -> None:
+ """Register the ``ntfcautobp`` command."""
+ super().__init__("ntfcautobp", gdb.COMMAND_USER)
+
+ @_result_check
+ def invoke(self, args: str, from_tty: bool) -> bool:
+ """Create an :class:`AutoBp` from command arguments.
+
+ :param args: Raw command arguments.
+ :param from_tty: ``True`` when invoked from an interactive tty.
+ :return: ``True`` when the breakpoint was created and is valid.
+ """
+ argv = gdb.string_to_argv(args)
+ if len(argv) < 3:
+ return False
+
+ bp_type = argv[0]
+ target = argv[1]
+ command_str = " ".join(argv[2:]).strip()
+ commands = command_str.split(";")
+
+ if bp_type not in ("bp", "wp"):
+ return False
+
+ bp = AutoBp(spec=target, commands=commands, bp_type=bp_type)
+
+ return bp.is_valid()
+
+
+class NtfcSetOutDir(gdb.Command):
+ """Command ``ntfcsetoutdir``: set the coredump output directory."""
+
+ def __init__(self) -> None:
+ """Register the ``ntfcsetoutdir`` command."""
+ super().__init__("ntfcsetoutdir", gdb.COMMAND_USER)
+
+ def invoke(self, args: str, from_tty: bool) -> None:
+ """Store the output directory and persist it to a backup file.
+
+ :param args: Raw command arguments (a single directory path).
+ :param from_tty: ``True`` when invoked from an interactive tty.
+ """
+ global _output_dir
+
+ argv = gdb.string_to_argv(args)
+ if len(argv) != 1:
+ print("Usage: ntfcsetoutdir <directory>")
+ return
+
+ output_dir = argv[0]
+
+ if not _output_dir_ensure(output_dir):
+ return
+
+ _output_dir = output_dir
+
+ try:
+ with open(NTFC_DIR_FILE, "w", encoding="utf-8") as f:
+ f.write(output_dir)
+ f.flush()
+ print(f"Also saved to file: {NTFC_DIR_FILE}")
+ except OSError as e:
+ print(f"Note: Could not write to backup file: {e}")
+
+ print(f"NTFC Output Directory Set: {output_dir}")
+
+
+class NtfcAutoGcore(gdb.Command):
+ """Command ``ntfcautogcore``: generate a timestamped coredump.
+
+ Usage: ``ntfcautogcore -d <dir|$RESULT_DIR> [-n <name>] <gcore-cmd>``
+ """
+
+ def __init__(self) -> None:
+ """Register the ``ntfcautogcore`` command."""
+ super().__init__("ntfcautogcore", gdb.COMMAND_USER)
+
+ def _parse_name(self, argv: List[str]) -> Optional[Tuple[str, int]]:
+ """Parse the optional ``-n <name>`` arguments.
+
+ :param argv: Parsed command arguments.
+ :return: Tuple of (name, gcore command index) or ``None``.
+ """
+ if len(argv) > 2 and argv[2] == "-n":
+ if len(argv) <= 3:
+ print("Error: -n flag requires a name argument")
+ return None
+ return argv[3], 4
+
+ return "", 2
+
+ def _verify_coredump(self, coredump: str) -> bool:
+ """Verify that the generated coredump is a valid ELF file.
+
+ :param coredump: Path of the generated coredump.
+ :return: ``True`` when the file exists and has an ELF header.
+ """
+ if not os.path.isfile(coredump):
+ print(f"Error: '{coredump}' does not exist or is not a file.\n")
+ return False
+
+ try:
+ with open(coredump, "rb") as f:
+ header = f.read(4)
+ except OSError as e:
+ print(f"Error reading '{coredump}': {e}.\n")
+ return False
+
+ if header != b"\x7fELF":
+ print(f"{coredump} is not valid core file: {header!r}")
+ return False
+
+ print(f"{coredump} is valid ELF core file.\n")
+ return True
+
+ @_result_check
+ def invoke(self, args: str, from_tty: bool) -> bool:
+ """Generate a coredump using the configured gcore command.
+
+ :param args: Raw command arguments.
+ :param from_tty: ``True`` when invoked from an interactive tty.
+ :return: ``True`` when a valid coredump was generated.
+ """
+ argv = gdb.string_to_argv(args)
+ print(f"NtfcAutoGcore: Received args: {argv}")
+
+ output_dir = _resolve_output_dir(argv)
+ if output_dir is None:
+ return False
+
+ parsed = self._parse_name(argv)
+ if parsed is None:
+ return False
+ name, index = parsed
+
+ if not _output_dir_ensure(output_dir):
+ return False
+
+ coredump = _coredump_path(output_dir, name)
+ print(f"Generate coredump: {coredump}")
+
+ cmd = " ".join(argv[index:]).strip() + f" {coredump}"
+ print(f"Executing gcore command: {cmd}")
+
+ try:
+ gdb.execute(cmd)
+ print("gcore command executed successfully")
+ except gdb.error as e:
+ print(f"gcore command failed with exception: {e}")
+ return False
+
+ return self._verify_coredump(coredump)
+
+
+class NtfcAutoMmleak(gdb.Command):
+ """Command ``ntfcmmleak``: dump a coredump on NuttX memory leaks.
+
+ Runs the NuttX ``mm leak`` GDB command; when leaked blocks are
+ reported the leak report is written next to a generated coredump.
+
+ Usage: ``ntfcmmleak -d <dir|$RESULT_DIR> <gcore-cmd>``
+ """
+
+ def __init__(self) -> None:
+ """Register the ``ntfcmmleak`` command."""
+ super().__init__("ntfcmmleak", gdb.COMMAND_USER)
+
+ @_result_check
+ def invoke(self, args: str, from_tty: bool) -> bool:
+ """Check for memory leaks and generate a coredump if leaking.
+
+ :param args: Raw command arguments.
+ :param from_tty: ``True`` when invoked from an interactive tty.
+ :return: ``True`` when the check completed.
+ """
+ argv = gdb.string_to_argv(args)
+ output_dir = _resolve_output_dir(argv)
+ if output_dir is None:
+ return False
+
+ coredump = _coredump_path(output_dir, "mmleak")
+ gcore_cmd = " ".join(argv[2:]).strip() + f" {coredump}"
+
+ ret = gdb.execute("mm leak", to_string=True)
+ print(ret)
+
+ match = re.search(r"Leaked (\d+) blks, (\d+) bytes", ret)
+ if match and int(match.group(1)) > 0:
+ result = os.path.join(output_dir, "nuttx.mmleak")
+ with open(result, "w", encoding="utf-8") as f:
+ f.write(ret)
+ print(f"Mmleak -> {gcore_cmd}")
+ print(gdb.execute(gcore_cmd, to_string=True))
+
+ return True
+
+
+class NtfcAutoPoweroffCheck(gdb.Command):
+ """Command ``ntfcautopoweroffcheck``: print a backtrace."""
+
+ def __init__(self) -> None:
+ """Register the ``ntfcautopoweroffcheck`` command."""
+ super().__init__("ntfcautopoweroffcheck", gdb.COMMAND_USER)
+
+ @_result_check
+ def invoke(self, args: str, from_tty: bool) -> bool:
+ """Print the current backtrace.
+
+ :param args: Raw command arguments (unused).
+ :param from_tty: ``True`` when invoked from an interactive tty.
+ :return: Always ``True``.
+ """
+ print(gdb.execute("bt"))
+ return True
+
+
+def gdb_autobp_stop_event(event: Any) -> None:
+ """Handle GDB stop events: delete pending breakpoints and resume.
+
+ Only stops caused exclusively by :class:`AutoBp` breakpoints are
+ resumed automatically; a stop at a user breakpoint (e.g. set via
+ ``setup_cmds``) leaves the inferior stopped for inspection.
+
+ :param event: GDB stop event instance.
+ """
+ if isinstance(event, gdb.BreakpointEvent):
+ bps = getattr(event, "breakpoints", ())
+ if not all(isinstance(bp, AutoBp) for bp in bps):
+ print("NTFC GDB: stop at user breakpoint, not resuming")
+ return
+ if AutoBp._pending_delete:
+ print(
+ f"NtfcAutoBp: Deleting {len(AutoBp._pending_delete)} "
+ f"marked breakpoint(s)."
+ )
+ for bp in list(AutoBp._pending_delete):
+ try:
+ bp.delete()
+ AutoBp._pending_delete.remove(bp)
+ except gdb.error as e:
+ print(f"Failed to delete breakpoint: {e}")
+ gdb.execute("continue")
+ if isinstance(event, gdb.SignalEvent):
+ if event.stop_signal == "SIGINT":
+ print("NTFC GDB SIGINT Event")
+ elif event.stop_signal == "SIGTERM":
+ print("NTFC GDB SIGTERM Event")
+ gdb.execute("q")
+ else:
+ print("NTFC GDB Unexpected Signal Event")
+
+
+def gdb_process_exited_event(event: Any) -> None:
+ """Handle GDB inferior exit events: quit GDB.
+
+ :param event: GDB exited event instance.
+ """
+ print("NTFC GDB Process Exited Event")
+ gdb.execute("q")
+
+
+NtfcAutoBp()
+NtfcSetOutDir()
+NtfcAutoGcore()
+NtfcAutoMmleak()
+NtfcAutoPoweroffCheck()
+
+gdb.events.stop.connect(gdb_autobp_stop_event)
+gdb.events.exited.connect(gdb_process_exited_event)
diff --git a/src/ntfc/debug/gdb/plugin/gdbprefix.py b/src/ntfc/debug/gdb/plugin/gdbprefix.py
new file mode 100644
index 0000000..769be9c
--- /dev/null
+++ b/src/ntfc/debug/gdb/plugin/gdbprefix.py
@@ -0,0 +1,55 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+"""Fail-fast GDB command wrapper loaded inside a GDB process.
+
+Provides the ``ntfcgdbprefix`` command: it executes the wrapped command
+and quits GDB when the command fails, so configuration errors are
+detected immediately instead of leaving a half-configured debugger.
+"""
+
+import gdb
+
+
+class NtfcGdbPrefix(gdb.Command):
+ """Command ``ntfcgdbprefix``: execute a command, quit GDB on error.
+
+ Usage: ``ntfcgdbprefix <gdb-command>``
+ """
+
+ def __init__(self) -> None:
+ """Register the ``ntfcgdbprefix`` command."""
+ super().__init__("ntfcgdbprefix", gdb.COMMAND_USER)
+
+ def invoke(self, args: str, from_tty: bool) -> None:
+ """Execute the wrapped command; quit GDB when it fails.
+
+ :param args: The GDB command to execute.
+ :param from_tty: ``True`` when invoked from an interactive tty.
+ """
+ try:
+ gdb.execute(args)
+ print(f"NTFC GDB cmd run successful: {args}")
+ except gdb.error as e:
+ print(f"NTFC GDB cmd run failed: {args}\n:{e}")
+ gdb.execute("q")
+
+
+NtfcGdbPrefix()
diff --git a/src/ntfc/debug/gdb/plugin/init.py b/src/ntfc/debug/gdb/plugin/init.py
new file mode 100644
index 0000000..87a2594
--- /dev/null
+++ b/src/ntfc/debug/gdb/plugin/init.py
@@ -0,0 +1,57 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+"""NTFC GDB plugin loader.
+
+This script is sourced by GDB (``gdb -ex "source init.py"``); it imports
+all sibling plugin modules so their commands get registered. GDB runs
+sourced scripts with ``__name__ == "__main__"``.
+"""
+
+import importlib
+import os
+import sys
+import traceback
+
+
+def load() -> None:
+ """Import all plugin modules located next to this file."""
+ here = os.path.dirname(os.path.abspath(__file__))
+
+ if here not in sys.path:
+ sys.path.insert(0, here)
+
+ self_name = os.path.basename(__file__)
+ for file in sorted(os.listdir(here)):
+ if not file.endswith(".py") or file == self_name:
+ continue
+
+ module_name = file[:-3]
+ print(f"import {module_name}")
+ try:
+ importlib.import_module(module_name)
+ except Exception:
+ print(traceback.format_exc())
+
+ print("NTFC GDB Plugin loaded Successfully")
+
+
+if __name__ == "__main__":
+ load()
diff --git a/src/ntfc/debug/getdebug.py b/src/ntfc/debug/getdebug.py
new file mode 100644
index 0000000..095b024
--- /dev/null
+++ b/src/ntfc/debug/getdebug.py
@@ -0,0 +1,282 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+"""Factory creating debug plugins from product configuration."""
+
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Dict, List
+
+from ntfc.debug.coredump.fastboot_handler import FastbootHandler
+from ntfc.debug.coredump.gdb_handler import GdbHandler
+from ntfc.debug.coredump.local_file_handler import LocalFileHandler
+from ntfc.debug.coredump.manager import CoredumpManager
+from ntfc.debug.coredump.syslog_handler import SyslogHandler
+from ntfc.debug.coredump.ymodem_handler import YmodemHandler
+from ntfc.debug.gdb.controller import GdbController
+from ntfc.lib.fastboot.controller import FastbootController
+from ntfc.lib.ymodem.controller import YmodemController
+from ntfc.log.logger import logger
+from ntfc.pytest.coredump_plugin import CoredumpPlugin
+
+if TYPE_CHECKING:
+ from ntfc.debug.config import DebugConfig
+ from ntfc.product import Product
+
+###############################################################################
+# Class: DebugSetup
+###############################################################################
+
+
+class DebugSetup:
+ """Owns debug plugins and controller lifecycles for one test session.
+
+ :param plugins: Pytest plugin instances to pass to the test run.
+ :param controllers_by_product: GDB controllers that must be opened
+ before the run and closed afterwards, keyed by product name so
+ a rebooted product's controller(s) can be re-attached.
+ """
+
+ def __init__(
+ self,
+ plugins: List[Any],
+ controllers_by_product: Dict[str, List[GdbController]],
+ ) -> None:
+ """Initialize :class:`DebugSetup`.
+
+ :param plugins: Pytest plugin instances.
+ :param controllers_by_product: GDB controllers managed by this
+ setup, keyed by the product name they belong to.
+ """
+ self._plugins = plugins
+ self._controllers_by_product = controllers_by_product
+ self._started: List[GdbController] = []
+ self._result_dir = ""
+
+ @property
+ def plugins(self) -> List[Any]:
+ """Return the pytest plugin instances for the test run."""
+ return list(self._plugins)
+
+ def _start_one(self, product_name: str, ctrl: GdbController) -> None:
+ """Start and set up a single controller, tracking it if it opens.
+
+ Each product gets its own subdirectory of the session result
+ directory so two products' in-GDB auto-generated coredumps
+ (written directly by GDB, with only 1-second timestamp
+ resolution) can never collide on the same path.
+
+ :param product_name: Name of the product the controller serves.
+ :param ctrl: Controller to start.
+ """
+ if not ctrl.start():
+ logger.warning("debug: failed to start GDB controller")
+ return
+ ctrl.setup(f"{self._result_dir}/{product_name}")
+ self._started.append(ctrl)
+
+ def start(self, result_dir: str) -> None:
+ """Start all GDB controllers and send their setup commands.
+
+ A controller that fails to start is logged and skipped so the
+ test session can continue without GDB support. No-ops when
+ *result_dir* is empty (log collection disabled).
+
+ :param result_dir: Session result directory used for coredumps.
+ """
+ if not result_dir:
+ return
+
+ self._result_dir = result_dir
+ for name, controllers in self._controllers_by_product.items():
+ for ctrl in controllers:
+ self._start_one(name, ctrl)
+
+ def stop(self) -> None:
+ """Stop all controllers that were successfully started."""
+ for ctrl in self._started:
+ ctrl.stop()
+ self._started.clear()
+
+ def restart_controllers(self, product_name: str) -> None:
+ """Re-attach the GDB controller(s) of one rebooted product.
+
+ A device reboot spawns a new process (or, for real hardware,
+ resets the target in place), so any GDB session already
+ attached to it is stale: it keeps polling as "running" while
+ talking to a process/target that is gone, silently breaking
+ coredump collection for the rest of the session.
+
+ :param product_name: Name of the product whose device rebooted.
+ """
+ if not self._result_dir:
+ return
+
+ for ctrl in self._controllers_by_product.get(product_name, []):
+ if ctrl in self._started:
+ ctrl.stop()
+ self._started.remove(ctrl)
+ self._start_one(product_name, ctrl)
+
+ def restart_all_controllers(self) -> None:
+ """Re-attach every product's GDB controller(s).
+
+ Called after ntfc's built-in crash recovery
+ (:meth:`~ntfc.pytest.configure.PytestConfigPlugin._device_reboot`)
+ reboots the session's products, since that reboot restarts
+ every product unconditionally and so invalidates every GDB
+ session attached to them.
+ """
+ for product_name in self._controllers_by_product:
+ self.restart_controllers(product_name)
+
+
+###############################################################################
+# Public functions
+###############################################################################
+
+
+def _gdb_handler_register(
+ mgr: CoredumpManager,
+ product: "Product",
+ controllers: List[GdbController],
+) -> None:
+ """Register a GDB coredump handler when prerequisites are met.
+
+ Requires a core 0 ``elf_path`` plus either a ``gdb.target`` to
+ attach to or ``gdb.attach`` (simulator PID attach); otherwise a
+ warning is logged and no handler is registered.
+
+ :param mgr: Coredump manager of the product.
+ :param product: Product being configured.
+ :param controllers: List collecting created GDB controllers.
+ """
+ cfg = product.conf.debug
+ elf_path = product.conf.cfg_core(0).elf_path
+
+ if not elf_path or not (cfg.gdb.target or cfg.gdb.attach):
+ logger.warning(
+ f"debug: product {product.name} has gdb enabled "
+ f"but coredump collection requires elf_path and "
+ f"either gdb.target or gdb.attach"
+ )
+ return
+
+ device = product.core(0).device
+ ctrl = GdbController(
+ Path(elf_path), cfg.gdb, pid_provider=lambda: device.pid
+ )
+ panic = product.force_panic if cfg.gdb.force_panic else None
+ mgr.register(
+ GdbHandler(
+ ctrl,
+ force_panic=panic,
+ crash_check=lambda: bool(product.crash),
+ auto_dump=cfg.gdb.auto_breakpoints,
+ )
+ )
+ controllers.append(ctrl)
+
+
+def _register_coredump_handlers(
+ mgr: CoredumpManager,
+ product: "Product",
+ cfg: "DebugConfig",
+ controllers: List[GdbController],
+) -> None:
+ """Register the coredump handlers configured for one product.
+
+ :param mgr: Coredump manager of the product.
+ :param product: Product being configured.
+ :param cfg: Debug configuration of the product.
+ :param controllers: List collecting created GDB controllers.
+ """
+ if cfg.gdb.enable:
+ _gdb_handler_register(mgr, product, controllers)
+
+ if cfg.fastboot.dev_sn:
+ fastboot = FastbootController(cfg.fastboot.dev_sn)
+ mgr.register(
+ FastbootHandler(
+ fastboot, cfg.fastboot.mem_addr, cfg.fastboot.mem_size
+ )
+ )
+
+ if cfg.ymodem.serial_port and cfg.ymodem.sbrb_path:
+ ymodem = YmodemController(
+ cfg.ymodem.serial_port,
+ Path(cfg.ymodem.sbrb_path),
+ cfg.ymodem.baud_rate,
+ )
+ mgr.register(YmodemHandler(ymodem, product.core(0).device))
+
+ if cfg.local_file.core_dir:
+ mgr.register(LocalFileHandler(cfg.local_file))
+
+ if cfg.syslog.enable:
+ mgr.register(SyslogHandler(product.core(0).device))
+
+
+def get_debug_setup(products: "List[Product]") -> DebugSetup:
+ """Create debug plugins and controllers from product configuration.
+
+ For every product with ``debug.coredump.enable`` set a
+ :class:`~ntfc.debug.coredump.manager.CoredumpManager` is created and
+ the handlers configured in the product ``debug`` section (GDB,
+ fastboot, Ymodem, local file, syslog) are registered. A single
+ :class:`~ntfc.pytest.coredump_plugin.CoredumpPlugin` serves all
+ managers.
+
+ Rebooting an unhealthy device between tests is core ntfc behavior
+ (:meth:`~ntfc.pytest.configure.PytestConfigPlugin._device_reboot`),
+ not part of this debug feature set.
+
+ :param products: Products created for the test session.
+ :return: :class:`DebugSetup` with the plugins and controllers.
+ """
+ managers: Dict[str, CoredumpManager] = {}
+ controllers_by_product: Dict[str, List[GdbController]] = {}
+ plugins: List[Any] = []
+
+ for product in products:
+ cfg = product.conf.debug
+
+ if not cfg.coredump.enable:
+ continue
+
+ if product.conf.cores_num > 1:
+ logger.warning(
+ f"debug: product {product.name} has multiple cores; "
+ f"coredump collection only supports core 0"
+ )
+
+ mgr = CoredumpManager(cfg.coredump)
+ controllers: List[GdbController] = []
+ _register_coredump_handlers(mgr, product, cfg, controllers)
+ controllers_by_product[product.name] = controllers
+
+ managers[product.name] = mgr
+ logger.info(f"debug: coredump collection enabled for {product.name}")
+
+ debug_setup = DebugSetup(plugins, controllers_by_product)
+
+ if managers:
+ plugins.append(CoredumpPlugin(managers, products))
+
+ return debug_setup
diff --git a/src/ntfc/device/common.py b/src/ntfc/device/common.py
index c0a5f8a..1c714b4 100644
--- a/src/ntfc/device/common.py
+++ b/src/ntfc/device/common.py
@@ -51,6 +51,7 @@ class DeviceCommon(ABC):
"""Device common interface."""
_BUSY_LOOP_TIMEOUT = 180 # 180 sec with no data read from target
+ _OUTPUT_TAIL_MAX = 4 * 1024 * 1024
def __init__(self, conf: "CoreConfig", echo: bool = True):
"""Initialize common device."""
@@ -74,6 +75,7 @@ class DeviceCommon(ABC):
self._read_all_sleep = 0.1
self._has_echo = echo
self._start_time: Optional[float] = None
+ self._output_tail_buf = bytearray()
def _mark_started(self) -> None:
"""Mark device start time for runtime tracking."""
@@ -85,6 +87,38 @@ class DeviceCommon(ABC):
return None
return time.monotonic() - self._start_time
+ @property
+ def pid(self) -> Optional[int]:
+ """Return the device process PID, or ``None`` when not host-based."""
+ return None
+
+ def _output_tail_append(self, data: bytes) -> None:
+ """Append raw output to the capped tail buffer.
+
+ :param data: Raw bytes read from the device.
+ """
+ self._output_tail_buf.extend(data)
+ if len(self._output_tail_buf) > self._OUTPUT_TAIL_MAX:
+ del self._output_tail_buf[: -self._OUTPUT_TAIL_MAX]
+
+ def output_tail(self) -> str:
+ """Return the most recent raw device output.
+
+ :return: Up to ``_OUTPUT_TAIL_MAX`` bytes of output, decoded.
+ """
+ return self._output_tail_buf.decode(errors="replace")
+
+ def reset_output_tail(self) -> None:
+ """Discard the buffered output tail.
+
+ Called by consumers (e.g.
+ :class:`~ntfc.debug.coredump.syslog_handler.SyslogHandler`)
+ once they have read :meth:`output_tail`, so markers from an
+ earlier, unrelated crash cannot be picked up again by a later
+ test that merely reuses the same handler.
+ """
+ self._output_tail_buf.clear()
+
def _log_device_line(self, line: str) -> None:
"""Log or buffer a pre-formatted device log line."""
if self._logs is None:
@@ -150,6 +184,7 @@ class DeviceCommon(ABC):
while True:
chunk = self._read()
output += chunk
+ self._output_tail_append(chunk)
time_now = time.time()
# check for any sign of system crash
diff --git a/src/ntfc/device/host.py b/src/ntfc/device/host.py
index 3c447cf..d92c442 100644
--- a/src/ntfc/device/host.py
+++ b/src/ntfc/device/host.py
@@ -53,6 +53,11 @@ class DeviceHost(DeviceCommon):
self._cwd = None
self._cmd: Optional[List[str]] = None
+ @property
+ def pid(self) -> Optional[int]:
+ """Return the spawned child PID, or ``None`` before start."""
+ return self._child.pid if self._child else None
+
def _dev_is_health_priv(self) -> bool:
"""Check if the host device is OK."""
if not self._child:
diff --git a/src/ntfc/lib/fastboot/controller.py b/src/ntfc/lib/fastboot/controller.py
new file mode 100644
index 0000000..72aeac2
--- /dev/null
+++ b/src/ntfc/lib/fastboot/controller.py
@@ -0,0 +1,124 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+"""Fastboot transport controller."""
+
+import subprocess
+from typing import TYPE_CHECKING, List, Optional
+
+from ntfc.log.logger import logger
+
+if TYPE_CHECKING:
+ from pathlib import Path
+
+
+class FastbootController:
+ """Controls a device via the fastboot protocol.
+
+ Provides device detection, generic command execution, and memory
+ dump retrieval. Commands that target a specific device are issued
+ with the ``-s <dev_sn>`` flag automatically.
+
+ :param dev_sn: Fastboot device serial number.
+ """
+
+ def __init__(self, dev_sn: str) -> None:
+ """Initialize :class:`FastbootController`.
+
+ :param dev_sn: Fastboot device serial number.
+ """
+ self._dev_sn = dev_sn
+
+ def is_connected(self) -> bool:
+ """Return ``True`` if the device is visible in fastboot mode.
+
+ :return: ``True`` when the device serial number appears in the
+ output of ``fastboot devices``. Always ``False`` for an
+ empty serial number, since an empty string is trivially a
+ substring of any output.
+ """
+ if not self._dev_sn:
+ return False
+ try:
+ result = subprocess.run(
+ ["fastboot", "devices"],
+ stdout=subprocess.PIPE,
+ stderr=subprocess.STDOUT,
+ timeout=10.0,
+ )
+ return self._dev_sn in result.stdout.decode(errors="replace")
+ except subprocess.TimeoutExpired:
+ logger.debug("fastboot: devices check timed out")
+ return False
+ except OSError as exc:
+ logger.debug(f"fastboot: devices check failed: {exc}")
+ return False
+
+ def memdump(
+ self, addr: str, size: str, output_path: "Path"
+ ) -> "Optional[Path]":
+ """Dump a memory region from the device to a local file.
+
+ Executes ``fastboot oem memdump <addr> <size>`` then
+ ``fastboot get_staged <output_path>``.
+
+ :param addr: Memory start address (hex string, e.g.
+ ``"0x40000000"``).
+ :param size: Memory region size (hex string, e.g.
+ ``"0x08000000"``).
+ :param output_path: Destination file path for the dump.
+ :return: *output_path* on success, ``None`` on failure.
+ """
+ output_path.parent.mkdir(parents=True, exist_ok=True)
+
+ if not self.run(["oem", "memdump", addr, size], timeout=30.0):
+ logger.warning("fastboot: oem memdump failed")
+ return None
+
+ if not self.run(["get_staged", str(output_path)], timeout=30.0):
+ logger.warning("fastboot: get_staged failed")
+ return None
+
+ return output_path
+
+ def run(self, args: List[str], timeout: float) -> bool:
+ r"""Run a fastboot subcommand against :attr:`_dev_sn`.
+
+ :param args: Arguments appended after ``fastboot -s <dev_sn>``.
+ :param timeout: Maximum seconds to wait for the command.
+ :return: ``True`` on exit code 0, ``False`` otherwise.
+ """
+ cmd = ["fastboot", "-s", self._dev_sn] + args
+ try:
+ result = subprocess.run(
+ cmd,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.STDOUT,
+ timeout=timeout,
+ )
+ output = result.stdout.decode(errors="replace")
+ logger.debug(f"fastboot {' '.join(args)}: {output.strip()}")
+ return result.returncode == 0
+ except subprocess.TimeoutExpired:
+ logger.warning(f"fastboot: {' '.join(args)} timed out")
+ return False
+ except OSError as exc:
+ logger.debug(f"fastboot: {' '.join(args)} failed: {exc}")
+ return False
diff --git a/src/ntfc/lib/ymodem/controller.py b/src/ntfc/lib/ymodem/controller.py
new file mode 100644
index 0000000..ee022ef
--- /dev/null
+++ b/src/ntfc/lib/ymodem/controller.py
@@ -0,0 +1,107 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+"""Ymodem transport controller."""
+
+import subprocess
+from typing import TYPE_CHECKING
+
+from ntfc.log.logger import logger
+
+if TYPE_CHECKING:
+ from pathlib import Path
+
+
+class YmodemController:
+ """Transfers files from a device using the Ymodem protocol.
+
+ Uses ``sbrb.py`` to receive a named file over a serial port. The
+ device must not hold the serial port open during :meth:`download`
+ — the caller is responsible for releasing the port beforehand and
+ reclaiming it afterwards.
+
+ :param serial_port: Host-side serial port path (e.g.
+ ``/dev/ttyUSB0``).
+ :param sbrb_path: Path to the ``sbrb.py`` Ymodem receiver script.
+ :param baud_rate: Baud rate for the serial transfer.
+ """
+
+ def __init__(
+ self,
+ serial_port: str,
+ sbrb_path: "Path",
+ baud_rate: int = 921600,
+ ) -> None:
+ """Initialize :class:`YmodemController`.
+
+ :param serial_port: Host-side serial port path.
+ :param sbrb_path: Path to the ``sbrb.py`` script.
+ :param baud_rate: Baud rate for the Ymodem transfer.
+ """
+ self._serial_port = serial_port
+ self._sbrb_path = sbrb_path
+ self._baud_rate = baud_rate
+
+ def is_ready(self) -> bool:
+ """Return ``True`` if the controller is configured and sbrb exists.
+
+ :return: ``True`` when *serial_port* is non-empty and the
+ ``sbrb.py`` script file exists on the host.
+ """
+ return bool(self._serial_port) and self._sbrb_path.is_file()
+
+ def download(self, output_dir: "Path", remote_filename: str) -> bool:
+ """Transfer a named file from the device to the host via sbrb.py.
+
+ The device's serial port must be released before calling this
+ method.
+
+ :param output_dir: Host directory where the file is saved.
+ :param remote_filename: Name of the file on the device.
+ :return: ``True`` on success, ``False`` on failure or timeout.
+ """
+ try:
+ result = subprocess.run(
+ [
+ str(self._sbrb_path),
+ "-r",
+ remote_filename,
+ "-t",
+ self._serial_port,
+ "-b",
+ str(self._baud_rate),
+ ],
+ cwd=str(output_dir),
+ stdout=subprocess.PIPE,
+ stderr=subprocess.STDOUT,
+ timeout=3600.0,
+ )
+ output = result.stdout.decode(errors="replace")
+ logger.debug(f"ymodem: sbrb output: {output.strip()}")
+ success = "sbrb success" in output
+ if not success:
+ logger.warning("ymodem: sbrb did not report success")
+ return success
+ except subprocess.TimeoutExpired:
+ logger.warning("ymodem: sbrb timed out")
+ return False
+ except OSError as exc:
+ logger.debug(f"ymodem: sbrb launch failed: {exc}")
+ return False
diff --git a/src/ntfc/productconfig.py b/src/ntfc/productconfig.py
index 76e3402..32316f3 100644
--- a/src/ntfc/productconfig.py
+++ b/src/ntfc/productconfig.py
@@ -23,6 +23,7 @@
from typing import Any, Dict, List
from ntfc.coreconfig import CoreConfig
+from ntfc.debug.config import DebugConfig
from ntfc.log.logger import logger
@@ -41,6 +42,9 @@ class ProductConfig:
self._core_index_to_name: Dict[int, str] = {}
self._init_cores()
+ # Optional debug features configuration
+ self._debug = DebugConfig(self._config.get("debug", {}))
+
def _init_cores(self) -> None:
"""Initialize cores dictionary from configuration."""
cores_config = self.cores
@@ -214,6 +218,11 @@ class ProductConfig:
"""Check if platform is AMP."""
return self._platform == "amp"
+ @property
+ def debug(self) -> DebugConfig:
+ """Return product debug configuration."""
+ return self._debug
+
@property
def ignored_cores(self) -> List[str]:
"""Return list of core names to ignore when collecting core info.
diff --git a/src/ntfc/pytest/configure.py b/src/ntfc/pytest/configure.py
index da58bc4..0115919 100644
--- a/src/ntfc/pytest/configure.py
+++ b/src/ntfc/pytest/configure.py
@@ -22,7 +22,7 @@
import os
import time
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Callable, Optional
import pytest
@@ -48,6 +48,21 @@ class PytestConfigPlugin:
self._config = config
self._verbose = verbose
self._recovery_failed = False
+ self._restart_gdb: Optional[Callable[[], None]] = None
+
+ def set_restart_gdb(self, callback: "Callable[[], None]") -> None:
+ """Register a callback fired after a successful device reboot.
+
+ A reboot restarts every product's device process, so any GDB
+ session already attached to one is left talking to a process
+ that no longer exists. Set by
+ :meth:`~ntfc.pytest.mypytest.MyPytest.runner` once the debug
+ setup (and its GDB controllers, if any) exists.
+
+ :param callback: Callable invoked with no arguments after a
+ successful reboot.
+ """
+ self._restart_gdb = callback
def _device_reboot(self) -> None: # pragma: no cover
"""Reboot the device with retry and exponential back-off.
@@ -76,6 +91,8 @@ class PytestConfigPlugin:
success = pytest.product.reboot()
if success:
logger.info("Device recovered on attempt %d", attempt)
+ if self._restart_gdb is not None:
+ self._restart_gdb()
return
logger.warning("Reboot attempt %d returned failure", attempt)
except Exception as exc:
@@ -237,6 +254,13 @@ class PytestConfigPlugin:
) -> Any:
"""Create a TestReport for each of the runtest phases.
+ On a crash this hook reboots the device via ``_device_reboot``.
+ ``ntfc.pytest.coredump_plugin.CoredumpPlugin`` is marked
+ ``hookimpl(hookwrapper=True, trylast=True)`` specifically so its
+ post-yield coredump collection resumes before this hook's
+ post-yield reboot; if that plugin's hook priority ever changes,
+ collection can race the reboot and lose the coredump.
+
:param item:
:param call: the CallInfo for the phase
"""
diff --git a/src/ntfc/pytest/coredump_plugin.py b/src/ntfc/pytest/coredump_plugin.py
new file mode 100644
index 0000000..888e2e2
--- /dev/null
+++ b/src/ntfc/pytest/coredump_plugin.py
@@ -0,0 +1,137 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+"""Pytest plugin that triggers coredump collection on test failures."""
+
+import re
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
+
+import pytest
+from pluggy import HookimplMarker
+
+from ntfc.log.logger import logger
+
+if TYPE_CHECKING:
+ from ntfc.debug.coredump.manager import CoredumpManager
+ from ntfc.product import Product
+
+hookimpl = HookimplMarker("pytest")
+
+
+class CoredumpPlugin:
+ """Collect coredumps from all registered products on test failures.
+
+ Each product maps to a :class:`~ntfc.debug.coredump.manager\
+.CoredumpManager`. When a test body fails the plugin calls
+ :meth:`~ntfc.debug.coredump.manager.CoredumpManager.collect` for
+ every product, writing output into a per-product subdirectory of
+ ``pytest.result_dir``.
+
+ When product instances are provided and at least one of them is in
+ an abnormal state (crashed, busy-looping, or dead), collection is
+ restricted to those products so healthy devices are not stalled and
+ their per-session limits are not consumed by unrelated failures.
+
+ :param managers: Mapping of product name to its
+ :class:`~ntfc.debug.coredump.manager.CoredumpManager`.
+ :param products: Products used for health-based targeting.
+ """
+
+ def __init__(
+ self,
+ managers: "Dict[str, CoredumpManager]",
+ products: "Optional[List[Product]]" = None,
+ ) -> None:
+ """Initialize :class:`CoredumpPlugin`.
+
+ :param managers: Mapping of product name to manager.
+ :param products: Products used for health-based targeting.
+ """
+ self._managers = managers
+ self._products = products or []
+ self._prefix_counts: Dict[str, int] = {}
+
+ def _target_managers(self) -> "Dict[str, CoredumpManager]":
+ """Return the managers to collect from for the current failure.
+
+ :return: Managers of unhealthy products when any product is
+ unhealthy, all managers otherwise.
+ """
+ unhealthy = {
+ p.name
+ for p in self._products
+ if p.crash or p.busyloop or p.notalive
+ }
+ if not unhealthy:
+ return self._managers
+
+ return {
+ name: mgr
+ for name, mgr in self._managers.items()
+ if name in unhealthy
+ }
+
+ def _unique_prefix(self, nodeid: str) -> str:
+ """Derive a session-unique filename prefix from a test node ID.
+
+ Tests in different files or classes can share their terminal
+ name (and reruns repeat it), so repeated prefixes get a numeric
+ suffix to avoid overwriting earlier coredumps.
+
+ :param nodeid: Pytest node ID of the failed test.
+ :return: Sanitized, session-unique prefix.
+ """
+ base = re.sub(r"[^\w.-]", "_", nodeid.split("::")[-1])
+ count = self._prefix_counts.get(base, 0)
+ self._prefix_counts[base] = count + 1
+ return base if count == 0 else f"{base}.{count + 1}"
+
+ def _on_test_failed(self, nodeid: str, result_dir: str) -> None:
+ """Trigger coredump collection for all products.
+
+ :param nodeid: Pytest node ID of the failed test.
+ :param result_dir: Session result directory path.
+ """
+ prefix = self._unique_prefix(nodeid)
+ out = Path(result_dir)
+ for name, mgr in self._target_managers().items():
+ logger.debug(
+ f"coredump: collecting for product {name} prefix {prefix}"
+ )
+ mgr.collect(out / name, prefix)
+
+ @hookimpl(hookwrapper=True, trylast=True)
+ def pytest_runtest_makereport(self, item: Any, call: Any) -> Any:
+ """Collect coredump when a test call phase fails.
+
+ :param item: The pytest test item.
+ :param call: The call info for the phase.
+ :yield: None
+ """
+ del call
+ outcome = yield
+ report = outcome.get_result()
+ if report.when != "call" or not report.failed:
+ return
+ result_dir = getattr(pytest, "result_dir", "")
+ if not result_dir:
+ return
+ self._on_test_failed(item.nodeid, result_dir)
diff --git a/src/ntfc/pytest/mypytest.py b/src/ntfc/pytest/mypytest.py
index 5817954..b85f695 100644
--- a/src/ntfc/pytest/mypytest.py
+++ b/src/ntfc/pytest/mypytest.py
@@ -30,6 +30,7 @@ import pytest
import yaml # type: ignore
from pluggy import HookimplMarker
+from ntfc.debug.getdebug import get_debug_setup
from ntfc.envconfig import EnvConfig
from ntfc.log.logger import logger
from ntfc.log.manager import LogManager
@@ -369,12 +370,19 @@ class MyPytest:
# run pytest with our custom test plugin
runner = RunnerPlugin(nologs)
+ # optional debug plugins (coredump collection)
+ debug = get_debug_setup(pytest.products)
+ self._ptconfig.set_restart_gdb(debug.restart_all_controllers)
+
# start device before test start
self._device_start()
try:
- return self._run(opt, [runner, collector])
+ # attach GDB controllers after the devices are up
+ debug.start(pytest.result_dir)
+ return self._run(opt, [runner, collector, *debug.plugins])
finally:
+ debug.stop()
self._device_stop()
def collect(self, testpath: str, reinit: bool = True) -> "Collected":
diff --git a/tests/conftest.py b/tests/conftest.py
index 8a7958a..0281a85 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -23,6 +23,8 @@ from dummydev import DeviceDummy
from ntfc.envconfig import EnvConfig
+pytest_plugins = ["pytester"]
+
@pytest.fixture
def config_dummy():
diff --git a/tests/debug/conftest.py b/tests/debug/conftest.py
new file mode 100644
index 0000000..20e44e8
--- /dev/null
+++ b/tests/debug/conftest.py
@@ -0,0 +1,164 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+"""Shared fixtures for debug tests: a fake in-GDB ``gdb`` module."""
+
+import shlex
+import sys
+import types
+from typing import Any, Callable, Generator, List, Optional, Tuple
+
+import pytest
+
+# Plugin modules are imported both as submodules (by tests) and as
+# top-level modules (by the init.py loader inside GDB).
+_PLUGIN_MODULES = (
+ "ntfc.debug.gdb.plugin.autobp",
+ "ntfc.debug.gdb.plugin.gdbprefix",
+ "ntfc.debug.gdb.plugin.init",
+ "autobp",
+ "gdbprefix",
+)
+
+
+def _breakpoint_class() -> Any:
+ """Build a fresh fake ``gdb.Breakpoint`` class."""
+
+ class FakeBreakpoint:
+ instances: List["FakeBreakpoint"] = []
+ valid = True
+ delete_error: Optional[Exception] = None
+
+ def __init__(
+ self,
+ spec: str,
+ type: int = 0, # noqa: A002
+ internal: bool = False,
+ ) -> None:
+ self.spec = spec
+ self.bp_type = type
+ self.internal = internal
+ self.deleted = False
+ FakeBreakpoint.instances.append(self)
+
+ def is_valid(self) -> bool:
+ return self.valid
+
+ def delete(self) -> None:
+ if FakeBreakpoint.delete_error is not None:
+ raise FakeBreakpoint.delete_error
+ self.deleted = True
+
+ return FakeBreakpoint
+
+
+def _command_class() -> Any:
+ """Build a fresh fake ``gdb.Command`` class."""
+
+ class FakeCommand:
+ registered: List[Tuple[str, "FakeCommand"]] = []
+
+ def __init__(self, name: str, command_class: int) -> None:
+ self.name = name
+ FakeCommand.registered.append((name, self))
+
+ return FakeCommand
+
+
+def _event_classes() -> Tuple[Any, Any, Any]:
+ """Build fresh fake GDB event registry and event classes."""
+
+ class FakeEventRegistry:
+ def __init__(self) -> None:
+ self.callbacks: List[Callable[[Any], None]] = []
+
+ def connect(self, callback: Callable[[Any], None]) -> None:
+ self.callbacks.append(callback)
+
+ class FakeBreakpointEvent:
+ def __init__(self, breakpoints: Any = ()) -> None:
+ self.breakpoints = breakpoints
+
+ class FakeSignalEvent:
+ def __init__(self, stop_signal: str) -> None:
+ self.stop_signal = stop_signal
+
+ return FakeEventRegistry, FakeBreakpointEvent, FakeSignalEvent
+
+
+def build_fake_gdb() -> types.ModuleType:
+ """Build a fake ``gdb`` module mimicking the in-GDB Python API.
+
+ :return: Module object providing the subset of the GDB Python API
+ used by the NTFC plugin scripts.
+ """
+ fake = types.ModuleType("gdb")
+
+ class FakeError(Exception):
+ pass
+
+ registry_cls, bp_event_cls, signal_event_cls = _event_classes()
+
+ executed: List[str] = []
+
+ def execute(command: str, to_string: bool = False) -> str:
+ executed.append(command)
+ return ""
+
+ fake.error = FakeError
+ fake.Breakpoint = _breakpoint_class()
+ fake.Command = _command_class()
+ fake.BreakpointEvent = bp_event_cls
+ fake.SignalEvent = signal_event_cls
+ fake.BP_WATCHPOINT = 1
+ fake.COMMAND_USER = 2
+ fake.string_to_argv = shlex.split
+ fake.events = types.SimpleNamespace(
+ stop=registry_cls(), exited=registry_cls()
+ )
+ fake.execute = execute
+ fake.executed = executed
+
+ return fake
+
+
+def _purge_plugin_modules() -> None:
+ """Remove plugin modules from :data:`sys.modules`."""
+ for name in _PLUGIN_MODULES:
+ sys.modules.pop(name, None)
+
+
+@pytest.fixture
+def fake_gdb(
+ monkeypatch: pytest.MonkeyPatch,
+) -> Generator[types.ModuleType, None, None]:
+ """Fixture: install a fake ``gdb`` module and reset plugin imports.
+
+ Plugin modules register commands and connect events at import time,
+ so they are removed from :data:`sys.modules` before and after each
+ test to force a fresh import against the fake module.
+
+ :yield: The fake ``gdb`` module.
+ """
+ fake = build_fake_gdb()
+ monkeypatch.setitem(sys.modules, "gdb", fake)
+ _purge_plugin_modules()
+ yield fake
+ _purge_plugin_modules()
diff --git a/tests/debug/test_config.py b/tests/debug/test_config.py
new file mode 100644
index 0000000..cfc6b87
--- /dev/null
+++ b/tests/debug/test_config.py
@@ -0,0 +1,331 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+import pytest
+
+from ntfc.debug.config import (
+ CoredumpConfig,
+ DebugConfig,
+ FastbootConfig,
+ GdbConfig,
+ LocalFileConfig,
+ SyslogConfig,
+ YmodemConfig,
+)
+
+
+class TestCoredumpConfig:
+ def test_defaults(self):
+ cfg = CoredumpConfig({})
+ assert cfg.enable is False
+ assert cfg.collection_type == "auto"
+ assert cfg.limit == 5
+
+ def test_full_config(self):
+ cfg = CoredumpConfig({"enable": True, "type": "gdb", "limit": 10})
+ assert cfg.enable is True
+ assert cfg.collection_type == "gdb"
+ assert cfg.limit == 10
+
+ def test_all_valid_types(self):
+ for t in ("auto", "gdb", "fastboot", "ymodem"):
+ cfg = CoredumpConfig({"type": t})
+ assert cfg.collection_type == t
+
+ def test_invalid_type_raises(self):
+ with pytest.raises(ValueError, match="Invalid coredump type"):
+ CoredumpConfig({"type": "invalid"})
+
+ def test_enable_false(self):
+ cfg = CoredumpConfig({"enable": False})
+ assert cfg.enable is False
+
+ def test_limit_zero(self):
+ cfg = CoredumpConfig({"limit": 0})
+ assert cfg.limit == 0
+
+
+class TestGdbConfig:
+ def test_defaults(self):
+ cfg = GdbConfig({})
+ assert cfg.enable is False
+ assert cfg.force_panic is False
+ assert cfg.target == ""
+
+ def test_full_config(self):
+ cfg = GdbConfig({"enable": True, "force_panic": True})
+ assert cfg.enable is True
+ assert cfg.force_panic is True
+
+ def test_enable_only(self):
+ cfg = GdbConfig({"enable": True})
+ assert cfg.enable is True
+ assert cfg.force_panic is False
+
+ def test_force_panic_only(self):
+ cfg = GdbConfig({"force_panic": True})
+ assert cfg.enable is False
+ assert cfg.force_panic is True
+
+ def test_target_tcp(self):
+ cfg = GdbConfig({"target": "localhost:1234"})
+ assert cfg.target == "localhost:1234"
+
+ def test_target_unix_socket(self):
+ cfg = GdbConfig({"target": "/tmp/gdb.socket"})
+ assert cfg.target == "/tmp/gdb.socket"
+
+ def test_target_default_empty(self):
+ cfg = GdbConfig({})
+ assert cfg.target == ""
+
+ def test_gdb_path_default(self):
+ cfg = GdbConfig({})
+ assert cfg.gdb_path == "gdb"
+
+ def test_gdb_path_custom(self):
+ cfg = GdbConfig({"gdb_path": "gdb-multiarch"})
+ assert cfg.gdb_path == "gdb-multiarch"
+
+ def test_plugin_default(self):
+ cfg = GdbConfig({})
+ assert cfg.plugin is False
+
+ def test_plugin_enabled(self):
+ cfg = GdbConfig({"plugin": True})
+ assert cfg.plugin is True
+
+ def test_setup_cmds_default_empty(self):
+ cfg = GdbConfig({})
+ assert cfg.setup_cmds == []
+
+ def test_setup_cmds_parsed(self):
+ cmds = ["break main", "continue"]
+ cfg = GdbConfig({"setup_cmds": cmds})
+ assert cfg.setup_cmds == cmds
+
+ def test_setup_cmds_copy_returned(self):
+ cfg = GdbConfig({"setup_cmds": ["break main"]})
+ cfg.setup_cmds.append("mutated")
+ assert cfg.setup_cmds == ["break main"]
+
+
+class TestFastbootConfig:
+ def test_defaults(self):
+ cfg = FastbootConfig({})
+ assert cfg.dev_sn == ""
+ assert cfg.mem_addr == "0x40000000"
+ assert cfg.mem_size == "0x08000000"
+
+ def test_custom_dev_sn(self):
+ cfg = FastbootConfig({"dev_sn": "ABC123"})
+ assert cfg.dev_sn == "ABC123"
+
+ def test_custom_mem_region(self):
+ cfg = FastbootConfig(
+ {"mem_addr": "0x20000000", "mem_size": "0x04000000"}
+ )
+ assert cfg.mem_addr == "0x20000000"
+ assert cfg.mem_size == "0x04000000"
+
+ def test_full_config(self):
+ cfg = FastbootConfig(
+ {
+ "dev_sn": "XYZ999",
+ "mem_addr": "0x10000000",
+ "mem_size": "0x01000000",
+ }
+ )
+ assert cfg.dev_sn == "XYZ999"
+ assert cfg.mem_addr == "0x10000000"
+ assert cfg.mem_size == "0x01000000"
+
+
+class TestYmodemConfig:
+ def test_defaults(self):
+ cfg = YmodemConfig({})
+ assert cfg.serial_port == ""
+ assert cfg.baud_rate == 921600
+ assert cfg.sbrb_path == ""
+
+ def test_custom_serial_port(self):
+ cfg = YmodemConfig({"serial_port": "/dev/ttyUSB0"})
+ assert cfg.serial_port == "/dev/ttyUSB0"
+
+ def test_custom_baud_rate(self):
+ cfg = YmodemConfig({"baud_rate": 115200})
+ assert cfg.baud_rate == 115200
+
+ def test_custom_sbrb_path(self):
+ cfg = YmodemConfig({"sbrb_path": "/opt/tools/sbrb.py"})
+ assert cfg.sbrb_path == "/opt/tools/sbrb.py"
+
+ def test_full_config(self):
+ cfg = YmodemConfig(
+ {
+ "serial_port": "/dev/ttyUSB1",
+ "baud_rate": 460800,
+ "sbrb_path": "/tools/sbrb.py",
+ }
+ )
+ assert cfg.serial_port == "/dev/ttyUSB1"
+ assert cfg.baud_rate == 460800
+ assert cfg.sbrb_path == "/tools/sbrb.py"
+
+
+class TestDebugConfig:
+ def test_defaults_empty(self):
+ cfg = DebugConfig({})
+ assert cfg.coredump.enable is False
+ assert cfg.coredump.collection_type == "auto"
+ assert cfg.coredump.limit == 5
+ assert cfg.gdb.enable is False
+ assert cfg.gdb.force_panic is False
+ assert cfg.gdb.target == ""
+ assert cfg.fastboot.dev_sn == ""
+ assert cfg.fastboot.mem_addr == "0x40000000"
+ assert cfg.ymodem.serial_port == ""
+ assert cfg.ymodem.baud_rate == 921600
+ assert cfg.ymodem.sbrb_path == ""
+
+ def test_full_config(self):
+ cfg = DebugConfig(
+ {
+ "coredump": {
+ "enable": True,
+ "type": "fastboot",
+ "limit": 3,
+ },
+ "gdb": {
+ "enable": True,
+ "force_panic": True,
+ },
+ "fastboot": {
+ "dev_sn": "ABC123",
+ },
+ "ymodem": {
+ "serial_port": "/dev/ttyUSB0",
+ "sbrb_path": "/tools/sbrb.py",
+ },
+ }
+ )
+ assert cfg.coredump.enable is True
+ assert cfg.coredump.collection_type == "fastboot"
+ assert cfg.coredump.limit == 3
+ assert cfg.gdb.enable is True
+ assert cfg.gdb.force_panic is True
+ assert cfg.fastboot.dev_sn == "ABC123"
+ assert cfg.ymodem.serial_port == "/dev/ttyUSB0"
+ assert cfg.ymodem.sbrb_path == "/tools/sbrb.py"
+
+ def test_coredump_only(self):
+ cfg = DebugConfig({"coredump": {"enable": True}})
+ assert cfg.coredump.enable is True
+ assert cfg.gdb.enable is False
+
+ def test_gdb_only(self):
+ cfg = DebugConfig({"gdb": {"enable": True}})
+ assert cfg.coredump.enable is False
+ assert cfg.gdb.enable is True
+
+ def test_invalid_coredump_type_propagates(self):
+ with pytest.raises(ValueError, match="Invalid coredump type"):
+ DebugConfig({"coredump": {"type": "unknown"}})
+
+
+class TestGdbConfigParity:
+ def test_new_defaults(self):
+ cfg = GdbConfig({})
+ assert cfg.gcore_cmd == "gcore"
+ assert cfg.nx_plugin == ""
+ assert cfg.osabi == ""
+ assert cfg.auto_breakpoints is False
+ assert cfg.mmleak is False
+ assert cfg.attach is False
+ assert cfg.use_sudo is False
+
+ def test_new_values(self):
+ cfg = GdbConfig(
+ {
+ "plugin": True,
+ "gcore_cmd": "gcore -t",
+ "nx_plugin": "/nx/gdbinit.py",
+ "osabi": "none",
+ "auto_breakpoints": True,
+ "mmleak": True,
+ "attach": True,
+ "use_sudo": True,
+ }
+ )
+ assert cfg.gcore_cmd == "gcore -t"
+ assert cfg.nx_plugin == "/nx/gdbinit.py"
+ assert cfg.osabi == "none"
+ assert cfg.auto_breakpoints is True
+ assert cfg.mmleak is True
+ assert cfg.attach is True
+ assert cfg.use_sudo is True
+
+ def test_auto_breakpoints_requires_plugin(self):
+ with pytest.raises(ValueError, match="auto_breakpoints"):
+ GdbConfig({"auto_breakpoints": True})
+
+ def test_mmleak_requires_auto_breakpoints(self):
+ with pytest.raises(ValueError, match="mmleak"):
+ GdbConfig({"plugin": True, "mmleak": True})
+
+ def test_attach_and_target_mutually_exclusive(self):
+ with pytest.raises(ValueError, match="attach"):
+ GdbConfig({"attach": True, "target": "localhost:1234"})
+
+
+class TestLocalFileConfig:
+ def test_defaults(self):
+ cfg = LocalFileConfig({})
+ assert cfg.core_dir == ""
+ assert cfg.pattern == "*.core"
+
+ def test_values(self):
+ cfg = LocalFileConfig({"core_dir": "/tmp/x", "pattern": "core*"})
+ assert cfg.core_dir == "/tmp/x"
+ assert cfg.pattern == "core*"
+
+
+class TestSyslogConfig:
+ def test_defaults(self):
+ assert SyslogConfig({}).enable is False
+
+ def test_enabled(self):
+ assert SyslogConfig({"enable": True}).enable is True
+
+
+class TestDebugConfigParity:
+ def test_new_sections(self):
+ cfg = DebugConfig(
+ {
+ "local_file": {"core_dir": "/cores"},
+ "syslog": {"enable": True},
+ }
+ )
+ assert cfg.local_file.core_dir == "/cores"
+ assert cfg.syslog.enable is True
+
+ def test_new_coredump_types_valid(self):
+ for t in ("local_file", "syslog"):
+ assert CoredumpConfig({"type": t}).collection_type == t
diff --git a/tests/debug/test_coredump_handlers.py b/tests/debug/test_coredump_handlers.py
new file mode 100644
index 0000000..e8feb8b
--- /dev/null
+++ b/tests/debug/test_coredump_handlers.py
@@ -0,0 +1,628 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+import base64
+import os
+import struct
+from typing import TYPE_CHECKING
+from unittest.mock import MagicMock
+
+import pytest
+
+if TYPE_CHECKING:
+ from pathlib import Path
+
+from ntfc.debug.config import LocalFileConfig
+from ntfc.debug.coredump.fastboot_handler import FastbootHandler
+from ntfc.debug.coredump.gdb_handler import GdbHandler
+from ntfc.debug.coredump.local_file_handler import LocalFileHandler
+from ntfc.debug.coredump.syslog_handler import SyslogHandler
+from ntfc.debug.coredump.ymodem_handler import YmodemHandler
+from ntfc.device.common import CmdReturn, CmdStatus
+
+
+class TestGdbHandler:
+ def _make_handler(
+ self, running: bool = True, corefile: "Path | None" = None
+ ) -> GdbHandler:
+ controller = MagicMock()
+ controller.is_running.return_value = running
+ controller.generate_coredump.return_value = corefile
+ return GdbHandler(controller)
+
+ def test_name(self) -> None:
+ h = self._make_handler()
+ assert h.name == "gdb"
+
+ def test_priority(self) -> None:
+ h = self._make_handler()
+ assert h.priority == 10
+
+ def test_is_available_when_running(self) -> None:
+ h = self._make_handler(running=True)
+ assert h.is_available() is True
+
+ def test_is_available_when_stopped(self) -> None:
+ h = self._make_handler(running=False)
+ assert h.is_available() is False
+
+ def test_collect_returns_true_when_controller_returns_path(
+ self, tmp_path: "Path"
+ ) -> None:
+ h = self._make_handler(corefile=tmp_path / "core")
+ assert h.collect(tmp_path, "prefix") is True
+
+ def test_collect_returns_false_when_controller_returns_none(
+ self, tmp_path: "Path"
+ ) -> None:
+ h = self._make_handler(corefile=None)
+ assert h.collect(tmp_path, "prefix") is False
+
+ def test_collect_passes_args_to_controller(self, tmp_path: "Path") -> None:
+ controller = MagicMock()
+ controller.generate_coredump.return_value = tmp_path / "x.core"
+ h = GdbHandler(controller)
+ h.collect(tmp_path, "myprefix")
+ controller.generate_coredump.assert_called_once_with(
+ tmp_path, "myprefix"
+ )
+
+ def test_collect_forces_panic_before_coredump(
+ self, tmp_path: "Path"
+ ) -> None:
+ calls = []
+ controller = MagicMock()
+ controller.generate_coredump.side_effect = (
+ lambda *a: calls.append("gcore") or tmp_path / "x.core"
+ )
+ panic = MagicMock(side_effect=lambda: calls.append("panic") or True)
+ h = GdbHandler(controller, force_panic=panic)
+ assert h.collect(tmp_path, "prefix") is True
+ assert calls == ["panic", "gcore"]
+
+ def test_collect_continues_when_panic_fails(
+ self, tmp_path: "Path"
+ ) -> None:
+ controller = MagicMock()
+ controller.generate_coredump.return_value = tmp_path / "x.core"
+ panic = MagicMock(return_value=False)
+ h = GdbHandler(controller, force_panic=panic)
+ assert h.collect(tmp_path, "prefix") is True
+ controller.generate_coredump.assert_called_once()
+
+ def test_collect_skips_panic_when_not_configured(
+ self, tmp_path: "Path"
+ ) -> None:
+ h = self._make_handler(corefile=tmp_path / "x.core")
+ assert h.collect(tmp_path, "prefix") is True
+
+ def test_enabled_by_default(self) -> None:
+ h = self._make_handler()
+ assert h.is_enabled() is True
+
+ def test_disable_then_enable(self) -> None:
+ h = self._make_handler()
+ h.disable()
+ assert h.is_enabled() is False
+ h.enable()
+ assert h.is_enabled() is True
+
+ def test_collect_waits_for_auto_dump_on_crash(
+ self, tmp_path: "Path"
+ ) -> None:
+ src = tmp_path / "crash.123.core"
+ src.write_bytes(b"\x7fELF")
+ controller = MagicMock()
+ controller.wait_corefile.return_value = src
+ h = GdbHandler(controller, crash_check=lambda: True, auto_dump=True)
+ out = tmp_path / "out"
+ out.mkdir()
+ assert h.collect(out, "test_x") is True
+ controller.wait_corefile.assert_called_once_with(
+ timeout=GdbHandler.AUTO_DUMP_TIMEOUT
+ )
+ controller.generate_coredump.assert_not_called()
+ assert (out / "test_x.core").read_bytes() == b"\x7fELF"
+
+ def test_collect_auto_dump_timeout_returns_false(
+ self, tmp_path: "Path"
+ ) -> None:
+ controller = MagicMock()
+ controller.wait_corefile.return_value = None
+ h = GdbHandler(controller, crash_check=lambda: True, auto_dump=True)
+ assert h.collect(tmp_path, "t") is False
+ controller.generate_coredump.assert_not_called()
+
+ def test_collect_pull_path_when_not_crashed(
+ self, tmp_path: "Path"
+ ) -> None:
+ controller = MagicMock()
+ controller.generate_coredump.return_value = tmp_path / "t.core"
+ h = GdbHandler(controller, crash_check=lambda: False, auto_dump=True)
+ assert h.collect(tmp_path, "t") is True
+ controller.wait_corefile.assert_not_called()
+
+ def test_collect_pull_path_without_auto_dump(
+ self, tmp_path: "Path"
+ ) -> None:
+ controller = MagicMock()
+ controller.generate_coredump.return_value = tmp_path / "t.core"
+ h = GdbHandler(controller, crash_check=lambda: True, auto_dump=False)
+ assert h.collect(tmp_path, "t") is True
+ controller.wait_corefile.assert_not_called()
+
+
+class TestFastbootHandler:
+ def _make_handler(
+ self,
+ connected: bool = True,
+ memdump_result: "Path | None" = None,
+ mem_addr: str = "0x40000000",
+ mem_size: str = "0x08000000",
+ ) -> FastbootHandler:
+ controller = MagicMock()
+ controller.is_connected.return_value = connected
+ controller.memdump.return_value = memdump_result
+ return FastbootHandler(controller, mem_addr, mem_size)
+
+ def test_name(self) -> None:
+ assert self._make_handler().name == "fastboot"
+
+ def test_priority(self) -> None:
+ assert self._make_handler().priority == 20
+
+ def test_is_available_when_connected(self) -> None:
+ assert self._make_handler(connected=True).is_available() is True
+
+ def test_is_available_when_disconnected(self) -> None:
+ assert self._make_handler(connected=False).is_available() is False
+
+ def test_collect_returns_true_when_memdump_returns_path(
+ self, tmp_path: "Path"
+ ) -> None:
+ h = self._make_handler(memdump_result=tmp_path / "core.bin")
+ assert h.collect(tmp_path, "prefix") is True
+
+ def test_collect_returns_false_when_memdump_returns_none(
+ self, tmp_path: "Path"
+ ) -> None:
+ h = self._make_handler(memdump_result=None)
+ assert h.collect(tmp_path, "prefix") is False
+
+ def test_collect_passes_addr_size_and_path_to_controller(
+ self, tmp_path: "Path"
+ ) -> None:
+ controller = MagicMock()
+ controller.memdump.return_value = tmp_path / "myprefix.bin"
+ h = FastbootHandler(controller, "0x20000000", "0x04000000")
+ h.collect(tmp_path, "myprefix")
+ controller.memdump.assert_called_once_with(
+ "0x20000000",
+ "0x04000000",
+ tmp_path / "myprefix.bin",
+ )
+
+ def test_enabled_by_default(self) -> None:
+ assert self._make_handler().is_enabled() is True
+
+ def test_disable_then_enable(self) -> None:
+ h = self._make_handler()
+ h.disable()
+ assert h.is_enabled() is False
+ h.enable()
+ assert h.is_enabled() is True
+
+
+def _make_device(
+ output: str = "", status: CmdStatus = CmdStatus.SUCCESS
+) -> MagicMock:
+ d = MagicMock()
+ d.send_cmd_read_until_pattern.return_value = CmdReturn(
+ status=status, output=output
+ )
+ return d
+
+
+class TestYmodemHandler:
+ def _make_handler(
+ self,
+ ready: bool = True,
+ ls_output: str = "app.core\n",
+ ls_status: CmdStatus = CmdStatus.SUCCESS,
+ download_result: bool = True,
+ ) -> YmodemHandler:
+ controller = MagicMock()
+ controller.is_ready.return_value = ready
+ controller.download.return_value = download_result
+ device = _make_device(output=ls_output, status=ls_status)
+ return YmodemHandler(controller, device)
+
+ def test_name(self) -> None:
+ assert self._make_handler().name == "ymodem"
+
+ def test_priority(self) -> None:
+ assert self._make_handler().priority == 30
+
+ def test_is_available_when_ready(self) -> None:
+ assert self._make_handler(ready=True).is_available() is True
+
+ def test_is_available_when_not_ready(self) -> None:
+ assert self._make_handler(ready=False).is_available() is False
+
+ def test_collect_returns_true_on_success(self, tmp_path: "Path") -> None:
+ assert self._make_handler().collect(tmp_path, "test") is True
+
+ def test_collect_returns_false_when_ls_finds_no_core(
+ self, tmp_path: "Path"
+ ) -> None:
+ h = self._make_handler(ls_output="no files here\n")
+ assert h.collect(tmp_path, "test") is False
+
+ def test_collect_returns_false_when_ls_times_out(
+ self, tmp_path: "Path"
+ ) -> None:
+ h = self._make_handler(ls_status=CmdStatus.TIMEOUT)
+ assert h.collect(tmp_path, "test") is False
+
+ def test_collect_returns_false_when_download_fails(
+ self, tmp_path: "Path"
+ ) -> None:
+ assert (
+ self._make_handler(download_result=False).collect(tmp_path, "t")
+ is False
+ )
+
+ def test_collect_does_not_download_when_find_fails(
+ self, tmp_path: "Path"
+ ) -> None:
+ controller = MagicMock()
+ controller.is_ready.return_value = True
+ controller.download.return_value = True
+ device = _make_device(output="no files here\n")
+ h = YmodemHandler(controller, device)
+ h.collect(tmp_path, "t")
+ controller.download.assert_not_called()
+ device.stop.assert_not_called()
+
+ def test_collect_releases_port_during_download(
+ self, tmp_path: "Path"
+ ) -> None:
+ """Device is stopped before download and restarted after."""
+ calls = []
+ controller = MagicMock()
+ controller.is_ready.return_value = True
+ controller.download.side_effect = (
+ lambda *a: calls.append("download") or True
+ )
+ device = _make_device(output="app.core\n")
+ device.stop.side_effect = lambda: calls.append("stop")
+ device.start.side_effect = lambda: calls.append("start")
+ h = YmodemHandler(controller, device)
+ assert h.collect(tmp_path, "t") is True
+ assert calls == ["stop", "download", "start"]
+
+ def test_collect_restarts_device_when_download_raises(
+ self, tmp_path: "Path"
+ ) -> None:
+ controller = MagicMock()
+ controller.is_ready.return_value = True
+ controller.download.side_effect = RuntimeError("boom")
+ device = _make_device(output="app.core\n")
+ h = YmodemHandler(controller, device)
+ with pytest.raises(RuntimeError):
+ h.collect(tmp_path, "t")
+ device.start.assert_called_once()
+
+ def test_enabled_by_default(self) -> None:
+ assert self._make_handler().is_enabled() is True
+
+ def test_disable_then_enable(self) -> None:
+ h = self._make_handler()
+ h.disable()
+ assert h.is_enabled() is False
+ h.enable()
+ assert h.is_enabled() is True
+
+
+class TestLocalFileHandler:
+ def _make(self, core_dir: "Path") -> LocalFileHandler:
+ return LocalFileHandler(LocalFileConfig({"core_dir": str(core_dir)}))
+
+ def test_name_and_priority(self, tmp_path: "Path") -> None:
+ h = self._make(tmp_path)
+ assert h.name == "local_file"
+ assert h.priority == 15
+
+ def test_unavailable_when_dir_missing(self, tmp_path: "Path") -> None:
+ assert self._make(tmp_path / "nope").is_available() is False
+
+ def test_collects_newest_matching_file(self, tmp_path: "Path") -> None:
+ src = tmp_path / "cores"
+ src.mkdir()
+ old = src / "old.core"
+ old.write_bytes(b"OLD1")
+ new = src / "new.core"
+ new.write_bytes(b"NEW1")
+ os.utime(old, (1, 1))
+ out = tmp_path / "out"
+ h = self._make(src)
+ assert h.collect(out, "test_x") is True
+ assert (out / "test_x.core").read_bytes() == b"NEW1"
+ assert not new.exists()
+
+ def test_does_not_recollect_consumed_file(self, tmp_path: "Path") -> None:
+ src = tmp_path / "cores"
+ src.mkdir()
+ (src / "a.core").write_bytes(b"AAAA")
+ out = tmp_path / "out"
+ h = self._make(src)
+ assert h.collect(out, "one") is True
+ assert h.collect(out, "two") is False
+
+ def test_ignores_empty_files(self, tmp_path: "Path") -> None:
+ src = tmp_path / "cores"
+ src.mkdir()
+ (src / "empty.core").write_bytes(b"")
+ h = self._make(src)
+ assert h.collect(tmp_path / "out", "t") is False
+
+ def test_respects_pattern(self, tmp_path: "Path") -> None:
+ src = tmp_path / "cores"
+ src.mkdir()
+ (src / "x.dump").write_bytes(b"DUMP")
+ h = LocalFileHandler(
+ LocalFileConfig({"core_dir": str(src), "pattern": "*.dump"})
+ )
+ assert h.collect(tmp_path / "out", "t") is True
+
+
+class TestYmodemHandlerParseCorefile:
+ def test_returns_name_for_unformatted_file(self) -> None:
+ assert YmodemHandler._parse_corefile("app.core\n") == "app.core"
+
+ def test_skips_already_stamped_files(self) -> None:
+ out = "crash.2025.01.01_12.00.00.core\n"
+ assert YmodemHandler._parse_corefile(out) is None
+
+ def test_returns_first_unformatted_when_mixed(self) -> None:
+ out = "crash.2025.01.01_12.00.00.core\nfresh.core\n"
+ assert YmodemHandler._parse_corefile(out) == "fresh.core"
+
+ def test_returns_none_when_no_core_files(self) -> None:
+ assert YmodemHandler._parse_corefile("no files here\n") is None
+
+ def test_returns_none_on_empty_output(self) -> None:
+ assert YmodemHandler._parse_corefile("") is None
+
+
+class TestYmodemHandlerFindCoredump:
+ def test_returns_renamed_name_on_success(self) -> None:
+ device = _make_device(output="app.core\n")
+ h = YmodemHandler(MagicMock(), device)
+ result = h._find_coredump("test")
+ assert result is not None
+ assert result.startswith("test.")
+ assert result.endswith(".core")
+
+ def test_returns_none_when_ls_times_out(self) -> None:
+ device = _make_device(output="", status=CmdStatus.TIMEOUT)
+ h = YmodemHandler(MagicMock(), device)
+ assert h._find_coredump("test") is None
+
+ def test_returns_none_when_all_files_stamped(self) -> None:
+ device = _make_device(output="crash.2025.01.01_00.00.00.core\n")
+ h = YmodemHandler(MagicMock(), device)
+ assert h._find_coredump("test") is None
+
+ def test_sends_rename_command_on_success(self) -> None:
+ device = _make_device(output="app.core\n")
+ h = YmodemHandler(MagicMock(), device)
+ h._find_coredump("myprefix")
+ assert device.send_cmd_read_until_pattern.call_count == 2
+ mv_call = device.send_cmd_read_until_pattern.call_args_list[1]
+ assert b"mv" in mv_call[0][0]
+ assert b"myprefix" in mv_call[0][0]
+
+
+def _syslog_output(payload_b64: str) -> str:
+ lines = ["[ 1.0] Start coredump\n"]
+ for i in range(0, len(payload_b64), 8):
+ lines.append(f"[ 1.1] {payload_b64[i:i + 8]}\n")
+ lines.append("[ 1.2] Finish coredump\n")
+ return "boot noise\n" + "".join(lines) + "after noise\n"
+
+
+def _raw_block(data: bytes) -> bytes:
+ return b"ZV\x00" + struct.pack(">H", len(data)) + data
+
+
+class TestSyslogHandler:
+ def _make(self, output: str) -> SyslogHandler:
+ device = MagicMock()
+ device.output_tail.return_value = output
+ return SyslogHandler(device)
+
+ def test_name_and_priority(self) -> None:
+ h = self._make("")
+ assert h.name == "syslog"
+ assert h.priority == 25
+ assert h.is_available() is True
+
+ def test_collect_decodes_raw_blocks(
+ self, monkeypatch: pytest.MonkeyPatch, tmp_path: "Path"
+ ) -> None:
+ monkeypatch.setattr(
+ "ntfc.debug.coredump.syslog_handler.lzf", MagicMock()
+ )
+ payload = base64.b64encode(_raw_block(b"\x7fELF-core")).decode()
+ h = self._make(_syslog_output(payload))
+ assert h.collect(tmp_path, "t") is True
+ assert (tmp_path / "t.core").read_bytes() == b"\x7fELF-core"
+ h._device.reset_output_tail.assert_called_once()
+
+ def test_collect_clears_tail_even_when_no_markers_found(
+ self, tmp_path: "Path"
+ ) -> None:
+ h = self._make("no markers here")
+ assert h.collect(tmp_path, "t") is False
+ h._device.reset_output_tail.assert_called_once()
+
+ def test_collect_decompresses_lzf_blocks(self, tmp_path: "Path") -> None:
+ """Round-trip a real liblzf-compressed block through the actual
+ ``lzf`` cffi package (not a mock) -- a mocked ``.decompress()``
+ previously hid an API mismatch with the real package, which has
+ no such module-level function."""
+ lzf = pytest.importorskip("lzf")
+ plaintext = b"\x7fELF-expanded" * 200
+ dest = bytearray(lzf.LZF_MAX_COMPRESSED_SIZE(len(plaintext)))
+ compressed_len = lzf.lib.lzf_compress(
+ lzf.ffi.from_buffer(plaintext),
+ len(plaintext),
+ lzf.ffi.from_buffer(dest),
+ len(dest),
+ )
+ assert compressed_len > 0
+ compressed = bytes(dest[:compressed_len])
+
+ block = (
+ b"ZV\x01"
+ + struct.pack(">H", compressed_len)
+ + struct.pack(">H", len(plaintext))
+ + compressed
+ )
+ payload = base64.b64encode(block).decode()
+ h = self._make(_syslog_output(payload))
+ assert h.collect(tmp_path, "t") is True
+ assert (tmp_path / "t.core").read_bytes() == plaintext
+
+ def test_collect_fails_when_lzf_block_decompression_fails(
+ self, monkeypatch: pytest.MonkeyPatch, tmp_path: "Path"
+ ) -> None:
+ fake_lzf = MagicMock()
+ fake_lzf.ffi.new.return_value = bytearray(13)
+ fake_lzf.lib.lzf_decompress.return_value = 0
+ monkeypatch.setattr("ntfc.debug.coredump.syslog_handler.lzf", fake_lzf)
+ block = (
+ b"ZV\x01" + struct.pack(">H", 4) + struct.pack(">H", 13) + b"COMP"
+ )
+ payload = base64.b64encode(block).decode()
+ h = self._make(_syslog_output(payload))
+ assert h.collect(tmp_path, "t") is False
+ fake_lzf.lib.lzf_decompress.assert_called_once_with(
+ b"COMP", 4, fake_lzf.ffi.new.return_value, 13
+ )
+
+ def test_collect_fails_without_markers(
+ self, monkeypatch: pytest.MonkeyPatch, tmp_path: "Path"
+ ) -> None:
+ monkeypatch.setattr(
+ "ntfc.debug.coredump.syslog_handler.lzf", MagicMock()
+ )
+ h = self._make("no coredump here\n")
+ assert h.collect(tmp_path, "t") is False
+
+ def test_collect_fails_on_bad_base64(
+ self, monkeypatch: pytest.MonkeyPatch, tmp_path: "Path"
+ ) -> None:
+ monkeypatch.setattr(
+ "ntfc.debug.coredump.syslog_handler.lzf", MagicMock()
+ )
+ h = self._make(_syslog_output("!!!not-base64!!!"))
+ assert h.collect(tmp_path, "t") is False
+
+ def test_collect_fails_on_bad_magic(
+ self, monkeypatch: pytest.MonkeyPatch, tmp_path: "Path"
+ ) -> None:
+ monkeypatch.setattr(
+ "ntfc.debug.coredump.syslog_handler.lzf", MagicMock()
+ )
+ payload = base64.b64encode(b"XX\x00garbage").decode()
+ h = self._make(_syslog_output(payload))
+ assert h.collect(tmp_path, "t") is False
+
+ def test_collect_fails_on_truncated_clen(
+ self, monkeypatch: pytest.MonkeyPatch, tmp_path: "Path"
+ ) -> None:
+ monkeypatch.setattr(
+ "ntfc.debug.coredump.syslog_handler.lzf", MagicMock()
+ )
+ payload = base64.b64encode(b"ZV\x00\x00").decode()
+ h = self._make(_syslog_output(payload))
+ assert h.collect(tmp_path, "t") is False
+
+ def test_collect_fails_on_truncated_ulen(
+ self, monkeypatch: pytest.MonkeyPatch, tmp_path: "Path"
+ ) -> None:
+ monkeypatch.setattr(
+ "ntfc.debug.coredump.syslog_handler.lzf", MagicMock()
+ )
+ block = b"ZV\x01" + struct.pack(">H", 4) + b"\x00"
+ payload = base64.b64encode(block).decode()
+ h = self._make(_syslog_output(payload))
+ assert h.collect(tmp_path, "t") is False
+
+ def test_collect_fails_on_unknown_block_type(
+ self, monkeypatch: pytest.MonkeyPatch, tmp_path: "Path"
+ ) -> None:
+ monkeypatch.setattr(
+ "ntfc.debug.coredump.syslog_handler.lzf", MagicMock()
+ )
+ block = b"ZV\x02" + struct.pack(">H", 0)
+ payload = base64.b64encode(block).decode()
+ h = self._make(_syslog_output(payload))
+ assert h.collect(tmp_path, "t") is False
+
+ def test_collect_handles_payload_line_without_prefix(
+ self, monkeypatch: pytest.MonkeyPatch, tmp_path: "Path"
+ ) -> None:
+ monkeypatch.setattr(
+ "ntfc.debug.coredump.syslog_handler.lzf", MagicMock()
+ )
+ payload = base64.b64encode(_raw_block(b"\x7fELF-core")).decode()
+ output = "Start coredump\n" + payload + "\nFinish coredump\n"
+ h = self._make(output)
+ assert h.collect(tmp_path, "t") is True
+ assert (tmp_path / "t.core").read_bytes() == b"\x7fELF-core"
+
+ def test_collect_uses_last_complete_pair_ignoring_trailing_start(
+ self, monkeypatch: pytest.MonkeyPatch, tmp_path: "Path"
+ ) -> None:
+ monkeypatch.setattr(
+ "ntfc.debug.coredump.syslog_handler.lzf", MagicMock()
+ )
+ payload = base64.b64encode(_raw_block(b"\x7fELF-core")).decode()
+ output = (
+ _syslog_output(payload)
+ + "[ 2.0] Start coredump\n"
+ + "[ 2.1] AAAA\n"
+ )
+ h = self._make(output)
+ assert h.collect(tmp_path, "t") is True
+ assert (tmp_path / "t.core").read_bytes() == b"\x7fELF-core"
+
+ def test_collect_fails_when_only_unterminated_start(
+ self, monkeypatch: pytest.MonkeyPatch, tmp_path: "Path"
+ ) -> None:
+ monkeypatch.setattr(
+ "ntfc.debug.coredump.syslog_handler.lzf", MagicMock()
+ )
+ output = "[ 1.0] Start coredump\n[ 1.1] AAAA\n"
+ h = self._make(output)
+ assert h.collect(tmp_path, "t") is False
diff --git a/tests/debug/test_coredump_manager.py b/tests/debug/test_coredump_manager.py
new file mode 100644
index 0000000..0aba74c
--- /dev/null
+++ b/tests/debug/test_coredump_manager.py
@@ -0,0 +1,263 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+from typing import TYPE_CHECKING
+from unittest.mock import MagicMock
+
+if TYPE_CHECKING:
+ from pathlib import Path
+
+import pytest
+
+from ntfc.debug.config import CoredumpConfig
+from ntfc.debug.coredump.base import CoredumpHandler
+from ntfc.debug.coredump.manager import CoredumpManager
+
+
+def _make_cfg(
+ enable: bool = True,
+ limit: int = 5,
+ collection_type: str = "auto",
+) -> CoredumpConfig:
+ raw: dict = {
+ "enable": enable,
+ "limit": limit,
+ "type": collection_type,
+ }
+ return CoredumpConfig(raw)
+
+
+def _mock_handler(
+ name: str = "mock",
+ priority: int = 10,
+ available: bool = True,
+ collect_result: bool = True,
+) -> CoredumpHandler:
+ handler = MagicMock(spec=CoredumpHandler)
+ handler.name = name
+ handler.priority = priority
+ handler.is_available.return_value = available
+ handler.collect.return_value = collect_result
+ handler.is_enabled.return_value = True
+ return handler
+
+
+class TestCoredumpManagerDisabled:
+ def test_collect_returns_false_when_disabled(self, tmp_path: "Path"):
+ cfg = _make_cfg(enable=False)
+ mgr = CoredumpManager(cfg)
+ h = _mock_handler()
+ mgr.register(h)
+ assert mgr.collect(tmp_path, "test") is False
+
+ def test_count_not_incremented_when_disabled(self, tmp_path: "Path"):
+ cfg = _make_cfg(enable=False)
+ mgr = CoredumpManager(cfg)
+ mgr.collect(tmp_path, "test")
+ assert mgr.collected_count == 0
+
+
+class TestCoredumpManagerLimit:
+ def test_collect_returns_false_when_limit_reached(self, tmp_path: "Path"):
+ cfg = _make_cfg(enable=True, limit=2)
+ mgr = CoredumpManager(cfg)
+ h = _mock_handler(collect_result=True)
+ mgr.register(h)
+
+ assert mgr.collect(tmp_path, "a") is True
+ assert mgr.collect(tmp_path, "b") is True
+ assert mgr.collect(tmp_path, "c") is False
+
+ def test_count_equals_limit_when_limit_reached(self, tmp_path: "Path"):
+ cfg = _make_cfg(enable=True, limit=1)
+ mgr = CoredumpManager(cfg)
+ h = _mock_handler(collect_result=True)
+ mgr.register(h)
+ mgr.collect(tmp_path, "x")
+ assert mgr.collected_count == 1
+
+
+class TestCoredumpManagerFallback:
+ def test_falls_back_to_next_handler_on_failure(self, tmp_path: "Path"):
+ cfg = _make_cfg()
+ mgr = CoredumpManager(cfg)
+ primary = _mock_handler(name="gdb", priority=10, collect_result=False)
+ fallback = _mock_handler(
+ name="fastboot", priority=20, collect_result=True
+ )
+ mgr.register(primary)
+ mgr.register(fallback)
+
+ assert mgr.collect(tmp_path, "test") is True
+ primary.collect.assert_called_once()
+ fallback.collect.assert_called_once()
+ assert mgr.collected_count == 1
+
+ def test_returns_false_when_all_handlers_fail(self, tmp_path: "Path"):
+ cfg = _make_cfg()
+ mgr = CoredumpManager(cfg)
+ a = _mock_handler(name="gdb", priority=10, collect_result=False)
+ b = _mock_handler(name="fastboot", priority=20, collect_result=False)
+ mgr.register(a)
+ mgr.register(b)
+
+ assert mgr.collect(tmp_path, "test") is False
+ a.collect.assert_called_once()
+ b.collect.assert_called_once()
+ assert mgr.collected_count == 0
+
+
+class TestCoredumpManagerPrioritySelection:
+ def test_selects_lower_priority_first(self, tmp_path: "Path"):
+ cfg = _make_cfg()
+ mgr = CoredumpManager(cfg)
+ h_low = _mock_handler(name="alpha", priority=5, collect_result=True)
+ h_high = _mock_handler(name="beta", priority=20, collect_result=True)
+ mgr.register(h_high)
+ mgr.register(h_low)
+ result = mgr.collect(tmp_path, "t")
+ assert result is True
+ h_low.collect.assert_called_once()
+ h_high.collect.assert_not_called()
+
+ def test_skips_unavailable_handler(self, tmp_path: "Path"):
+ cfg = _make_cfg()
+ mgr = CoredumpManager(cfg)
+ unavail = _mock_handler(name="unavail", priority=1, available=False)
+ avail = _mock_handler(
+ name="avail", priority=10, available=True, collect_result=True
+ )
+ mgr.register(unavail)
+ mgr.register(avail)
+ result = mgr.collect(tmp_path, "t")
+ assert result is True
+ avail.collect.assert_called_once()
+ unavail.collect.assert_not_called()
+
+
+class TestCoredumpManagerNameSelection:
+ def test_selects_by_name_when_not_auto(self, tmp_path: "Path"):
+ cfg = _make_cfg(collection_type="gdb")
+ mgr = CoredumpManager(cfg)
+ gdb_h = _mock_handler(name="gdb", priority=10, collect_result=True)
+ other_h = _mock_handler(
+ name="fastboot", priority=5, collect_result=True
+ )
+ mgr.register(gdb_h)
+ mgr.register(other_h)
+ result = mgr.collect(tmp_path, "t")
+ assert result is True
+ gdb_h.collect.assert_called_once()
+ other_h.collect.assert_not_called()
+
+ def test_returns_false_when_named_handler_not_registered(
+ self, tmp_path: "Path"
+ ):
+ cfg = _make_cfg(collection_type="ymodem")
+ mgr = CoredumpManager(cfg)
+ h = _mock_handler(name="gdb", collect_result=True)
+ mgr.register(h)
+ assert mgr.collect(tmp_path, "t") is False
+
+
+class TestCoredumpManagerNoHandler:
+ def test_returns_false_when_no_handler_registered(self, tmp_path: "Path"):
+ cfg = _make_cfg()
+ mgr = CoredumpManager(cfg)
+ assert mgr.collect(tmp_path, "t") is False
+
+ def test_returns_false_when_all_handlers_unavailable(
+ self, tmp_path: "Path"
+ ):
+ cfg = _make_cfg()
+ mgr = CoredumpManager(cfg)
+ h = _mock_handler(available=False)
+ mgr.register(h)
+ assert mgr.collect(tmp_path, "t") is False
+
+
+class TestCoredumpManagerCount:
+ def test_count_incremented_only_on_success(self, tmp_path: "Path"):
+ cfg = _make_cfg()
+ mgr = CoredumpManager(cfg)
+ success_h = _mock_handler(
+ name="good", priority=10, collect_result=True
+ )
+ fail_h = _mock_handler(name="bad", priority=20, collect_result=False)
+ mgr.register(success_h)
+ mgr.collect(tmp_path, "s")
+ assert mgr.collected_count == 1
+
+ mgr2 = CoredumpManager(cfg)
+ mgr2.register(fail_h)
+ mgr2.collect(tmp_path, "f")
+ assert mgr2.collected_count == 0
+
+ def test_reset_clears_count(self, tmp_path: "Path"):
+ cfg = _make_cfg()
+ mgr = CoredumpManager(cfg)
+ h = _mock_handler(collect_result=True)
+ mgr.register(h)
+ mgr.collect(tmp_path, "a")
+ mgr.collect(tmp_path, "b")
+ assert mgr.collected_count == 2
+ mgr.reset()
+ assert mgr.collected_count == 0
+
+
+class TestCoredumpManagerRegister:
+ def test_register_multiple_handlers(self, tmp_path: "Path"):
+ cfg = _make_cfg()
+ mgr = CoredumpManager(cfg)
+ handlers = [
+ _mock_handler(name="h1", priority=30, collect_result=False),
+ _mock_handler(name="h2", priority=20, collect_result=False),
+ _mock_handler(name="h3", priority=10, collect_result=True),
+ ]
+ for h in handlers:
+ mgr.register(h)
+ result = mgr.collect(tmp_path, "t")
+ assert result is True
+ handlers[2].collect.assert_called_once()
+
+ def test_disabled_handler_skipped(self, tmp_path: "Path"):
+ cfg = _make_cfg()
+ mgr = CoredumpManager(cfg)
+ h = MagicMock(spec=CoredumpHandler)
+ h.name = "mock"
+ h.priority = 5
+ h.is_available.return_value = True
+ h.collect.return_value = True
+ h.is_enabled.return_value = False
+ mgr.register(h)
+ assert mgr.collect(tmp_path, "t") is False
+
+ @pytest.mark.parametrize("limit", [0, 1, 10])
+ def test_limit_zero_always_returns_false(
+ self, tmp_path: "Path", limit: int
+ ):
+ cfg = _make_cfg(limit=limit)
+ mgr = CoredumpManager(cfg)
+ h = _mock_handler(collect_result=True)
+ mgr.register(h)
+ for _ in range(limit):
+ mgr.collect(tmp_path, "fill")
+ # now at limit
+ assert mgr.collect(tmp_path, "over") is False
diff --git a/tests/debug/test_gdb_controller.py b/tests/debug/test_gdb_controller.py
new file mode 100644
index 0000000..48eaf2f
--- /dev/null
+++ b/tests/debug/test_gdb_controller.py
@@ -0,0 +1,746 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+import io
+import os
+import threading
+import time
+from typing import TYPE_CHECKING, BinaryIO, Generator, Iterator, List, Tuple
+from unittest.mock import MagicMock, patch
+
+if TYPE_CHECKING:
+ from pathlib import Path
+
+import pytest
+
+from ntfc.debug.config import GdbConfig
+from ntfc.debug.gdb.controller import GdbController
+
+
+def _cfg(target: str = "", **kwargs: object) -> GdbConfig:
+ return GdbConfig({"target": target, **kwargs})
+
+
+def _make_stdout(lines: List[str]) -> io.BytesIO:
+ """Return a BytesIO that yields the given lines when iterated."""
+ content = "".join(line + "\n" for line in lines)
+ buf = io.BytesIO(content.encode())
+ return buf
+
+
+def _make_process(
+ stdout_lines: List[str],
+ returncode: "int | None" = None,
+) -> MagicMock:
+ proc = MagicMock()
+ proc.stdout = _make_stdout(stdout_lines)
+ proc.stdin = MagicMock()
+ proc.returncode = returncode
+ proc.poll.return_value = returncode
+ proc.wait.return_value = 0
+ return proc
+
+
+def _pipe_process() -> Tuple[MagicMock, BinaryIO]:
+ """Create a mock process backed by a real OS pipe.
+
+ Returns ``(proc, write_file)`` where *write_file* is the writable end.
+ The caller must close *write_file* when done.
+ """
+ r_fd, w_fd = os.pipe()
+ r_file = os.fdopen(r_fd, "rb")
+ w_file = os.fdopen(w_fd, "wb", buffering=0)
+ proc: MagicMock = MagicMock()
+ proc.stdout = r_file
+ proc.stdin = MagicMock()
+ proc.returncode = None
+ proc.poll.return_value = None
+ proc.wait.return_value = 0
+ return proc, w_file
+
+
+@pytest.fixture
+def elf(tmp_path: "Path") -> "Path":
+ p = tmp_path / "app.elf"
+ p.write_bytes(b"\x7fELF")
+ return p
+
+
+class TestGdbControllerStart:
+ def test_start_returns_true_when_prompt_appears(self, elf: "Path"):
+ proc = _make_process(["Reading symbols from app.elf...", "(gdb) "])
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg())
+ result = ctrl.start(timeout=5.0)
+ assert result is True
+
+ def test_start_returns_true_on_type_help_prompt(self, elf: "Path"):
+ proc = _make_process(['Type "help" for more information.'])
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg())
+ result = ctrl.start(timeout=5.0)
+ assert result is True
+
+ def test_start_returns_false_when_gdb_missing(self, elf: "Path"):
+ with patch(
+ "subprocess.Popen", side_effect=FileNotFoundError("no gdb")
+ ):
+ ctrl = GdbController(elf, _cfg(gdb_path="missing-gdb"))
+ result = ctrl.start(timeout=5.0)
+ assert result is False
+ assert ctrl.is_running() is False
+
+ def test_start_returns_true_on_ready_marker(self, elf: "Path"):
+ proc = _make_process([GdbController.READY_MARKER])
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg())
+ result = ctrl.start(timeout=5.0)
+ assert result is True
+
+ def test_start_returns_false_on_timeout(self, elf: "Path"):
+ proc = _make_process([]) # no prompt ever
+ block = threading.Event()
+
+ class BlockingStdout:
+ def __iter__(self) -> Iterator[bytes]:
+ block.wait() # block until test ends
+ return iter([])
+
+ proc.stdout = BlockingStdout()
+
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg())
+ result = ctrl.start(timeout=0.05)
+ block.set() # unblock the reader thread
+ assert result is False
+
+ def test_start_returns_false_when_process_exits_immediately(
+ self, elf: "Path"
+ ):
+ proc = _make_process(["(gdb) "], returncode=1)
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg())
+ result = ctrl.start(timeout=5.0)
+ assert result is False
+
+ def test_start_sends_target_remote_when_target_set(self, elf: "Path"):
+ proc = _make_process(
+ [
+ "(gdb) ",
+ "Remote debugging using localhost:1234",
+ GdbController.ATTACH_MARKER,
+ ]
+ )
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg(target="localhost:1234"))
+ result = ctrl.start(timeout=5.0)
+ assert result is True
+ written = b"".join(
+ call.args[0] for call in proc.stdin.write.call_args_list
+ )
+ assert b"target remote localhost:1234" in written
+
+ def test_start_returns_false_when_attach_fails(self, elf: "Path"):
+ # error output instead of "Remote debugging using", then marker
+ proc = _make_process(
+ [
+ "(gdb) ",
+ "localhost:1234: Connection refused.",
+ GdbController.ATTACH_MARKER,
+ ]
+ )
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg(target="localhost:1234"))
+ result = ctrl.start(timeout=5.0)
+ assert result is False
+ proc.terminate.assert_called_once()
+
+ def test_remote_attach_failure_logs_generic_message(
+ self, elf: "Path", caplog: object
+ ) -> None:
+ """Remote attach failure should not mention ptrace_scope."""
+ proc = _make_process(
+ [
+ "(gdb) ",
+ "localhost:1234: Connection refused.",
+ GdbController.ATTACH_MARKER,
+ ]
+ )
+ with patch("subprocess.Popen", return_value=proc):
+ with caplog.at_level("WARNING"): # type: ignore[attr-defined]
+ ctrl = GdbController(elf, _cfg(target="localhost:1234"))
+ result = ctrl.start(timeout=5.0)
+ assert result is False
+ assert "ptrace" not in caplog.text # type: ignore[attr-defined]
+ assert "failed to attach" in caplog.text # type: ignore[attr-defined]
+
+ def test_start_returns_false_on_attach_timeout(self, elf: "Path"):
+ block = threading.Event()
+
+ class BlockAfterReady:
+ def __iter__(self) -> Generator[bytes, None, None]:
+ yield b"(gdb) \n"
+ block.wait() # no attach marker ever arrives
+
+ proc = _make_process([])
+ proc.stdout = BlockAfterReady()
+
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg(target="localhost:1234"))
+ result = ctrl.start(timeout=0.05)
+ block.set()
+ assert result is False
+ proc.terminate.assert_called_once()
+
+ def test_start_does_not_send_target_remote_when_target_empty(
+ self, elf: "Path"
+ ):
+ proc = _make_process(["(gdb) "])
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg(target=""))
+ ctrl.start(timeout=5.0)
+ written = b"".join(
+ call.args[0] for call in proc.stdin.write.call_args_list
+ )
+ assert b"target remote" not in written
+
+
+class TestGdbControllerCommandLine:
+ def test_start_uses_default_gdb_path(self, elf: "Path"):
+ proc = _make_process(["(gdb) "])
+ with patch("subprocess.Popen", return_value=proc) as popen:
+ ctrl = GdbController(elf, _cfg())
+ ctrl.start(timeout=5.0)
+ argv = popen.call_args.args[0]
+ assert argv[0] == "gdb"
+ assert argv[1] == "-q"
+ assert argv[2] == "-nx"
+ assert argv[3] == str(elf)
+ assert "set pagination off" in argv
+ assert "set confirm off" in argv
+ assert f"echo {GdbController.READY_MARKER}\\n" in argv
+
+ def test_start_uses_custom_gdb_path(self, elf: "Path"):
+ proc = _make_process(["(gdb) "])
+ with patch("subprocess.Popen", return_value=proc) as popen:
+ ctrl = GdbController(elf, _cfg(gdb_path="gdb-multiarch"))
+ ctrl.start(timeout=5.0)
+ argv = popen.call_args.args[0]
+ assert argv[0] == "gdb-multiarch"
+
+ def test_start_sources_plugin_when_enabled(self, elf: "Path"):
+ proc = _make_process(["(gdb) "])
+ with patch("subprocess.Popen", return_value=proc) as popen:
+ ctrl = GdbController(elf, _cfg(plugin=True))
+ ctrl.start(timeout=5.0)
+ argv = popen.call_args.args[0]
+ sourced = [a for a in argv if a.startswith("source ")]
+ assert len(sourced) == 1
+ assert sourced[0].endswith("plugin/init.py")
+
+ def test_start_does_not_source_plugin_by_default(self, elf: "Path"):
+ proc = _make_process(["(gdb) "])
+ with patch("subprocess.Popen", return_value=proc) as popen:
+ ctrl = GdbController(elf, _cfg())
+ ctrl.start(timeout=5.0)
+ argv = popen.call_args.args[0]
+ assert not any(a.startswith("source ") for a in argv)
+
+ def test_start_sources_nx_plugin_after_ntfc_plugin(self, elf: "Path"):
+ proc = _make_process(["(gdb) "])
+ with patch("subprocess.Popen", return_value=proc) as popen:
+ ctrl = GdbController(
+ elf, _cfg(plugin=True, nx_plugin="/nx/gdbinit.py")
+ )
+ ctrl.start(timeout=5.0)
+ argv = popen.call_args.args[0]
+ sourced = [a for a in argv if a.startswith("source ")]
+ assert len(sourced) == 2
+ assert sourced[0].endswith("plugin/init.py")
+ assert sourced[1] == "source /nx/gdbinit.py"
+
+ def test_start_sets_osabi_when_configured(self, elf: "Path"):
+ proc = _make_process(["(gdb) "])
+ with patch("subprocess.Popen", return_value=proc) as popen:
+ ctrl = GdbController(elf, _cfg(osabi="none"))
+ ctrl.start(timeout=5.0)
+ assert "set osabi none" in popen.call_args.args[0]
+
+ def test_start_no_osabi_by_default(self, elf: "Path"):
+ proc = _make_process(["(gdb) "])
+ with patch("subprocess.Popen", return_value=proc) as popen:
+ ctrl = GdbController(elf, _cfg())
+ ctrl.start(timeout=5.0)
+ assert not any(
+ a.startswith("set osabi") for a in popen.call_args.args[0]
+ )
+
+
+class TestGdbControllerAttach:
+ def _attach_cfg(self, **kw: object) -> GdbConfig:
+ return GdbConfig({"attach": True, **kw})
+
+ def test_attach_success(self, elf: "Path"):
+ proc = _make_process(
+ [
+ "(gdb) ",
+ "Attaching to program: /x/nuttx, process 4242",
+ GdbController.ATTACH_MARKER,
+ ]
+ )
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(
+ elf, self._attach_cfg(), pid_provider=lambda: 4242
+ )
+ assert ctrl.start(timeout=5.0) is True
+ written = b"".join(c.args[0] for c in proc.stdin.write.call_args_list)
+ assert b"attach 4242\n" in written
+
+ def test_attach_fails_without_pid(self, elf: "Path"):
+ proc = _make_process(["(gdb) "])
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(
+ elf, self._attach_cfg(), pid_provider=lambda: None
+ )
+ assert ctrl.start(timeout=5.0) is False
+ proc.terminate.assert_called_once()
+
+ def test_attach_fails_without_provider(self, elf: "Path"):
+ proc = _make_process(["(gdb) "])
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, self._attach_cfg())
+ assert ctrl.start(timeout=5.0) is False
+
+ def test_attach_denied_reports_failure(self, elf: "Path"):
+ # ptrace denied: error text instead of "Attaching to program"
+ proc = _make_process(
+ [
+ "(gdb) ",
+ "ptrace: Operation not permitted.",
+ GdbController.ATTACH_MARKER,
+ ]
+ )
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(
+ elf, self._attach_cfg(), pid_provider=lambda: 4242
+ )
+ assert ctrl.start(timeout=5.0) is False
+ proc.terminate.assert_called_once()
+
+ def test_attach_denied_logs_ptrace_guidance(
+ self, elf: "Path", caplog: object
+ ) -> None:
+ """PID attach denied should mention ptrace_scope guidance."""
+ proc = _make_process(
+ [
+ "(gdb) ",
+ "ptrace: Operation not permitted.",
+ GdbController.ATTACH_MARKER,
+ ]
+ )
+ with patch("subprocess.Popen", return_value=proc):
+ with caplog.at_level("WARNING"): # type: ignore[attr-defined]
+ ctrl = GdbController(
+ elf, self._attach_cfg(), pid_provider=lambda: 4242
+ )
+ result = ctrl.start(timeout=5.0)
+ assert result is False
+ assert "ptrace_scope" in caplog.text # type: ignore[attr-defined]
+
+ def test_attach_use_sudo_prefixes_argv(self, elf: "Path"):
+ proc = _make_process(["(gdb) "])
+ with patch("subprocess.Popen", return_value=proc) as popen:
+ ctrl = GdbController(
+ elf,
+ self._attach_cfg(use_sudo=True),
+ pid_provider=lambda: 1,
+ )
+ ctrl.start(timeout=5.0)
+ argv = popen.call_args.args[0]
+ assert argv[0] == "sudo"
+ assert argv[1] == "gdb"
+
+
+class TestGdbControllerSetup:
+ def _written(self, proc: MagicMock) -> bytes:
+ return b"".join(
+ call.args[0] for call in proc.stdin.write.call_args_list
+ )
+
+ def test_setup_sends_outdir_when_plugin_enabled(self, elf: "Path"):
+ proc = _make_process(["(gdb) "])
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg(plugin=True))
+ ctrl.start(timeout=5.0)
+ ctrl.setup("/tmp/results")
+ assert b"ntfcsetoutdir /tmp/results\n" in self._written(proc)
+
+ def test_setup_skips_outdir_without_plugin(self, elf: "Path"):
+ proc = _make_process(["(gdb) "])
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg())
+ ctrl.start(timeout=5.0)
+ ctrl.setup("/tmp/results")
+ assert b"ntfcsetoutdir" not in self._written(proc)
+
+ def test_setup_sends_setup_cmds(self, elf: "Path"):
+ cmds = ["break main", "continue"]
+ proc = _make_process(["(gdb) "])
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg(setup_cmds=cmds))
+ ctrl.start(timeout=5.0)
+ ctrl.setup("/tmp/results")
+ written = self._written(proc)
+ assert b"break main\n" in written
+ assert b"continue\n" in written
+
+ def test_setup_plants_auto_breakpoints(self, elf: "Path"):
+ proc = _make_process(["(gdb) "])
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg(plugin=True, auto_breakpoints=True))
+ ctrl.start(timeout=5.0)
+ ctrl.setup("/results")
+ written = self._written(proc).decode()
+ for spec in GdbController.CRASH_BP_SPECS:
+ assert (
+ f"ntfcautobp bp {spec} ntfcautogcore -d $RESULT_DIR"
+ f" -n crash gcore;bt\n" in written
+ )
+ assert (
+ "ntfcautobp bp reboot_notifier_call_chain"
+ " ntfcautopoweroffcheck\n" in written
+ )
+ assert written.endswith("continue\n")
+
+ def test_setup_auto_breakpoints_with_mmleak(self, elf: "Path"):
+ proc = _make_process(["(gdb) "])
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(
+ elf,
+ _cfg(plugin=True, auto_breakpoints=True, mmleak=True),
+ )
+ ctrl.start(timeout=5.0)
+ ctrl.setup("/results")
+ assert (
+ "ntfcautobp bp reboot_notifier_call_chain"
+ " ntfcmmleak -d $RESULT_DIR gcore;ntfcautopoweroffcheck\n"
+ in self._written(proc).decode()
+ )
+
+ def test_setup_no_auto_breakpoints_by_default(self, elf: "Path"):
+ proc = _make_process(["(gdb) "])
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg(plugin=True))
+ ctrl.start(timeout=5.0)
+ ctrl.setup("/results")
+ assert b"ntfcautobp" not in self._written(proc)
+
+
+class TestGdbControllerGenerateCoredump:
+ def test_generate_coredump_returns_path_on_success(
+ self, elf: "Path", tmp_path: "Path"
+ ) -> None:
+ """Use a real pipe so the reader blocks until we feed the response."""
+ proc, w_file = _pipe_process()
+ corefile = tmp_path / "test.core"
+
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg())
+ w_file.write(b"(gdb) \n")
+ start_ok = ctrl.start(timeout=5.0)
+
+ assert start_ok is True
+
+ w_file.write(f"Saved corefile {corefile}\n".encode())
+ w_file.write(f"{GdbController.GCORE_MARKER}\n".encode())
+ result = ctrl.generate_coredump(tmp_path, "test", timeout=5.0)
+ # Close write end → EOF → reader exits; join to avoid ResourceWarning
+ w_file.close()
+ ctrl.stop()
+ proc.stdout.close()
+
+ assert result == corefile
+
+ def test_generate_coredump_returns_none_when_gcore_fails(
+ self, elf: "Path", tmp_path: "Path"
+ ) -> None:
+ """Marker without 'Saved corefile' signals a failed gcore."""
+ proc, w_file = _pipe_process()
+
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg())
+ w_file.write(b"(gdb) \n")
+ ctrl.start(timeout=5.0)
+
+ w_file.write(b"Unable to fetch a corefile\n")
+ w_file.write(f"{GdbController.GCORE_MARKER}\n".encode())
+ result = ctrl.generate_coredump(tmp_path, "test", timeout=5.0)
+ w_file.close()
+ ctrl.stop()
+ proc.stdout.close()
+
+ assert result is None
+
+ def test_generate_coredump_returns_none_on_timeout(
+ self, elf: "Path", tmp_path: "Path"
+ ) -> None:
+ block = threading.Event()
+
+ class BlockAfterPrompt:
+ def __iter__(self) -> Generator[bytes, None, None]:
+ yield b"(gdb) \n"
+ block.wait() # block: no "Saved corefile" ever arrives
+
+ proc = _make_process([])
+ proc.stdout = BlockAfterPrompt()
+
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg())
+ ctrl.start(timeout=5.0)
+ result = ctrl.generate_coredump(tmp_path, "test", timeout=0.05)
+ block.set()
+ assert result is None
+
+ def test_generate_coredump_returns_none_when_process_dies(
+ self, elf: "Path", tmp_path: "Path"
+ ) -> None:
+ """Closing the write end mid-wait simulates process death."""
+ proc, w_file = _pipe_process()
+
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg())
+ w_file.write(b"(gdb) \n")
+ ctrl.start(timeout=5.0)
+
+ def close_after_delay() -> None:
+ import time
+
+ time.sleep(0.05)
+ w_file.close()
+
+ t = threading.Thread(target=close_after_delay)
+ t.start()
+ result = ctrl.generate_coredump(tmp_path, "test", timeout=5.0)
+ t.join()
+ # Reader exited due to EOF; join + close to avoid ResourceWarning
+ ctrl.stop()
+ proc.stdout.close()
+
+ assert result is None
+
+ def test_generate_coredump_uses_gcore_cmd(
+ self, elf: "Path", tmp_path: "Path"
+ ) -> None:
+ proc, w_file = _pipe_process()
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg(gcore_cmd="gcore -t nuttx"))
+ w_file.write(b"(gdb) \n")
+ ctrl.start(timeout=5.0)
+ w_file.write(f"{GdbController.GCORE_MARKER}\n".encode())
+ ctrl.generate_coredump(tmp_path, "t", timeout=5.0)
+ w_file.close()
+ ctrl.stop()
+ proc.stdout.close()
+ written = b"".join(
+ call.args[0] for call in proc.stdin.write.call_args_list
+ )
+ assert f"gcore -t nuttx {tmp_path}/t.core\n".encode() in written
+
+
+class TestGdbControllerIsRunning:
+ def test_is_running_true_when_process_alive(self, elf: "Path"):
+ proc = _make_process(["(gdb) "])
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg())
+ ctrl.start(timeout=5.0)
+ assert ctrl.is_running() is True
+
+ def test_is_running_false_before_open(self, elf: "Path"):
+ ctrl = GdbController(elf, _cfg())
+ assert ctrl.is_running() is False
+
+ def test_is_running_false_after_process_exits(self, elf: "Path"):
+ proc = _make_process(["(gdb) "])
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg())
+ ctrl.start(timeout=5.0)
+ # Simulate process exit after startup: only poll() notices
+ proc.poll.return_value = 0
+ assert ctrl.is_running() is False
+
+
+class TestGdbControllerStop:
+ def test_stop_sends_quit(self, elf: "Path"):
+ proc = _make_process(["(gdb) "])
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg())
+ ctrl.start(timeout=5.0)
+ ctrl.stop()
+ written = b"".join(
+ call.args[0] for call in proc.stdin.write.call_args_list
+ )
+ assert b"quit" in written
+
+ def test_stop_terminates_if_wait_times_out(self, elf: "Path"):
+ import subprocess as sp
+
+ proc = _make_process(["(gdb) "])
+ # quit wait times out, post-terminate wait reaps the process
+ proc.wait.side_effect = [
+ sp.TimeoutExpired(cmd="gdb", timeout=5),
+ 0,
+ ]
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg())
+ ctrl.start(timeout=5.0)
+ ctrl.stop()
+ proc.terminate.assert_called_once()
+ proc.kill.assert_not_called()
+
+ def test_stop_kills_if_terminate_wait_times_out(self, elf: "Path"):
+ import subprocess as sp
+
+ proc = _make_process(["(gdb) "])
+ # quit wait and post-terminate wait time out, kill reaps
+ proc.wait.side_effect = [
+ sp.TimeoutExpired(cmd="gdb", timeout=5),
+ sp.TimeoutExpired(cmd="gdb", timeout=5),
+ 0,
+ ]
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg())
+ ctrl.start(timeout=5.0)
+ ctrl.stop()
+ proc.terminate.assert_called_once()
+ proc.kill.assert_called_once()
+
+ def test_stop_before_start_does_not_raise(self, elf: "Path"):
+ ctrl = GdbController(elf, _cfg())
+ ctrl.stop() # must not raise
+
+ def test_send_swallows_oserror(self, elf: "Path"):
+ """OSError on stdin.write must be caught and logged, not raised."""
+ proc = _make_process(["(gdb) "])
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg())
+ ctrl.start(timeout=5.0)
+ # Simulate broken pipe after open
+ proc.stdin.write.side_effect = OSError("broken pipe")
+ ctrl.stop() # calls _send("quit\n") → OSError caught internally
+
+
+class TestGdbControllerCorefileLedger:
+ def test_unsolicited_saved_corefile_recorded(
+ self, elf: "Path", tmp_path: "Path"
+ ) -> None:
+ proc, w_file = _pipe_process()
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg())
+ w_file.write(b"(gdb) \n")
+ ctrl.start(timeout=5.0)
+ core = tmp_path / "crash.2026.core"
+ w_file.write(f"Saved corefile {core}\n".encode())
+ result = ctrl.wait_corefile(timeout=5.0)
+ w_file.close()
+ ctrl.stop()
+ proc.stdout.close()
+ assert result == core
+ assert ctrl.saved_corefiles == [core]
+
+ def test_wait_corefile_times_out(self, elf: "Path") -> None:
+ proc, w_file = _pipe_process()
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg())
+ w_file.write(b"(gdb) \n")
+ ctrl.start(timeout=5.0)
+ result = ctrl.wait_corefile(timeout=0.05)
+ w_file.close()
+ ctrl.stop()
+ proc.stdout.close()
+ assert result is None
+
+ def test_wait_corefile_ignores_earlier_cores(
+ self, elf: "Path", tmp_path: "Path"
+ ) -> None:
+ proc, w_file = _pipe_process()
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg())
+ w_file.write(b"(gdb) \n")
+ ctrl.start(timeout=5.0)
+ w_file.write(f"Saved corefile {tmp_path}/old.core\n".encode())
+ ctrl.wait_corefile(timeout=5.0) # consume the first
+ result = ctrl.wait_corefile(timeout=0.05) # nothing new
+ w_file.close()
+ ctrl.stop()
+ proc.stdout.close()
+ assert result is None
+ assert len(ctrl.saved_corefiles) == 1
+
+ def test_wait_corefile_returns_corefile_recorded_before_call(
+ self, elf: "Path", tmp_path: "Path"
+ ) -> None:
+ """Regression test for the baseline race: a corefile already
+ recorded (e.g. the in-GDB auto-dump firing within ~1s of a
+ crash) before :meth:`~GdbController.wait_corefile` is even
+ called must still be returned, not lost to a stale
+ "wait for growth beyond baseline" snapshot.
+ """
+ proc, w_file = _pipe_process()
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg())
+ w_file.write(b"(gdb) \n")
+ ctrl.start(timeout=5.0)
+ core = tmp_path / "crash.2026.core"
+ w_file.write(f"Saved corefile {core}\n".encode())
+ # Deterministically wait until the reader thread has recorded
+ # the corefile BEFORE calling wait_corefile, so the test does
+ # not depend on thread-scheduling luck to reproduce the race.
+ deadline = time.monotonic() + 5.0
+ while not ctrl.saved_corefiles and time.monotonic() < deadline:
+ time.sleep(0.01)
+ assert ctrl.saved_corefiles == [core]
+ result = ctrl.wait_corefile(timeout=1.0)
+ w_file.close()
+ ctrl.stop()
+ proc.stdout.close()
+ assert result == core
+
+ def test_generate_coredump_result_not_returned_by_later_wait_corefile(
+ self, elf: "Path", tmp_path: "Path"
+ ) -> None:
+ """A corefile already handed out via generate_coredump (the
+ pull path) must never be re-handed-out by a later
+ wait_corefile call for an unrelated crash-aware harvest.
+ """
+ proc, w_file = _pipe_process()
+ corefile = tmp_path / "pull.core"
+ with patch("subprocess.Popen", return_value=proc):
+ ctrl = GdbController(elf, _cfg())
+ w_file.write(b"(gdb) \n")
+ ctrl.start(timeout=5.0)
+ w_file.write(f"Saved corefile {corefile}\n".encode())
+ w_file.write(f"{GdbController.GCORE_MARKER}\n".encode())
+ result = ctrl.generate_coredump(tmp_path, "pull", timeout=5.0)
+ assert result == corefile
+ later = ctrl.wait_corefile(timeout=0.05)
+ w_file.close()
+ ctrl.stop()
+ proc.stdout.close()
+ assert later is None
diff --git a/tests/debug/test_gdb_plugin.py b/tests/debug/test_gdb_plugin.py
new file mode 100644
index 0000000..cf4de31
--- /dev/null
+++ b/tests/debug/test_gdb_plugin.py
@@ -0,0 +1,696 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+"""Tests for the NTFC in-GDB plugin scripts (with a fake gdb module)."""
+
+import builtins
+import importlib
+import sys
+from typing import TYPE_CHECKING, Any
+
+import pytest
+
+if TYPE_CHECKING:
+ from pathlib import Path
+ from types import ModuleType
+
+
+@pytest.fixture
+def autobp(
+ fake_gdb: "ModuleType",
+ monkeypatch: pytest.MonkeyPatch,
+ tmp_path: "Path",
+) -> "ModuleType":
+ """Import a fresh autobp module against the fake gdb module."""
+ mod = importlib.import_module("ntfc.debug.gdb.plugin.autobp")
+ monkeypatch.setattr(mod, "NTFC_DIR_FILE", str(tmp_path / ".ntfc_dir"))
+ return mod
+
+
+def _command(fake_gdb: "ModuleType", name: str) -> Any:
+ """Return the registered command instance by name."""
+ return dict(fake_gdb.Command.registered)[name]
+
+
+class TestResultCheck:
+ def test_success_printed(
+ self, autobp: "ModuleType", capsys: pytest.CaptureFixture[str]
+ ) -> None:
+ wrapped = autobp._result_check(lambda self, a, tty: True)
+ assert wrapped(object(), "arg", False) is True
+ assert "Success: object arg" in capsys.readouterr().out
+
+ def test_failure_printed(
+ self, autobp: "ModuleType", capsys: pytest.CaptureFixture[str]
+ ) -> None:
+ wrapped = autobp._result_check(lambda self, a, tty: False)
+ assert wrapped(object(), "arg", False) is False
+ assert "Failed: object arg" in capsys.readouterr().out
+
+ def test_no_args_formatting(
+ self, autobp: "ModuleType", capsys: pytest.CaptureFixture[str]
+ ) -> None:
+ wrapped = autobp._result_check(lambda self: True)
+ assert wrapped(object()) is True
+ assert "Success: object" in capsys.readouterr().out
+
+
+class TestOutputDirHelpers:
+ def test_get_from_global(
+ self, autobp: "ModuleType", monkeypatch: pytest.MonkeyPatch
+ ) -> None:
+ monkeypatch.setattr(autobp, "_output_dir", "/some/dir")
+ assert autobp._output_dir_get() == "/some/dir"
+
+ def test_get_from_backup_file(
+ self, autobp: "ModuleType", tmp_path: "Path"
+ ) -> None:
+ backup = tmp_path / ".ntfc_dir"
+ backup.write_text("/from/file")
+ assert autobp._output_dir_get() == "/from/file"
+
+ def test_get_from_empty_backup_file(
+ self, autobp: "ModuleType", tmp_path: "Path"
+ ) -> None:
+ backup = tmp_path / ".ntfc_dir"
+ backup.write_text("")
+ assert autobp._output_dir_get() is None
+
+ def test_get_backup_file_read_error(
+ self,
+ autobp: "ModuleType",
+ monkeypatch: pytest.MonkeyPatch,
+ tmp_path: "Path",
+ capsys: pytest.CaptureFixture[str],
+ ) -> None:
+ # A directory triggers OSError on open()
+ (tmp_path / ".ntfc_dir").mkdir()
+ assert autobp._output_dir_get() is None
+ assert "Could not read backup file" in capsys.readouterr().out
+
+ def test_get_unset(self, autobp: "ModuleType") -> None:
+ assert autobp._output_dir_get() is None
+
+ def test_ensure_existing_dir(
+ self, autobp: "ModuleType", tmp_path: "Path"
+ ) -> None:
+ assert autobp._output_dir_ensure(str(tmp_path)) is True
+
+ def test_ensure_creates_dir(
+ self, autobp: "ModuleType", tmp_path: "Path"
+ ) -> None:
+ target = tmp_path / "new" / "dir"
+ assert autobp._output_dir_ensure(str(target)) is True
+ assert target.is_dir()
+
+ def test_ensure_failure(
+ self, autobp: "ModuleType", tmp_path: "Path"
+ ) -> None:
+ blocker = tmp_path / "file"
+ blocker.write_text("x")
+ assert autobp._output_dir_ensure(str(blocker / "sub")) is False
+
+
+class TestAutoBp:
+ def test_breakpoint_type(self, autobp: "ModuleType") -> None:
+ bp = autobp.AutoBp("main", ["bt"], bp_type="bp")
+ assert bp.bp_type == 0
+
+ def test_watchpoint_type(
+ self, autobp: "ModuleType", fake_gdb: "ModuleType"
+ ) -> None:
+ bp = autobp.AutoBp("var", ["bt"], bp_type="wp")
+ assert bp.bp_type == fake_gdb.BP_WATCHPOINT
+
+ def test_crash_coredump_registration(self, autobp: "ModuleType") -> None:
+ bp = autobp.AutoBp("dump_mini_info", ["ntfcautogcore -d /x gcore"])
+ assert (
+ autobp.AutoBp._crash_coredump_breakpoints["dump_mini_info"] is bp
+ )
+
+ def test_no_registration_without_autogcore(
+ self, autobp: "ModuleType"
+ ) -> None:
+ autobp.AutoBp("dump_mini_info", ["bt"])
+ assert autobp.AutoBp._crash_coredump_breakpoints == {}
+
+ def test_no_registration_for_other_spec(
+ self, autobp: "ModuleType"
+ ) -> None:
+ autobp.AutoBp("main", ["ntfcautogcore -d /x gcore"])
+ assert autobp.AutoBp._crash_coredump_breakpoints == {}
+
+ def test_stop_executes_commands(
+ self, autobp: "ModuleType", fake_gdb: "ModuleType"
+ ) -> None:
+ bp = autobp.AutoBp("main", ["bt", "info registers"])
+ assert bp.stop() is True
+ assert "bt" in fake_gdb.executed
+ assert "info registers" in fake_gdb.executed
+ assert bp in autobp.AutoBp._pending_delete
+
+ def test_stop_swallows_gdb_error(
+ self,
+ autobp: "ModuleType",
+ fake_gdb: "ModuleType",
+ monkeypatch: pytest.MonkeyPatch,
+ capsys: pytest.CaptureFixture[str],
+ ) -> None:
+ def fail(cmd: str, to_string: bool = False) -> str:
+ raise fake_gdb.error("boom")
+
+ monkeypatch.setattr(fake_gdb, "execute", fail)
+ bp = autobp.AutoBp("main", ["bt"])
+ assert bp.stop() is True
+ assert "Command failed" in capsys.readouterr().out
+
+ def test_stop_deletes_crash_siblings(
+ self, autobp: "ModuleType", capsys: pytest.CaptureFixture[str]
+ ) -> None:
+ gcore = ["ntfcautogcore -d /x gcore"]
+ bp_a = autobp.AutoBp("dump_mini_info", gcore)
+ bp_b = autobp.AutoBp("dump_core_info", gcore)
+ assert bp_a.stop() is True
+ assert bp_a in autobp.AutoBp._pending_delete
+ assert bp_b in autobp.AutoBp._pending_delete
+ assert autobp.AutoBp._crash_coredump_breakpoints == {}
+ assert "deleting sibling" in capsys.readouterr().out.lower()
+
+
+class TestNtfcAutoBpCommand:
+ def test_registered(self, autobp: "ModuleType", fake_gdb: "ModuleType"):
+ assert "ntfcautobp" in dict(fake_gdb.Command.registered)
+
+ def test_too_few_args(
+ self, autobp: "ModuleType", fake_gdb: "ModuleType"
+ ) -> None:
+ cmd = _command(fake_gdb, "ntfcautobp")
+ assert cmd.invoke("bp main", False) is False
+
+ def test_invalid_bp_type(
+ self, autobp: "ModuleType", fake_gdb: "ModuleType"
+ ) -> None:
+ cmd = _command(fake_gdb, "ntfcautobp")
+ assert cmd.invoke("xx main bt", False) is False
+
+ def test_valid_breakpoint(
+ self, autobp: "ModuleType", fake_gdb: "ModuleType"
+ ) -> None:
+ cmd = _command(fake_gdb, "ntfcautobp")
+ assert cmd.invoke("bp main bt;continue", False) is True
+ bp = fake_gdb.Breakpoint.instances[-1]
+ assert bp.spec == "main"
+ assert bp._cmds == ["bt", "continue"]
+
+ def test_valid_watchpoint(
+ self, autobp: "ModuleType", fake_gdb: "ModuleType"
+ ) -> None:
+ cmd = _command(fake_gdb, "ntfcautobp")
+ assert cmd.invoke("wp counter bt", False) is True
+ bp = fake_gdb.Breakpoint.instances[-1]
+ assert bp.bp_type == fake_gdb.BP_WATCHPOINT
+
+ def test_invalid_breakpoint(
+ self,
+ autobp: "ModuleType",
+ fake_gdb: "ModuleType",
+ monkeypatch: pytest.MonkeyPatch,
+ ) -> None:
+ monkeypatch.setattr(fake_gdb.Breakpoint, "valid", False)
+ cmd = _command(fake_gdb, "ntfcautobp")
+ assert cmd.invoke("bp main bt", False) is False
+
+
+class TestNtfcSetOutDirCommand:
+ def test_usage_on_wrong_argc(
+ self,
+ autobp: "ModuleType",
+ fake_gdb: "ModuleType",
+ capsys: pytest.CaptureFixture[str],
+ ) -> None:
+ cmd = _command(fake_gdb, "ntfcsetoutdir")
+ cmd.invoke("a b", False)
+ assert "Usage:" in capsys.readouterr().out
+
+ def test_sets_dir_and_backup_file(
+ self,
+ autobp: "ModuleType",
+ fake_gdb: "ModuleType",
+ tmp_path: "Path",
+ ) -> None:
+ out = tmp_path / "results"
+ cmd = _command(fake_gdb, "ntfcsetoutdir")
+ cmd.invoke(str(out), False)
+ assert autobp._output_dir == str(out)
+ assert out.is_dir()
+ backup = tmp_path / ".ntfc_dir"
+ assert backup.read_text() == str(out)
+
+ def test_dir_create_failure(
+ self,
+ autobp: "ModuleType",
+ fake_gdb: "ModuleType",
+ tmp_path: "Path",
+ ) -> None:
+ blocker = tmp_path / "file"
+ blocker.write_text("x")
+ cmd = _command(fake_gdb, "ntfcsetoutdir")
+ cmd.invoke(str(blocker / "sub"), False)
+ assert autobp._output_dir is None
+
+ def test_backup_file_write_failure(
+ self,
+ autobp: "ModuleType",
+ fake_gdb: "ModuleType",
+ monkeypatch: pytest.MonkeyPatch,
+ tmp_path: "Path",
+ capsys: pytest.CaptureFixture[str],
+ ) -> None:
+ monkeypatch.setattr(
+ autobp, "NTFC_DIR_FILE", str(tmp_path / "no" / "dir" / "f")
+ )
+ cmd = _command(fake_gdb, "ntfcsetoutdir")
+ cmd.invoke(str(tmp_path), False)
+ assert autobp._output_dir == str(tmp_path)
+ assert "Could not write to backup file" in capsys.readouterr().out
+
+
+class TestNtfcAutoGcoreCommand:
+ def _gcore_creates(
+ self,
+ fake_gdb: "ModuleType",
+ monkeypatch: pytest.MonkeyPatch,
+ content: bytes = b"\x7fELF-core",
+ ) -> None:
+ """Make fake gcore commands create the target file.
+
+ Only ``gcore <path>`` commands reach the fake in these tests, so
+ the last token is always the corefile path.
+ """
+ import pathlib
+
+ def execute(cmd: str, to_string: bool = False) -> str:
+ fake_gdb.executed.append(cmd)
+ pathlib.Path(cmd.split()[-1]).write_bytes(content)
+ return ""
+
+ monkeypatch.setattr(fake_gdb, "execute", execute)
+
+ def test_no_args(self, autobp: "ModuleType", fake_gdb: "ModuleType"):
+ cmd = _command(fake_gdb, "ntfcautogcore")
+ assert cmd.invoke("", False) is False
+
+ def test_wrong_first_flag(
+ self, autobp: "ModuleType", fake_gdb: "ModuleType"
+ ) -> None:
+ cmd = _command(fake_gdb, "ntfcautogcore")
+ assert cmd.invoke("-x /tmp gcore", False) is False
+
+ def test_dynamic_dir_not_set(
+ self, autobp: "ModuleType", fake_gdb: "ModuleType"
+ ) -> None:
+ cmd = _command(fake_gdb, "ntfcautogcore")
+ assert cmd.invoke("-d $RESULT_DIR gcore", False) is False
+
+ def test_dynamic_dir_from_backup_file(
+ self,
+ autobp: "ModuleType",
+ fake_gdb: "ModuleType",
+ monkeypatch: pytest.MonkeyPatch,
+ tmp_path: "Path",
+ ) -> None:
+ (tmp_path / ".ntfc_dir").write_text(str(tmp_path))
+ self._gcore_creates(fake_gdb, monkeypatch)
+ cmd = _command(fake_gdb, "ntfcautogcore")
+ assert cmd.invoke("-d $RESULT_DIR gcore", False) is True
+ assert list(tmp_path.glob("*.core"))
+
+ def test_name_flag_missing_value(
+ self, autobp: "ModuleType", fake_gdb: "ModuleType", tmp_path: "Path"
+ ) -> None:
+ cmd = _command(fake_gdb, "ntfcautogcore")
+ assert cmd.invoke(f"-d {tmp_path} -n", False) is False
+
+ def test_name_flag_used_in_filename(
+ self,
+ autobp: "ModuleType",
+ fake_gdb: "ModuleType",
+ monkeypatch: pytest.MonkeyPatch,
+ tmp_path: "Path",
+ ) -> None:
+ self._gcore_creates(fake_gdb, monkeypatch)
+ cmd = _command(fake_gdb, "ntfcautogcore")
+ assert cmd.invoke(f"-d {tmp_path} -n myname gcore", False) is True
+ cores = list(tmp_path.glob("myname.*.core"))
+ assert len(cores) == 1
+
+ def test_dir_create_failure(
+ self, autobp: "ModuleType", fake_gdb: "ModuleType", tmp_path: "Path"
+ ) -> None:
+ blocker = tmp_path / "file"
+ blocker.write_text("x")
+ cmd = _command(fake_gdb, "ntfcautogcore")
+ assert cmd.invoke(f"-d {blocker / 'sub'} gcore", False) is False
+
+ def test_gcore_command_fails(
+ self,
+ autobp: "ModuleType",
+ fake_gdb: "ModuleType",
+ monkeypatch: pytest.MonkeyPatch,
+ tmp_path: "Path",
+ ) -> None:
+ def fail(cmd: str, to_string: bool = False) -> str:
+ raise fake_gdb.error("no such command")
+
+ monkeypatch.setattr(fake_gdb, "execute", fail)
+ cmd = _command(fake_gdb, "ntfcautogcore")
+ assert cmd.invoke(f"-d {tmp_path} gcore", False) is False
+
+ def test_corefile_not_created(
+ self, autobp: "ModuleType", fake_gdb: "ModuleType", tmp_path: "Path"
+ ) -> None:
+ # default fake execute does not create any file
+ cmd = _command(fake_gdb, "ntfcautogcore")
+ assert cmd.invoke(f"-d {tmp_path} gcore", False) is False
+
+ def test_corefile_invalid_header(
+ self,
+ autobp: "ModuleType",
+ fake_gdb: "ModuleType",
+ monkeypatch: pytest.MonkeyPatch,
+ tmp_path: "Path",
+ ) -> None:
+ self._gcore_creates(fake_gdb, monkeypatch, content=b"XXXX")
+ cmd = _command(fake_gdb, "ntfcautogcore")
+ assert cmd.invoke(f"-d {tmp_path} gcore", False) is False
+
+ def test_corefile_read_error(
+ self,
+ autobp: "ModuleType",
+ fake_gdb: "ModuleType",
+ monkeypatch: pytest.MonkeyPatch,
+ tmp_path: "Path",
+ ) -> None:
+ corefile = tmp_path / "x.core"
+ corefile.write_bytes(b"\x7fELF")
+ real_open = builtins.open
+
+ def bad_open(file: Any, *args: Any, **kwargs: Any) -> Any:
+ if str(file) == str(corefile):
+ raise OSError("boom")
+ return real_open(file, *args, **kwargs)
+
+ monkeypatch.setattr(builtins, "open", bad_open)
+ cmd = _command(fake_gdb, "ntfcautogcore")
+ assert cmd._verify_coredump(str(corefile)) is False
+ # exercise the pass-through branch of bad_open
+ with bad_open(tmp_path / "other.txt", "w") as f:
+ f.write("ok")
+
+
+class TestNtfcAutoMmleakCommand:
+ def test_no_args(self, autobp: "ModuleType", fake_gdb: "ModuleType"):
+ cmd = _command(fake_gdb, "ntfcmmleak")
+ assert cmd.invoke("", False) is False
+
+ def test_wrong_first_flag(
+ self, autobp: "ModuleType", fake_gdb: "ModuleType"
+ ) -> None:
+ cmd = _command(fake_gdb, "ntfcmmleak")
+ assert cmd.invoke("-x /tmp gcore", False) is False
+
+ def test_dynamic_dir_not_set(
+ self, autobp: "ModuleType", fake_gdb: "ModuleType"
+ ) -> None:
+ cmd = _command(fake_gdb, "ntfcmmleak")
+ assert cmd.invoke("-d $RESULT_DIR gcore", False) is False
+
+ def test_dynamic_dir_from_global(
+ self,
+ autobp: "ModuleType",
+ fake_gdb: "ModuleType",
+ monkeypatch: pytest.MonkeyPatch,
+ tmp_path: "Path",
+ ) -> None:
+ monkeypatch.setattr(autobp, "_output_dir", str(tmp_path))
+ cmd = _command(fake_gdb, "ntfcmmleak")
+ assert cmd.invoke("-d $RESULT_DIR gcore", False) is True
+
+ def test_no_leak_no_dump(
+ self, autobp: "ModuleType", fake_gdb: "ModuleType", tmp_path: "Path"
+ ) -> None:
+ cmd = _command(fake_gdb, "ntfcmmleak")
+ assert cmd.invoke(f"-d {tmp_path} gcore", False) is True
+ assert not (tmp_path / "nuttx.mmleak").exists()
+
+ def test_zero_leaked_blocks_no_dump(
+ self,
+ autobp: "ModuleType",
+ fake_gdb: "ModuleType",
+ monkeypatch: pytest.MonkeyPatch,
+ tmp_path: "Path",
+ ) -> None:
+ def execute(cmd: str, to_string: bool = False) -> str:
+ fake_gdb.executed.append(cmd)
+ return "Leaked 0 blks, 0 bytes"
+
+ monkeypatch.setattr(fake_gdb, "execute", execute)
+ cmd = _command(fake_gdb, "ntfcmmleak")
+ assert cmd.invoke(f"-d {tmp_path} gcore", False) is True
+ assert not (tmp_path / "nuttx.mmleak").exists()
+
+ def test_leak_writes_report_and_dumps(
+ self,
+ autobp: "ModuleType",
+ fake_gdb: "ModuleType",
+ monkeypatch: pytest.MonkeyPatch,
+ tmp_path: "Path",
+ ) -> None:
+ def execute(cmd: str, to_string: bool = False) -> str:
+ fake_gdb.executed.append(cmd)
+ if cmd == "mm leak":
+ return "Leaked 3 blks, 24 bytes"
+ return ""
+
+ monkeypatch.setattr(fake_gdb, "execute", execute)
+ cmd = _command(fake_gdb, "ntfcmmleak")
+ assert cmd.invoke(f"-d {tmp_path} gcore", False) is True
+ report = tmp_path / "nuttx.mmleak"
+ assert report.read_text() == "Leaked 3 blks, 24 bytes"
+ gcores = [c for c in fake_gdb.executed if c.startswith("gcore ")]
+ assert len(gcores) == 1
+ assert ".core" in gcores[0]
+
+
+class TestNtfcAutoPoweroffCheckCommand:
+ def test_prints_backtrace(
+ self, autobp: "ModuleType", fake_gdb: "ModuleType"
+ ) -> None:
+ cmd = _command(fake_gdb, "ntfcautopoweroffcheck")
+ assert cmd.invoke("", False) is True
+ assert "bt" in fake_gdb.executed
+
+
+class TestEventHandlers:
+ def test_module_connects_events(
+ self, autobp: "ModuleType", fake_gdb: "ModuleType"
+ ) -> None:
+ assert autobp.gdb_autobp_stop_event in fake_gdb.events.stop.callbacks
+ assert (
+ autobp.gdb_process_exited_event in fake_gdb.events.exited.callbacks
+ )
+
+ def test_stop_event_deletes_pending_and_continues(
+ self, autobp: "ModuleType", fake_gdb: "ModuleType"
+ ) -> None:
+ bp = autobp.AutoBp("main", ["bt"])
+ autobp.AutoBp._pending_delete.add(bp)
+ autobp.gdb_autobp_stop_event(fake_gdb.BreakpointEvent())
+ assert bp.deleted is True
+ assert autobp.AutoBp._pending_delete == set()
+ assert "continue" in fake_gdb.executed
+
+ def test_stop_event_delete_failure(
+ self,
+ autobp: "ModuleType",
+ fake_gdb: "ModuleType",
+ monkeypatch: pytest.MonkeyPatch,
+ capsys: pytest.CaptureFixture[str],
+ ) -> None:
+ bp = autobp.AutoBp("main", ["bt"])
+ autobp.AutoBp._pending_delete.add(bp)
+ monkeypatch.setattr(
+ fake_gdb.Breakpoint, "delete_error", fake_gdb.error("x")
+ )
+ autobp.gdb_autobp_stop_event(fake_gdb.BreakpointEvent())
+ assert "Failed to delete breakpoint" in capsys.readouterr().out
+ assert bp in autobp.AutoBp._pending_delete
+
+ def test_stop_event_no_pending(
+ self, autobp: "ModuleType", fake_gdb: "ModuleType"
+ ) -> None:
+ autobp.gdb_autobp_stop_event(fake_gdb.BreakpointEvent())
+ assert "continue" in fake_gdb.executed
+
+ def test_stop_event_continues_on_autobp_breakpoint(
+ self, autobp: "ModuleType", fake_gdb: "ModuleType"
+ ) -> None:
+ bp = autobp.AutoBp("main", ["bt"])
+ event = fake_gdb.BreakpointEvent(breakpoints=(bp,))
+ autobp.gdb_autobp_stop_event(event)
+ assert "continue" in fake_gdb.executed
+
+ def test_stop_event_does_not_resume_user_breakpoint(
+ self,
+ autobp: "ModuleType",
+ fake_gdb: "ModuleType",
+ capsys: pytest.CaptureFixture[str],
+ ) -> None:
+ user_bp = fake_gdb.Breakpoint("my_function")
+ pending = autobp.AutoBp("main", ["bt"])
+ autobp.AutoBp._pending_delete.add(pending)
+ event = fake_gdb.BreakpointEvent(breakpoints=(user_bp,))
+ autobp.gdb_autobp_stop_event(event)
+ assert "continue" not in fake_gdb.executed
+ assert pending.deleted is False
+ assert "not resuming" in capsys.readouterr().out
+
+ def test_stop_event_mixed_breakpoints_not_resumed(
+ self, autobp: "ModuleType", fake_gdb: "ModuleType"
+ ) -> None:
+ user_bp = fake_gdb.Breakpoint("my_function")
+ auto_bp = autobp.AutoBp("main", ["bt"])
+ event = fake_gdb.BreakpointEvent(breakpoints=(user_bp, auto_bp))
+ autobp.gdb_autobp_stop_event(event)
+ assert "continue" not in fake_gdb.executed
+
+ def test_stop_event_sigint(
+ self,
+ autobp: "ModuleType",
+ fake_gdb: "ModuleType",
+ capsys: pytest.CaptureFixture[str],
+ ) -> None:
+ autobp.gdb_autobp_stop_event(fake_gdb.SignalEvent("SIGINT"))
+ assert "SIGINT" in capsys.readouterr().out
+
+ def test_stop_event_sigterm_quits(
+ self, autobp: "ModuleType", fake_gdb: "ModuleType"
+ ) -> None:
+ autobp.gdb_autobp_stop_event(fake_gdb.SignalEvent("SIGTERM"))
+ assert "q" in fake_gdb.executed
+
+ def test_stop_event_unexpected_signal(
+ self,
+ autobp: "ModuleType",
+ fake_gdb: "ModuleType",
+ capsys: pytest.CaptureFixture[str],
+ ) -> None:
+ autobp.gdb_autobp_stop_event(fake_gdb.SignalEvent("SIGUSR1"))
+ assert "Unexpected Signal" in capsys.readouterr().out
+
+ def test_stop_event_other_event_ignored(
+ self, autobp: "ModuleType", fake_gdb: "ModuleType"
+ ) -> None:
+ autobp.gdb_autobp_stop_event(object())
+ assert fake_gdb.executed == []
+
+ def test_exited_event_quits(
+ self, autobp: "ModuleType", fake_gdb: "ModuleType"
+ ) -> None:
+ autobp.gdb_process_exited_event(object())
+ assert "q" in fake_gdb.executed
+
+
+class TestGdbPrefix:
+ @pytest.fixture
+ def gdbprefix(self, fake_gdb: "ModuleType") -> "ModuleType":
+ return importlib.import_module("ntfc.debug.gdb.plugin.gdbprefix")
+
+ def test_registered(
+ self, gdbprefix: "ModuleType", fake_gdb: "ModuleType"
+ ) -> None:
+ assert "ntfcgdbprefix" in dict(fake_gdb.Command.registered)
+
+ def test_successful_command(
+ self,
+ gdbprefix: "ModuleType",
+ fake_gdb: "ModuleType",
+ capsys: pytest.CaptureFixture[str],
+ ) -> None:
+ cmd = _command(fake_gdb, "ntfcgdbprefix")
+ cmd.invoke("info registers", False)
+ assert "info registers" in fake_gdb.executed
+ assert "run successful" in capsys.readouterr().out
+
+ def test_failed_command_quits_gdb(
+ self,
+ gdbprefix: "ModuleType",
+ fake_gdb: "ModuleType",
+ monkeypatch: pytest.MonkeyPatch,
+ capsys: pytest.CaptureFixture[str],
+ ) -> None:
+ def execute(cmd: str, to_string: bool = False) -> str:
+ fake_gdb.executed.append(cmd)
+ if cmd != "q":
+ raise fake_gdb.error("bad command")
+ return ""
+
+ monkeypatch.setattr(fake_gdb, "execute", execute)
+ cmd = _command(fake_gdb, "ntfcgdbprefix")
+ cmd.invoke("bogus", False)
+ assert "run failed" in capsys.readouterr().out
+ assert "q" in fake_gdb.executed
+
+
+class TestInitLoader:
+ def test_load_imports_plugin_modules(
+ self,
+ fake_gdb: "ModuleType",
+ monkeypatch: pytest.MonkeyPatch,
+ capsys: pytest.CaptureFixture[str],
+ ) -> None:
+ monkeypatch.setattr(sys, "path", list(sys.path))
+ init = importlib.import_module("ntfc.debug.gdb.plugin.init")
+ init.load()
+ out = capsys.readouterr().out
+ assert "import autobp" in out
+ assert "import gdbprefix" in out
+ assert "NTFC GDB Plugin loaded Successfully" in out
+ assert "ntfcautobp" in dict(fake_gdb.Command.registered)
+ # Second call: plugin dir is already in sys.path
+ init.load()
+
+ def test_load_reports_import_errors(
+ self,
+ fake_gdb: "ModuleType",
+ monkeypatch: pytest.MonkeyPatch,
+ capsys: pytest.CaptureFixture[str],
+ ) -> None:
+ monkeypatch.setattr(sys, "path", list(sys.path))
+ init = importlib.import_module("ntfc.debug.gdb.plugin.init")
+
+ def fail(name: str) -> "ModuleType":
+ raise ImportError(f"broken {name}")
+
+ monkeypatch.setattr(init.importlib, "import_module", fail)
+ init.load()
+ out = capsys.readouterr().out
+ assert "ImportError" in out
+ assert "NTFC GDB Plugin loaded Successfully" in out
diff --git a/tests/debug/test_getdebug.py b/tests/debug/test_getdebug.py
new file mode 100644
index 0000000..9961b99
--- /dev/null
+++ b/tests/debug/test_getdebug.py
@@ -0,0 +1,373 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Dict, Optional
+from unittest.mock import MagicMock, patch
+
+if TYPE_CHECKING:
+ import pytest
+
+from ntfc.debug.config import DebugConfig
+from ntfc.debug.getdebug import DebugSetup, get_debug_setup
+from ntfc.pytest.coredump_plugin import CoredumpPlugin
+
+
+def _make_product(
+ name: str = "prod",
+ debug: Optional[Dict[str, Any]] = None,
+ elf: str = "",
+ cores_num: int = 1,
+) -> MagicMock:
+ p = MagicMock()
+ p.name = name
+ p.conf.debug = DebugConfig(debug or {})
+ p.conf.cfg_core.return_value.elf_path = elf
+ p.conf.cores_num = cores_num
+ return p
+
+
+def _plugin_types(setup: DebugSetup) -> list:
+ return [type(plugin) for plugin in setup.plugins]
+
+
+class TestDebugSetupCreate:
+ def test_no_products(self) -> None:
+ setup = get_debug_setup([])
+ assert setup.plugins == []
+
+ def test_nothing_enabled(self) -> None:
+ setup = get_debug_setup([_make_product()])
+ assert setup.plugins == []
+
+ def test_coredump_without_handlers(self) -> None:
+ p = _make_product(debug={"coredump": {"enable": True}})
+ setup = get_debug_setup([p])
+ assert _plugin_types(setup) == [CoredumpPlugin]
+ mgr = setup.plugins[0]._managers["prod"]
+ assert mgr._handlers == []
+ assert setup.plugins[0]._products == [p]
+
+ def test_gdb_handler_registered(self) -> None:
+ p = _make_product(
+ debug={
+ "coredump": {"enable": True},
+ "gdb": {"enable": True, "target": "localhost:1234"},
+ },
+ elf="/fw/nuttx",
+ )
+ with (
+ patch("ntfc.debug.getdebug.GdbController") as ctrl_cls,
+ patch("ntfc.debug.getdebug.GdbHandler") as handler_cls,
+ ):
+ setup = get_debug_setup([p])
+ args, ctrl_kwargs = ctrl_cls.call_args
+ assert args == (Path("/fw/nuttx"), p.conf.debug.gdb)
+ assert callable(ctrl_kwargs["pid_provider"])
+ _, handler_kwargs = handler_cls.call_args
+ assert handler_cls.call_args.args == (ctrl_cls.return_value,)
+ assert handler_kwargs["force_panic"] is None
+ assert callable(handler_kwargs["crash_check"])
+ assert handler_kwargs["auto_dump"] is False
+ mgr = setup.plugins[0]._managers["prod"]
+ assert mgr._handlers == [handler_cls.return_value]
+ assert setup._controllers_by_product == {
+ "prod": [ctrl_cls.return_value]
+ }
+
+ def test_gdb_handler_gets_force_panic_callable(self) -> None:
+ p = _make_product(
+ debug={
+ "coredump": {"enable": True},
+ "gdb": {
+ "enable": True,
+ "target": "localhost:1234",
+ "force_panic": True,
+ },
+ },
+ elf="/fw/nuttx",
+ )
+ with (
+ patch("ntfc.debug.getdebug.GdbController"),
+ patch("ntfc.debug.getdebug.GdbHandler") as handler_cls,
+ ):
+ get_debug_setup([p])
+ _, kwargs = handler_cls.call_args
+ assert kwargs["force_panic"] is p.force_panic
+
+ def test_gdb_enabled_without_elf_skipped(self) -> None:
+ p = _make_product(
+ debug={
+ "coredump": {"enable": True},
+ "gdb": {"enable": True, "target": "localhost:1234"},
+ },
+ elf="",
+ )
+ setup = get_debug_setup([p])
+ mgr = setup.plugins[0]._managers["prod"]
+ assert mgr._handlers == []
+ assert setup._controllers_by_product == {"prod": []}
+
+ def test_gdb_enabled_without_target_or_attach_skipped(self) -> None:
+ p = _make_product(
+ debug={
+ "coredump": {"enable": True},
+ "gdb": {"enable": True},
+ },
+ elf="/fw/nuttx",
+ )
+ setup = get_debug_setup([p])
+ mgr = setup.plugins[0]._managers["prod"]
+ assert mgr._handlers == []
+ assert setup._controllers_by_product == {"prod": []}
+
+ def test_gdb_attach_mode_registers_with_pid_provider(self) -> None:
+ p = _make_product(
+ debug={
+ "coredump": {"enable": True},
+ "gdb": {"enable": True, "attach": True},
+ },
+ elf="/fw/nuttx",
+ )
+ p.core.return_value.device.pid = 4242
+ with (
+ patch("ntfc.debug.getdebug.GdbController") as ctrl_cls,
+ patch("ntfc.debug.getdebug.GdbHandler"),
+ ):
+ get_debug_setup([p])
+ _, kwargs = ctrl_cls.call_args
+ assert kwargs["pid_provider"]() == 4242
+
+ def test_gdb_handler_gets_crash_check_and_auto_dump(self) -> None:
+ p = _make_product(
+ debug={
+ "coredump": {"enable": True},
+ "gdb": {
+ "enable": True,
+ "target": "localhost:1234",
+ "plugin": True,
+ "auto_breakpoints": True,
+ },
+ },
+ elf="/fw/nuttx",
+ )
+ p.crash = True
+ with (
+ patch("ntfc.debug.getdebug.GdbController"),
+ patch("ntfc.debug.getdebug.GdbHandler") as handler_cls,
+ ):
+ get_debug_setup([p])
+ _, kwargs = handler_cls.call_args
+ assert kwargs["auto_dump"] is True
+ assert kwargs["crash_check"]() is True
+
+ def test_local_file_handler_registered(self, tmp_path: Path) -> None:
+ p = _make_product(
+ debug={
+ "coredump": {"enable": True},
+ "local_file": {"core_dir": str(tmp_path)},
+ }
+ )
+ setup = get_debug_setup([p])
+ mgr = setup.plugins[0]._managers["prod"]
+ assert [h.name for h in mgr._handlers] == ["local_file"]
+
+ def test_syslog_handler_registered(self) -> None:
+ p = _make_product(
+ debug={
+ "coredump": {"enable": True},
+ "syslog": {"enable": True},
+ }
+ )
+ setup = get_debug_setup([p])
+ mgr = setup.plugins[0]._managers["prod"]
+ assert [h.name for h in mgr._handlers] == ["syslog"]
+
+ def test_fastboot_handler_registered(self) -> None:
+ p = _make_product(
+ debug={
+ "coredump": {"enable": True},
+ "fastboot": {
+ "dev_sn": "SN1",
+ "mem_addr": "0x1000",
+ "mem_size": "0x2000",
+ },
+ }
+ )
+ with (
+ patch("ntfc.debug.getdebug.FastbootController") as ctrl_cls,
+ patch("ntfc.debug.getdebug.FastbootHandler") as handler_cls,
+ ):
+ setup = get_debug_setup([p])
+ ctrl_cls.assert_called_once_with("SN1")
+ handler_cls.assert_called_once_with(
+ ctrl_cls.return_value, "0x1000", "0x2000"
+ )
+ mgr = setup.plugins[0]._managers["prod"]
+ assert mgr._handlers == [handler_cls.return_value]
+
+ def test_ymodem_handler_registered(self) -> None:
+ p = _make_product(
+ debug={
+ "coredump": {"enable": True},
+ "ymodem": {
+ "serial_port": "/dev/ttyUSB0",
+ "baud_rate": 115200,
+ "sbrb_path": "/tools/sbrb.py",
+ },
+ }
+ )
+ with (
+ patch("ntfc.debug.getdebug.YmodemController") as ctrl_cls,
+ patch("ntfc.debug.getdebug.YmodemHandler") as handler_cls,
+ ):
+ setup = get_debug_setup([p])
+ ctrl_cls.assert_called_once_with(
+ "/dev/ttyUSB0", Path("/tools/sbrb.py"), 115200
+ )
+ handler_cls.assert_called_once_with(
+ ctrl_cls.return_value, p.core.return_value.device
+ )
+ mgr = setup.plugins[0]._managers["prod"]
+ assert mgr._handlers == [handler_cls.return_value]
+
+ def test_ymodem_requires_serial_and_sbrb(self) -> None:
+ p = _make_product(
+ debug={
+ "coredump": {"enable": True},
+ "ymodem": {"serial_port": "/dev/ttyUSB0"},
+ }
+ )
+ setup = get_debug_setup([p])
+ mgr = setup.plugins[0]._managers["prod"]
+ assert mgr._handlers == []
+
+ def test_multicore_product_warns_but_registers(
+ self, caplog: "pytest.LogCaptureFixture"
+ ) -> None:
+ p = _make_product(debug={"coredump": {"enable": True}}, cores_num=2)
+ with caplog.at_level("WARNING"):
+ setup = get_debug_setup([p])
+ assert "only supports core 0" in caplog.text
+ assert _plugin_types(setup) == [CoredumpPlugin]
+
+ def test_multiple_products(self) -> None:
+ p1 = _make_product(name="p1", debug={"coredump": {"enable": True}})
+ p2 = _make_product(name="p2", debug={"coredump": {"enable": True}})
+ setup = get_debug_setup([p1, p2])
+ assert _plugin_types(setup) == [CoredumpPlugin]
+ assert list(setup.plugins[0]._managers) == ["p1", "p2"]
+
+
+class TestDebugSetup:
+ def test_plugins_returns_copy(self) -> None:
+ plugin = MagicMock()
+ setup = DebugSetup([plugin], {})
+ plugins = setup.plugins
+ plugins.append(MagicMock())
+ assert setup.plugins == [plugin]
+
+ def test_start_and_stop(self) -> None:
+ ctrl = MagicMock()
+ ctrl.start.return_value = True
+ setup = DebugSetup([], {"prod": [ctrl]})
+ setup.start("/results")
+ ctrl.start.assert_called_once()
+ ctrl.setup.assert_called_once_with("/results/prod")
+ setup.stop()
+ ctrl.stop.assert_called_once()
+ # stop() clears the started list
+ setup.stop()
+ ctrl.stop.assert_called_once()
+
+ def test_start_failure_skips_setup(self) -> None:
+ good = MagicMock()
+ good.start.return_value = True
+ bad = MagicMock()
+ bad.start.return_value = False
+ setup = DebugSetup([], {"prod": [bad, good]})
+ setup.start("/results")
+ bad.setup.assert_not_called()
+ good.setup.assert_called_once_with("/results/prod")
+ setup.stop()
+ bad.stop.assert_not_called()
+ good.stop.assert_called_once()
+
+ def test_start_noop_without_result_dir(self) -> None:
+ ctrl = MagicMock()
+ setup = DebugSetup([], {"prod": [ctrl]})
+ setup.start("")
+ ctrl.start.assert_not_called()
+
+ def test_restart_controllers_reattaches(self) -> None:
+ ctrl = MagicMock()
+ ctrl.start.return_value = True
+ setup = DebugSetup([], {"prod": [ctrl]})
+ setup.start("/results")
+ ctrl.start.reset_mock()
+ ctrl.setup.reset_mock()
+
+ setup.restart_controllers("prod")
+
+ ctrl.stop.assert_called_once()
+ ctrl.start.assert_called_once()
+ ctrl.setup.assert_called_once_with("/results/prod")
+
+ def test_restart_controllers_starts_previously_failed(self) -> None:
+ ctrl = MagicMock()
+ ctrl.start.return_value = False
+ setup = DebugSetup([], {"prod": [ctrl]})
+ setup.start("/results")
+ ctrl.start.assert_called_once()
+
+ ctrl.start.return_value = True
+ setup.restart_controllers("prod")
+
+ # never started, so no stop() before the retry
+ ctrl.stop.assert_not_called()
+ assert ctrl.start.call_count == 2
+
+ def test_restart_controllers_unknown_product_is_noop(self) -> None:
+ setup = DebugSetup([], {})
+ setup.start("/results")
+ setup.restart_controllers("missing") # must not raise
+
+ def test_restart_controllers_noop_before_start(self) -> None:
+ ctrl = MagicMock()
+ setup = DebugSetup([], {"prod": [ctrl]})
+ setup.restart_controllers("prod")
+ ctrl.start.assert_not_called()
+
+ def test_restart_all_controllers_restarts_every_product(self) -> None:
+ ctrl_a = MagicMock()
+ ctrl_a.start.return_value = True
+ ctrl_b = MagicMock()
+ ctrl_b.start.return_value = True
+ setup = DebugSetup([], {"a": [ctrl_a], "b": [ctrl_b]})
+ setup.start("/results")
+ ctrl_a.start.reset_mock()
+ ctrl_b.start.reset_mock()
+
+ setup.restart_all_controllers()
+
+ ctrl_a.stop.assert_called_once()
+ ctrl_a.start.assert_called_once()
+ ctrl_b.stop.assert_called_once()
+ ctrl_b.start.assert_called_once()
diff --git a/tests/device/test_common.py b/tests/device/test_common.py
index 5a474f4..a8f848b 100644
--- a/tests/device/test_common.py
+++ b/tests/device/test_common.py
@@ -288,4 +288,69 @@ def test_device_common_log_helpers():
assert "buffered-cmd" in output
+def test_pid_default_none():
+
+ with patch("ntfc.envconfig.EnvConfig") as mockdevice:
+ config = mockdevice.return_value
+ dev = DeviceMock(config)
+
+ assert dev.pid is None
+
+
+def test_output_tail_accumulates():
+
+ with patch("ntfc.envconfig.EnvConfig") as mockdevice:
+ config = mockdevice.return_value
+ dev = DeviceMock(config)
+
+ dev._output_tail_append(b"hello ")
+ dev._output_tail_append(b"world")
+ assert dev.output_tail() == "hello world"
+
+
+def test_output_tail_capped():
+
+ with patch("ntfc.envconfig.EnvConfig") as mockdevice:
+ config = mockdevice.return_value
+ dev = DeviceMock(config)
+
+ dev._output_tail_append(b"x" * (4 * 1024 * 1024))
+ dev._output_tail_append(b"tail-marker")
+ tail = dev.output_tail()
+ assert len(tail) <= 4 * 1024 * 1024
+ assert tail.endswith("tail-marker")
+
+
+def test_reset_output_tail_clears_buffer():
+
+ with patch("ntfc.envconfig.EnvConfig") as mockdevice:
+ config = mockdevice.return_value
+ dev = DeviceMock(config)
+
+ dev._output_tail_append(b"stale crash markers")
+ dev.reset_output_tail()
+
+ assert dev.output_tail() == ""
+ dev._output_tail_append(b"fresh")
+ assert dev.output_tail() == "fresh"
+
+
+def test_read_all_appends_chunks_to_output_tail(monkeypatch):
+
+ with patch("ntfc.envconfig.EnvConfig") as mockdevice:
+ config = mockdevice.return_value
+ dev = DeviceMock(config)
+
+ chunks = [b"AAA", b"BBB"]
+
+ def fake_read(_=0):
+ return chunks.pop(0) if chunks else b""
+
+ monkeypatch.setattr(dev, "_read", fake_read)
+ dev._read_all(timeout=0.05)
+
+ # per-chunk append, not cumulative re-append of growing output
+ assert dev.output_tail() == "AAABBB"
+
+
# TODO: missing tests
diff --git a/tests/device/test_host.py b/tests/device/test_host.py
index fb1b2d4..6ddebd9 100644
--- a/tests/device/test_host.py
+++ b/tests/device/test_host.py
@@ -18,6 +18,8 @@
#
############################################################################
+from unittest.mock import MagicMock
+
import pytest
from pexpect.exceptions import ExceptionPexpect
@@ -188,6 +190,18 @@ def test_device_host_reboot_boot_wait_failure(envconfig_dummy, monkeypatch):
assert dev.reboot(1, hard=False) is False
+def test_device_host_pid(envconfig_dummy):
+
+ conf = envconfig_dummy.product[0].cfg_core(0)
+ dev = DeviceHost2(conf)
+
+ assert dev.pid is None
+
+ dev._child = MagicMock()
+ dev._child.pid = 4242
+ assert dev.pid == 4242
+
+
# TODO: more tests for host device !!!!
# - test for timeout
# - test for very long output
diff --git a/tests/lib/test_fastboot_controller.py b/tests/lib/test_fastboot_controller.py
new file mode 100644
index 0000000..d9e0fda
--- /dev/null
+++ b/tests/lib/test_fastboot_controller.py
@@ -0,0 +1,184 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+import subprocess
+from typing import TYPE_CHECKING
+from unittest.mock import MagicMock, patch
+
+if TYPE_CHECKING:
+ from pathlib import Path
+
+from ntfc.lib.fastboot.controller import FastbootController
+
+
+def _ctrl(dev_sn: str = "ABC123") -> FastbootController:
+ return FastbootController(dev_sn)
+
+
+def _run_ok() -> MagicMock:
+ m = MagicMock()
+ m.returncode = 0
+ m.stdout = b""
+ return m
+
+
+def _run_fail() -> MagicMock:
+ m = MagicMock()
+ m.returncode = 1
+ m.stdout = b"FAILED\n"
+ return m
+
+
+class TestFastbootControllerIsConnected:
+ def test_true_when_dev_sn_in_output(self) -> None:
+ with patch("subprocess.run") as mock_run:
+ mock_run.return_value = MagicMock(
+ stdout=b"ABC123\tfastboot\n", returncode=0
+ )
+ assert _ctrl().is_connected() is True
+
+ def test_false_when_dev_sn_not_in_output(self) -> None:
+ with patch("subprocess.run") as mock_run:
+ mock_run.return_value = MagicMock(
+ stdout=b"OTHER\tfastboot\n", returncode=0
+ )
+ assert _ctrl().is_connected() is False
+
+ def test_false_when_output_empty(self) -> None:
+ with patch("subprocess.run") as mock_run:
+ mock_run.return_value = MagicMock(stdout=b"", returncode=0)
+ assert _ctrl().is_connected() is False
+
+ def test_false_on_timeout_expired(self) -> None:
+ with patch(
+ "subprocess.run",
+ side_effect=subprocess.TimeoutExpired(cmd="fastboot", timeout=10),
+ ):
+ assert _ctrl().is_connected() is False
+
+ def test_false_on_oserror(self) -> None:
+ with patch(
+ "subprocess.run", side_effect=OSError("fastboot not found")
+ ):
+ assert _ctrl().is_connected() is False
+
+ def test_false_when_dev_sn_empty_even_with_devices_listed(self) -> None:
+ with patch("subprocess.run") as mock_run:
+ mock_run.return_value = MagicMock(
+ stdout=b"ABC123\tfastboot\n", returncode=0
+ )
+ assert _ctrl(dev_sn="").is_connected() is False
+ mock_run.assert_not_called()
+
+
+class TestFastbootControllerMemdump:
+ def test_returns_path_on_success(self, tmp_path: "Path") -> None:
+ with patch("subprocess.run", return_value=_run_ok()):
+ result = _ctrl().memdump(
+ "0x40000000", "0x08000000", tmp_path / "core.bin"
+ )
+ assert result == tmp_path / "core.bin"
+
+ def test_creates_parent_directory(self, tmp_path: "Path") -> None:
+ nested = tmp_path / "sub" / "dir" / "core.bin"
+ with patch("subprocess.run", return_value=_run_ok()):
+ _ctrl().memdump("0x40000000", "0x08000000", nested)
+ assert nested.parent.is_dir()
+
+ def test_returns_none_when_memdump_fails(self, tmp_path: "Path") -> None:
+ with patch("subprocess.run", return_value=_run_fail()):
+ result = _ctrl().memdump(
+ "0x40000000", "0x08000000", tmp_path / "core.bin"
+ )
+ assert result is None
+
+ def test_returns_none_when_get_staged_fails(
+ self, tmp_path: "Path"
+ ) -> None:
+ with patch("subprocess.run") as mock_run:
+ mock_run.side_effect = [_run_ok(), _run_fail()]
+ result = _ctrl().memdump(
+ "0x40000000", "0x08000000", tmp_path / "core.bin"
+ )
+ assert result is None
+
+ def test_passes_addr_and_size_to_command(self, tmp_path: "Path") -> None:
+ ctrl = FastbootController("DEV1")
+ with patch("subprocess.run", return_value=_run_ok()) as mock_run:
+ ctrl.memdump("0x20000000", "0x04000000", tmp_path / "core.bin")
+ memdump_cmd = mock_run.call_args_list[0][0][0]
+ assert "0x20000000" in memdump_cmd
+ assert "0x04000000" in memdump_cmd
+
+ def test_uses_dev_sn_in_commands(self, tmp_path: "Path") -> None:
+ ctrl = FastbootController("MYDEVICE")
+ with patch("subprocess.run", return_value=_run_ok()) as mock_run:
+ ctrl.memdump("0x40000000", "0x08000000", tmp_path / "core.bin")
+ for c in mock_run.call_args_list:
+ assert "MYDEVICE" in c[0][0]
+
+ def test_returns_none_on_timeout(self, tmp_path: "Path") -> None:
+ with patch(
+ "subprocess.run",
+ side_effect=subprocess.TimeoutExpired(cmd="fastboot", timeout=30),
+ ):
+ result = _ctrl().memdump(
+ "0x40000000", "0x08000000", tmp_path / "core.bin"
+ )
+ assert result is None
+
+ def test_returns_none_on_oserror(self, tmp_path: "Path") -> None:
+ with patch(
+ "subprocess.run", side_effect=OSError("fastboot not found")
+ ):
+ result = _ctrl().memdump(
+ "0x40000000", "0x08000000", tmp_path / "core.bin"
+ )
+ assert result is None
+
+
+class TestFastbootControllerRun:
+ def test_returns_true_on_success(self) -> None:
+ with patch("subprocess.run", return_value=_run_ok()):
+ assert _ctrl().run(["oem", "test"], timeout=10.0) is True
+
+ def test_returns_false_on_nonzero_exit(self) -> None:
+ with patch("subprocess.run", return_value=_run_fail()):
+ assert _ctrl().run(["oem", "test"], timeout=10.0) is False
+
+ def test_returns_false_on_timeout(self) -> None:
+ with patch(
+ "subprocess.run",
+ side_effect=subprocess.TimeoutExpired(cmd="fastboot", timeout=10),
+ ):
+ assert _ctrl().run(["oem", "test"], timeout=10.0) is False
+
+ def test_returns_false_on_oserror(self) -> None:
+ with patch(
+ "subprocess.run", side_effect=OSError("fastboot not found")
+ ):
+ assert _ctrl().run(["oem", "test"], timeout=10.0) is False
+
+ def test_includes_dev_sn_in_command(self) -> None:
+ ctrl = FastbootController("SN123")
+ with patch("subprocess.run", return_value=_run_ok()) as mock_run:
+ ctrl.run(["oem", "test"], timeout=10.0)
+ cmd = mock_run.call_args[0][0]
+ assert "SN123" in cmd
diff --git a/tests/lib/test_ymodem_controller.py b/tests/lib/test_ymodem_controller.py
new file mode 100644
index 0000000..29d1d91
--- /dev/null
+++ b/tests/lib/test_ymodem_controller.py
@@ -0,0 +1,107 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+import subprocess
+from typing import TYPE_CHECKING
+from unittest.mock import MagicMock, patch
+
+if TYPE_CHECKING:
+ from pathlib import Path
+
+from ntfc.lib.ymodem.controller import YmodemController
+
+
+def _run_ok(output: str = "sbrb success\n") -> MagicMock:
+ m = MagicMock()
+ m.returncode = 0
+ m.stdout = output.encode()
+ return m
+
+
+def _run_fail() -> MagicMock:
+ m = MagicMock()
+ m.returncode = 1
+ m.stdout = b"sbrb failed\n"
+ return m
+
+
+class TestYmodemControllerIsReady:
+ def test_true_when_configured(self, tmp_path: "Path") -> None:
+ sbrb = tmp_path / "sbrb.py"
+ sbrb.write_text("x")
+ ctrl = YmodemController("/dev/ttyUSB0", sbrb)
+ assert ctrl.is_ready() is True
+
+ def test_false_when_sbrb_missing(self, tmp_path: "Path") -> None:
+ ctrl = YmodemController("/dev/ttyUSB0", tmp_path / "missing.py")
+ assert ctrl.is_ready() is False
+
+ def test_false_when_serial_port_empty(self, tmp_path: "Path") -> None:
+ sbrb = tmp_path / "sbrb.py"
+ sbrb.write_text("x")
+ ctrl = YmodemController("", sbrb)
+ assert ctrl.is_ready() is False
+
+
+class TestYmodemControllerDownload:
+ def test_returns_true_on_success(self, tmp_path: "Path") -> None:
+ sbrb = tmp_path / "sbrb.py"
+ sbrb.write_text("x")
+ ctrl = YmodemController("/dev/ttyUSB0", sbrb)
+ with patch("subprocess.run", return_value=_run_ok()):
+ assert ctrl.download(tmp_path, "app.core") is True
+
+ def test_returns_false_when_sbrb_reports_failure(
+ self, tmp_path: "Path"
+ ) -> None:
+ sbrb = tmp_path / "sbrb.py"
+ sbrb.write_text("x")
+ ctrl = YmodemController("/dev/ttyUSB0", sbrb)
+ with patch("subprocess.run", return_value=_run_fail()):
+ assert ctrl.download(tmp_path, "app.core") is False
+
+ def test_returns_false_on_timeout(self, tmp_path: "Path") -> None:
+ sbrb = tmp_path / "sbrb.py"
+ sbrb.write_text("x")
+ ctrl = YmodemController("/dev/ttyUSB0", sbrb)
+ with patch(
+ "subprocess.run",
+ side_effect=subprocess.TimeoutExpired(cmd="sbrb.py", timeout=3600),
+ ):
+ assert ctrl.download(tmp_path, "app.core") is False
+
+ def test_returns_false_on_oserror(self, tmp_path: "Path") -> None:
+ sbrb = tmp_path / "sbrb.py"
+ sbrb.write_text("x")
+ ctrl = YmodemController("/dev/ttyUSB0", sbrb)
+ with patch(
+ "subprocess.run", side_effect=OSError("sbrb.py not executable")
+ ):
+ assert ctrl.download(tmp_path, "app.core") is False
+
+ def test_passes_serial_port_and_baud_rate(self, tmp_path: "Path") -> None:
+ sbrb = tmp_path / "sbrb.py"
+ sbrb.write_text("x")
+ ctrl = YmodemController("/dev/ttyUSB1", sbrb, baud_rate=115200)
+ with patch("subprocess.run", return_value=_run_ok()) as mock_run:
+ ctrl.download(tmp_path, "app.core")
+ cmd = mock_run.call_args[0][0]
+ assert "/dev/ttyUSB1" in cmd
+ assert "115200" in cmd
diff --git a/tests/pytest/test_configure.py b/tests/pytest/test_configure.py
index 9ba1075..65f8fbd 100644
--- a/tests/pytest/test_configure.py
+++ b/tests/pytest/test_configure.py
@@ -32,6 +32,16 @@ def test_test_pytestconfigureplugin_init(config_dummy):
_ = PytestConfigPlugin(config_dummy)
+def test_set_restart_gdb_stores_callback(config_dummy):
+ plugin = PytestConfigPlugin(config_dummy)
+ assert plugin._restart_gdb is None
+
+ callback = MagicMock()
+ plugin.set_restart_gdb(callback)
+
+ assert plugin._restart_gdb is callback
+
+
def test_pytest_configure_log_file_uses_result_dir(
config_dummy, monkeypatch, tmp_path
):
diff --git a/tests/pytest/test_coredump_plugin.py b/tests/pytest/test_coredump_plugin.py
new file mode 100644
index 0000000..d0866b2
--- /dev/null
+++ b/tests/pytest/test_coredump_plugin.py
@@ -0,0 +1,212 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+from typing import TYPE_CHECKING
+from unittest.mock import MagicMock
+
+import pytest
+
+from ntfc.pytest.coredump_plugin import CoredumpPlugin
+
+if TYPE_CHECKING:
+ from pathlib import Path
+
+
+def _make_report(when: str = "call", failed: bool = True) -> MagicMock:
+ r = MagicMock()
+ r.when = when
+ r.failed = failed
+ return r
+
+
+def _make_outcome(report: MagicMock) -> MagicMock:
+ outcome = MagicMock()
+ outcome.get_result.return_value = report
+ return outcome
+
+
+def _drive_hook(
+ plugin: CoredumpPlugin, nodeid: str, report: MagicMock
+) -> None:
+ """Drive the hookwrapper generator through one cycle."""
+ item = MagicMock()
+ item.nodeid = nodeid
+ gen = plugin.pytest_runtest_makereport(item, MagicMock())
+ next(gen)
+ outcome = _make_outcome(report)
+ try:
+ gen.send(outcome)
+ except StopIteration:
+ pass
+
+
+class TestCoredumpPluginOnTestFailed:
+ def test_calls_collect_for_each_manager(self, tmp_path: "Path") -> None:
+ mgr_a = MagicMock()
+ mgr_b = MagicMock()
+ plugin = CoredumpPlugin({"a": mgr_a, "b": mgr_b})
+ plugin._on_test_failed("mod::test_foo", str(tmp_path))
+ mgr_a.collect.assert_called_once()
+ mgr_b.collect.assert_called_once()
+
+ def test_prefix_derived_from_last_nodeid_segment(
+ self, tmp_path: "Path"
+ ) -> None:
+ mgr = MagicMock()
+ plugin = CoredumpPlugin({"p": mgr})
+ plugin._on_test_failed("module::TestClass::test_bar", str(tmp_path))
+ _, prefix = mgr.collect.call_args[0]
+ assert prefix == "test_bar"
+
+ def test_prefix_sanitizes_special_chars(self, tmp_path: "Path") -> None:
+ mgr = MagicMock()
+ plugin = CoredumpPlugin({"p": mgr})
+ plugin._on_test_failed("mod::test_foo[param=1]", str(tmp_path))
+ _, prefix = mgr.collect.call_args[0]
+ assert "[" not in prefix
+ assert "]" not in prefix
+ assert "=" not in prefix
+
+ def test_repeated_test_name_gets_unique_prefix(
+ self, tmp_path: "Path"
+ ) -> None:
+ mgr = MagicMock()
+ plugin = CoredumpPlugin({"p": mgr})
+ plugin._on_test_failed("mod_a::test_open", str(tmp_path))
+ plugin._on_test_failed("mod_b::test_open", str(tmp_path))
+ plugin._on_test_failed("mod_c::test_open", str(tmp_path))
+ prefixes = [c.args[1] for c in mgr.collect.call_args_list]
+ assert prefixes == ["test_open", "test_open.2", "test_open.3"]
+
+ def test_distinct_test_names_not_suffixed(self, tmp_path: "Path") -> None:
+ mgr = MagicMock()
+ plugin = CoredumpPlugin({"p": mgr})
+ plugin._on_test_failed("mod::test_a", str(tmp_path))
+ plugin._on_test_failed("mod::test_b", str(tmp_path))
+ prefixes = [c.args[1] for c in mgr.collect.call_args_list]
+ assert prefixes == ["test_a", "test_b"]
+
+ def test_output_dir_is_product_subdir(self, tmp_path: "Path") -> None:
+ mgr = MagicMock()
+ plugin = CoredumpPlugin({"myproduct": mgr})
+ plugin._on_test_failed("::test_x", str(tmp_path))
+ out_dir, _ = mgr.collect.call_args[0]
+ assert out_dir == tmp_path / "myproduct"
+
+
+def _make_product(
+ name: str,
+ crash: bool = False,
+ busyloop: bool = False,
+ notalive: bool = False,
+) -> MagicMock:
+ p = MagicMock()
+ p.name = name
+ p.crash = crash
+ p.busyloop = busyloop
+ p.notalive = notalive
+ return p
+
+
+class TestCoredumpPluginTargeting:
+ def test_all_products_when_all_healthy(self, tmp_path: "Path") -> None:
+ mgr_a = MagicMock()
+ mgr_b = MagicMock()
+ products = [_make_product("a"), _make_product("b")]
+ plugin = CoredumpPlugin({"a": mgr_a, "b": mgr_b}, products)
+ plugin._on_test_failed("mod::test_x", str(tmp_path))
+ mgr_a.collect.assert_called_once()
+ mgr_b.collect.assert_called_once()
+
+ def test_only_unhealthy_product_collected(self, tmp_path: "Path") -> None:
+ mgr_a = MagicMock()
+ mgr_b = MagicMock()
+ products = [_make_product("a", crash=True), _make_product("b")]
+ plugin = CoredumpPlugin({"a": mgr_a, "b": mgr_b}, products)
+ plugin._on_test_failed("mod::test_x", str(tmp_path))
+ mgr_a.collect.assert_called_once()
+ mgr_b.collect.assert_not_called()
+
+ def test_unhealthy_product_without_manager_skips_all(
+ self, tmp_path: "Path"
+ ) -> None:
+ mgr_b = MagicMock()
+ products = [_make_product("a", notalive=True), _make_product("b")]
+ plugin = CoredumpPlugin({"b": mgr_b}, products)
+ plugin._on_test_failed("mod::test_x", str(tmp_path))
+ mgr_b.collect.assert_not_called()
+
+ def test_all_managers_without_products(self, tmp_path: "Path") -> None:
+ mgr = MagicMock()
+ plugin = CoredumpPlugin({"p": mgr})
+ plugin._on_test_failed("mod::test_x", str(tmp_path))
+ mgr.collect.assert_called_once()
+
+
+class TestCoredumpPluginHook:
+ def test_collects_on_failed_call(self, tmp_path: "Path") -> None:
+ mgr = MagicMock()
+ plugin = CoredumpPlugin({"p": mgr})
+ pytest.result_dir = str(tmp_path)
+ _drive_hook(plugin, "mod::test_x", _make_report(failed=True))
+ mgr.collect.assert_called_once()
+
+ def test_skips_on_passed_call(self, tmp_path: "Path") -> None:
+ mgr = MagicMock()
+ plugin = CoredumpPlugin({"p": mgr})
+ pytest.result_dir = str(tmp_path)
+ _drive_hook(plugin, "mod::test_x", _make_report(failed=False))
+ mgr.collect.assert_not_called()
+
+ def test_skips_on_setup_phase(self, tmp_path: "Path") -> None:
+ mgr = MagicMock()
+ plugin = CoredumpPlugin({"p": mgr})
+ pytest.result_dir = str(tmp_path)
+ _drive_hook(
+ plugin, "mod::test_x", _make_report(when="setup", failed=True)
+ )
+ mgr.collect.assert_not_called()
+
+ def test_skips_on_teardown_phase(self, tmp_path: "Path") -> None:
+ mgr = MagicMock()
+ plugin = CoredumpPlugin({"p": mgr})
+ pytest.result_dir = str(tmp_path)
+ _drive_hook(
+ plugin,
+ "mod::test_x",
+ _make_report(when="teardown", failed=True),
+ )
+ mgr.collect.assert_not_called()
+
+ def test_skips_when_result_dir_empty(self) -> None:
+ mgr = MagicMock()
+ plugin = CoredumpPlugin({"p": mgr})
+ pytest.result_dir = ""
+ _drive_hook(plugin, "mod::test_x", _make_report(failed=True))
+ mgr.collect.assert_not_called()
+
+ def test_skips_when_result_dir_missing(
+ self, monkeypatch: pytest.MonkeyPatch
+ ) -> None:
+ mgr = MagicMock()
+ plugin = CoredumpPlugin({"p": mgr})
+ monkeypatch.delattr(pytest, "result_dir", raising=False)
+ _drive_hook(plugin, "mod::test_x", _make_report(failed=True))
+ mgr.collect.assert_not_called()
diff --git a/tests/pytest/test_hook_ordering.py b/tests/pytest/test_hook_ordering.py
new file mode 100644
index 0000000..098f1a7
--- /dev/null
+++ b/tests/pytest/test_hook_ordering.py
@@ -0,0 +1,90 @@
+############################################################################
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+############################################################################
+
+"""Regression test: CoredumpPlugin's trylast wrapper must exit before a
+plain hookwrapper registered earlier, matching pluggy's registration-order
+semantics (last-registered wrapper is outermost / exits last; trylast
+forces innermost / exits first)."""
+
+import textwrap
+
+import pytest
+
+
+def test_trylast_wrapper_exits_before_plain_wrapper(
+ pytester: pytest.Pytester,
+) -> None:
+ """A trylast hookwrapper must run its post-yield code first, even
+ though it is registered AFTER a plain hookwrapper.
+
+ This mirrors the real bug: ``PytestConfigPlugin`` (plain
+ hookwrapper, registered first / innermost) reboots the device on
+ crash, while ``CoredumpPlugin`` (registered last / outermost) must
+ collect the coredump before that reboot happens. Marking
+ ``CoredumpPlugin``'s hook ``trylast=True`` forces it innermost so
+ its post-yield collection always runs before the plain wrapper's
+ post-yield reboot, regardless of registration order.
+ """
+ pytester.makeconftest(textwrap.dedent("""
+ import pytest
+
+
+ class PlainWrapperFirst:
+ @pytest.hookimpl(hookwrapper=True)
+ def pytest_runtest_makereport(self, item, call):
+ yield
+ if call.when == "call":
+ with open("order.log", "a") as f:
+ f.write("plain_first\\n")
+
+
+ class TrylastWrapperLast:
+ @pytest.hookimpl(hookwrapper=True, trylast=True)
+ def pytest_runtest_makereport(self, item, call):
+ yield
+ if call.when == "call":
+ with open("order.log", "a") as f:
+ f.write("trylast_last\\n")
+
+
+ def pytest_configure(config):
+ # Registration order matches the real bug: the plain
+ # wrapper (ptconfig) is registered first, the trylast
+ # wrapper (CoredumpPlugin) is registered last.
+ config.pluginmanager.register(PlainWrapperFirst())
+ config.pluginmanager.register(TrylastWrapperLast())
+ """))
+ pytester.makepyfile("""
+ def test_something():
+ assert True
+ """)
+ # -p no:sugar: pytest-sugar's reporter replaces the standard "N
+ # passed" summary line with its own progress-bar output whenever
+ # stdout is a real TTY, which breaks assert_outcomes()'s parsing of
+ # this inner session's result regardless of the outer session's own
+ # outcome.
+ result = pytester.runpytest("-p", "no:sugar")
+ result.assert_outcomes(passed=1)
+
+ order_log = pytester.path / "order.log"
+ assert order_log.read_text().splitlines() == [
+ "trylast_last",
+ "plain_first",
+ ]
diff --git a/tests/pytest/test_mypytest.py b/tests/pytest/test_mypytest.py
index 2f91f9b..c8affd5 100644
--- a/tests/pytest/test_mypytest.py
+++ b/tests/pytest/test_mypytest.py
@@ -234,6 +234,86 @@ def test_device_stop_calls_stop(config_dummy, device_dummy):
core._device.stop.assert_called_once()
+def test_runner_wires_debug_plugins(config_dummy, device_dummy, monkeypatch):
+ """runner() opens/closes the debug setup and passes its plugins."""
+ with patch("ntfc.cores.get_device", return_value=device_dummy):
+ p = MyPytest(config_dummy)
+
+ debug_plugin = MagicMock()
+ debug = MagicMock()
+ debug.plugins = [debug_plugin]
+
+ run_plugins = []
+
+ def _run(_opt, plugins):
+ run_plugins.extend(plugins)
+ return 0
+
+ monkeypatch.setattr(p, "_init_pytest", lambda _path: None)
+ monkeypatch.setattr(p, "_device_start", lambda: None)
+ monkeypatch.setattr(p, "_device_stop", lambda: None)
+ monkeypatch.setattr(p, "_run", _run)
+ monkeypatch.setattr(
+ "ntfc.pytest.mypytest.get_debug_setup",
+ lambda _products: debug,
+ )
+
+ import pytest as _pytest
+
+ monkeypatch.setattr(_pytest, "products", [], raising=False)
+
+ assert p.runner("tests", {}, nologs=True) == 0
+
+ assert debug_plugin in run_plugins
+ debug.start.assert_called_once_with("")
+ debug.stop.assert_called_once()
+ assert p._ptconfig._restart_gdb is debug.restart_all_controllers
+
+
+def test_runner_stops_devices_when_debug_start_raises(
+ config_dummy, device_dummy, monkeypatch
+):
+ """A debug setup failure must not leave devices running."""
+ with patch("ntfc.cores.get_device", return_value=device_dummy):
+ p = MyPytest(config_dummy)
+
+ debug = MagicMock()
+ debug.start.side_effect = RuntimeError("gdb exploded")
+
+ stop_mock = MagicMock()
+ monkeypatch.setattr(p, "_init_pytest", lambda _path: None)
+ monkeypatch.setattr(p, "_device_start", lambda: None)
+ monkeypatch.setattr(p, "_device_stop", stop_mock)
+ monkeypatch.setattr(
+ "ntfc.pytest.mypytest.get_debug_setup",
+ lambda _products: debug,
+ )
+
+ import pytest as _pytest
+
+ monkeypatch.setattr(_pytest, "products", [], raising=False)
+
+ with pytest.raises(RuntimeError, match="gdb exploded"):
+ p.runner("tests", {}, nologs=True)
+
+ debug.stop.assert_called_once()
+ stop_mock.assert_called_once()
+
+
+def test_runner_no_debug_plugins_by_default(config_dummy, device_dummy):
+ """Products without a debug section produce no debug plugins."""
+ with patch("ntfc.cores.get_device", return_value=device_dummy):
+ p = MyPytest(config_dummy)
+ path = "./tests/resources/tests_exitcode/test_success.py"
+ assert p.runner(path, {}, nologs=True) == 0
+
+ import pytest as _pytest
+
+ from ntfc.debug.getdebug import get_debug_setup
+
+ assert get_debug_setup(_pytest.products).plugins == []
+
+
def test_write_session_config_file(config_dummy, tmp_path):
p = MyPytest(config_dummy)
diff --git a/tests/test_productconfig.py b/tests/test_productconfig.py
index 3f40abb..f786efe 100644
--- a/tests/test_productconfig.py
+++ b/tests/test_productconfig.py
@@ -244,6 +244,57 @@ def test_product_config_cfg_core():
p.cfg_core(5)
+def test_product_config_debug_defaults():
+ """Test debug property defaults when the section is absent."""
+
+ conf = {
+ "name": "product",
+ "cores": {"core0": {"name": "main", "device": "sim"}},
+ }
+
+ p = ProductConfig(conf)
+ assert p.debug.coredump.enable is False
+ assert p.debug.coredump.collection_type == "auto"
+ assert p.debug.gdb.enable is False
+
+
+def test_product_config_debug_parsing():
+ """Test debug property with a full debug section."""
+
+ conf = {
+ "name": "product",
+ "cores": {"core0": {"name": "main", "device": "sim"}},
+ "debug": {
+ "coredump": {"enable": True, "type": "gdb", "limit": 3},
+ "gdb": {"enable": True, "target": "localhost:1234"},
+ "fastboot": {"dev_sn": "ABC123"},
+ "ymodem": {"serial_port": "/dev/ttyUSB0"},
+ },
+ }
+
+ p = ProductConfig(conf)
+ assert p.debug.coredump.enable is True
+ assert p.debug.coredump.collection_type == "gdb"
+ assert p.debug.coredump.limit == 3
+ assert p.debug.gdb.enable is True
+ assert p.debug.gdb.target == "localhost:1234"
+ assert p.debug.fastboot.dev_sn == "ABC123"
+ assert p.debug.ymodem.serial_port == "/dev/ttyUSB0"
+
+
+def test_product_config_debug_invalid():
+ """Test that an invalid coredump type raises ValueError."""
+
+ conf = {
+ "name": "product",
+ "cores": {"core0": {"name": "main", "device": "sim"}},
+ "debug": {"coredump": {"type": "bogus"}},
+ }
+
+ with pytest.raises(ValueError):
+ ProductConfig(conf)
+
+
def test_product_config_platform_properties():
"""Test platform, is_smp, and is_amp properties."""
diff --git a/tox.ini b/tox.ini
index f45569e..ead1b14 100644
--- a/tox.ini
+++ b/tox.ini
@@ -30,6 +30,7 @@ deps =
pytest-mock
pytest-sugar
coverage>=6
+ lzf
commands =
coverage run -m pytest {posargs}
# ensure 100% coverage of tests
@@ -43,6 +44,7 @@ deps =
pytest-mock
pytest-sugar
pytest-xdist
+ lzf
commands =
pytest -n 4 {posargs}
diff --git a/typings/gdb/__init__.pyi b/typings/gdb/__init__.pyi
new file mode 100644
index 0000000..0094735
--- /dev/null
+++ b/typings/gdb/__init__.pyi
@@ -0,0 +1,62 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Typing stub for the gdb module available inside a GDB process."""
+
+from typing import Any, Callable, List
+
+BP_WATCHPOINT: int
+COMMAND_USER: int
+
+class error(Exception): ...
+
+class Breakpoint:
+ def __init__(
+ self, spec: str, type: int = ..., internal: bool = ...
+ ) -> None: ...
+ def stop(self) -> bool: ...
+ def delete(self) -> None: ...
+ def is_valid(self) -> bool: ...
+
+class Command:
+ def __init__(self, name: str, command_class: int) -> None: ...
+ def invoke(self, args: str, from_tty: bool) -> Any: ...
+
+class Event: ...
+
+class BreakpointEvent(Event):
+ breakpoints: List[Breakpoint]
+
+class SignalEvent(Event):
+ stop_signal: str
+
+class ExitedEvent(Event): ...
+
+class _EventRegistry:
+ def connect(self, callback: Callable[[Any], None]) -> None: ...
+ def disconnect(self, callback: Callable[[Any], None]) -> None: ...
+
+class _Events:
+ stop: _EventRegistry
+ exited: _EventRegistry
+
+events: _Events
+
+def execute(command: str, to_string: bool = ...) -> Any: ...
+def string_to_argv(args: str) -> List[str]: ...