mirror of
https://github.com/espressif/esp-idf.git
synced 2026-09-22 13:01:16 +03:00
249 lines
9.5 KiB
Python
249 lines
9.5 KiB
Python
# SPDX-FileCopyrightText: 2022-2026 Espressif Systems (Shanghai) CO LTD
|
|
# SPDX-License-Identifier: Unlicense OR CC0-1.0
|
|
import os.path
|
|
import re
|
|
import time
|
|
import typing
|
|
|
|
import pexpect
|
|
import pytest
|
|
import serial
|
|
from pytest_embedded_idf import IdfDut
|
|
from pytest_embedded_idf.utils import idf_parametrize
|
|
from pytest_embedded_idf.utils import soc_filtered_targets
|
|
|
|
if typing.TYPE_CHECKING:
|
|
from conftest import OpenOCD
|
|
|
|
|
|
STOP_EVENT_ID = 0x0B # SYSVIEW_EVTID_TRACE_STOP
|
|
|
|
|
|
def _assert_has_stop_record(segment: bytes, label: str) -> None:
|
|
"""Assert a SysView data segment ends with a TRACE_STOP record.
|
|
|
|
A STOP record is the STOP event ID followed by a variable-length timestamp
|
|
delta. Walk back over the trailing continuation bytes (0x80 bit set) to find
|
|
the event ID, since its offset is not fixed.
|
|
"""
|
|
size = len(segment)
|
|
assert size >= 2, f'{label}: segment too small to contain STOP record'
|
|
i = size - 2
|
|
while i >= 0 and (segment[i] & 0x80):
|
|
i -= 1
|
|
assert i >= 0 and segment[i] == STOP_EVENT_ID, f'{label}: does not end with a TRACE_STOP record'
|
|
|
|
|
|
def _split_mcore_file(content: bytes) -> list[bytes]:
|
|
"""Split an ``esp sysview_mcore`` combined file into per-core data segments.
|
|
|
|
Layout is an ASCII comment header followed by core0 data then core1 data::
|
|
|
|
; ...
|
|
; Offset Core0 0
|
|
; Offset Core1 <size of core0 data in bytes>
|
|
;
|
|
<core0 data><core1 data>
|
|
|
|
The header ends at the first line that does not start with ';', (the data
|
|
begins with the all-zero sync sequence).
|
|
"""
|
|
header = re.match(rb'(?:;[^\n]*\n)*', content)
|
|
assert header is not None
|
|
pos = header.end()
|
|
m = re.search(rb';\s*Offset\s+Core1\s+(\d+)', content[:pos])
|
|
assert m is not None, 'mcore file header missing "Offset Core1"'
|
|
offset_core1 = int(m.group(1))
|
|
core0 = content[pos : pos + offset_core1]
|
|
core1 = content[pos + offset_core1 :]
|
|
return [core0, core1]
|
|
|
|
|
|
def _validate_trace_data(trace_log: str, target: str, dual_core: bool = False, is_uart: bool = False) -> None:
|
|
"""Validate SysView trace data in a single trace log file.
|
|
|
|
Args:
|
|
trace_log: Path to the trace log file
|
|
target: Target chip name (e.g., 'esp32', 'esp32s3')
|
|
dual_core: If True, this is an 'esp sysview_mcore' combined file that
|
|
embeds core0 and core1 data segments (see '_split_mcore_file');
|
|
each segment is validated separately.
|
|
is_uart: If True, also validate STOP record at end of file
|
|
"""
|
|
with open(trace_log, 'rb') as f:
|
|
content = f.read()
|
|
|
|
for idx in range(2 if dual_core else 1):
|
|
search_str = f'N=FreeRTOS Application,D={target},C=core{idx},O=FreeRTOS'.encode()
|
|
assert search_str in content, f'SysView core{idx} trace data not found in {trace_log}'
|
|
|
|
if dual_core:
|
|
# Seek to each core's segment via the header offset
|
|
# and require each to end with its own TRACE_STOP record.
|
|
for idx, segment in enumerate(_split_mcore_file(content)):
|
|
_assert_has_stop_record(segment, f'core{idx}')
|
|
else:
|
|
_assert_has_stop_record(content, 'trace')
|
|
|
|
|
|
def _capture_sysview_trace(ser: serial.Serial, trace_log_path: str) -> None:
|
|
"""Capture SysView trace data from serial port.
|
|
|
|
Args:
|
|
ser: Serial port instance
|
|
trace_log_path: Path to save the trace log
|
|
"""
|
|
START_CMD = b'\x01'
|
|
STOP_CMD = b'\x02'
|
|
STOP_EVENT_ID = 0x0B # SYSVIEW_EVTID_TRACE_STOP
|
|
|
|
ser.reset_input_buffer()
|
|
# Send Start command to start SysView tracing
|
|
ser.write(START_CMD)
|
|
|
|
data = bytearray()
|
|
|
|
# Capture for 3 seconds
|
|
end_time = time.time() + 3.0
|
|
while time.time() < end_time:
|
|
try:
|
|
if ser.in_waiting:
|
|
data += ser.read(ser.in_waiting)
|
|
except serial.SerialTimeoutException:
|
|
assert False, 'Timeout reached while reading from serial port, exiting...'
|
|
|
|
# Give pending trace data a short window to reach the host before requesting STOP.
|
|
time.sleep(0.2)
|
|
|
|
# Send Stop command
|
|
ser.write(STOP_CMD)
|
|
|
|
# Capture the final data produced by STOP and the transport flush.
|
|
end_time = time.time() + 3.0
|
|
last_data_time = time.time()
|
|
while time.time() < end_time and (time.time() - last_data_time) <= 1.0:
|
|
try:
|
|
if ser.in_waiting:
|
|
data += ser.read(ser.in_waiting)
|
|
last_data_time = time.time()
|
|
except serial.SerialTimeoutException:
|
|
assert False, 'Timeout reached while reading from serial port, exiting...'
|
|
|
|
# Drop anything after the TRACE_STOP record (e.g. the ROM boot banner printed
|
|
# on the shared UART when pytest-embedded resets the DUT after the test).
|
|
stop_pos = data.rfind(STOP_EVENT_ID)
|
|
if stop_pos != -1:
|
|
end = stop_pos + 1
|
|
# Consume the timestamp delta: continuation bytes have the 0x80 bit set,
|
|
# terminated by a single byte with 0x80 clear.
|
|
while end < len(data) and (data[end] & 0x80):
|
|
end += 1
|
|
if end < len(data):
|
|
end += 1
|
|
data = data[:end]
|
|
|
|
with open(trace_log_path, 'w+b') as f:
|
|
f.write(data)
|
|
|
|
|
|
def _test_sysview_tracing_jtag(openocd_dut: 'OpenOCD', dut: IdfDut) -> None:
|
|
# Single multi-core capture file (esp sysview_mcore): one file is enough for
|
|
# both single- and dual-core targets.
|
|
trace_log = os.path.join(dut.logdir, 'sys_log.svdat') # pylint: disable=protected-access
|
|
dual_core = not dut.app.sdkconfig.get('ESP_SYSTEM_SINGLE_CORE_MODE') or dut.target == 'esp32s3'
|
|
|
|
# Prepare gdbinit file pointing at this run's capture file
|
|
gdb_logfile = os.path.join(dut.logdir, 'gdb.txt')
|
|
gdbinit_orig = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'gdbinit')
|
|
gdbinit = os.path.join(dut.logdir, 'gdbinit')
|
|
with open(gdbinit_orig) as f_r, open(gdbinit, 'w') as f_w:
|
|
for line in f_r:
|
|
if line.startswith('mon esp sysview_mcore start'):
|
|
f_w.write(f'mon esp sysview_mcore start file://{trace_log}\n')
|
|
else:
|
|
f_w.write(line)
|
|
|
|
def dut_expect_task_event() -> None:
|
|
dut.expect(re.compile(rb'example: Task\[0x[0-9A-Fa-f]+\]: received event \d+'), timeout=30)
|
|
|
|
time.sleep(1) # Wait for the USJ port to be ready
|
|
dut.expect_exact('example: Hello from sysview_tracing example!', timeout=5)
|
|
with (
|
|
openocd_dut.run() as openocd,
|
|
open(gdb_logfile, 'w') as gdb_log,
|
|
pexpect.spawn(
|
|
f'idf.py -B {dut.app.binary_path} gdb --batch -x {gdbinit}',
|
|
timeout=60,
|
|
logfile=gdb_log,
|
|
encoding='utf-8',
|
|
codec_errors='ignore',
|
|
) as p,
|
|
):
|
|
p.expect_exact('hit Breakpoint 1, app_main ()')
|
|
dut.expect('example: Created task') # dut has been restarted by gdb since the last dut.expect()
|
|
dut_expect_task_event()
|
|
|
|
# Capture sysview samples and keep reading telnet so OpenOCD's blocking log writes don't stall its
|
|
# main loop and leave the 'stop' below unserviced.
|
|
openocd.consume_output(3)
|
|
openocd.write('esp sysview_mcore stop')
|
|
openocd.apptrace_wait_stop()
|
|
|
|
_validate_trace_data(trace_log, dut.target, dual_core=dual_core)
|
|
|
|
|
|
@pytest.mark.jtag
|
|
@idf_parametrize('config', ['sysview_jtag'], indirect=['config'])
|
|
@idf_parametrize('target', ['esp32', 'esp32c2', 'esp32s2'], indirect=['target'])
|
|
def test_sysview_tracing_jtag(openocd_dut: 'OpenOCD', dut: IdfDut) -> None:
|
|
_test_sysview_tracing_jtag(openocd_dut, dut)
|
|
|
|
|
|
@pytest.mark.usb_serial_jtag
|
|
@idf_parametrize('config', ['sysview_jtag'], indirect=['config'])
|
|
@idf_parametrize(
|
|
'target',
|
|
soc_filtered_targets('SOC_USB_SERIAL_JTAG_SUPPORTED == 1'),
|
|
indirect=['target'],
|
|
)
|
|
@idf_parametrize('port', ['/dev/serial_ports/ttyUSB-esp32'], indirect=['port'])
|
|
def test_sysview_tracing_usj(openocd_dut: 'OpenOCD', dut: IdfDut) -> None:
|
|
_test_sysview_tracing_jtag(openocd_dut, dut)
|
|
|
|
|
|
def _test_sysview_tracing_uart(dut: IdfDut) -> None:
|
|
dut.serial.close()
|
|
time.sleep(2) # Wait for the DUT to reboot
|
|
with serial.Serial(dut.serial.port, baudrate=dut.app.sdkconfig.get('APPTRACE_UART_BAUDRATE'), timeout=3) as ser:
|
|
trace_log = os.path.join(dut.logdir, 'sys_log_uart.svdat') # pylint: disable=protected-access
|
|
_capture_sysview_trace(ser, trace_log)
|
|
_validate_trace_data(trace_log, dut.target, is_uart=True)
|
|
|
|
|
|
@pytest.mark.generic
|
|
@idf_parametrize('config', ['sysview_uart'], indirect=['config'])
|
|
@idf_parametrize('target', ['supported_targets'], indirect=['target'])
|
|
def test_sysview_tracing_uart(dut: IdfDut) -> None:
|
|
_test_sysview_tracing_uart(dut)
|
|
|
|
|
|
@pytest.mark.generic
|
|
@pytest.mark.xtal_26mhz
|
|
@idf_parametrize('config', ['sysview_uart_esp32c2_26Mhz'], indirect=['config'])
|
|
@idf_parametrize('target', ['esp32c2'], indirect=['target'])
|
|
def test_sysview_tracing_uart_c2(dut: IdfDut) -> None:
|
|
_test_sysview_tracing_uart(dut)
|
|
|
|
|
|
@pytest.mark.usb_serial_jtag
|
|
@idf_parametrize('target', soc_filtered_targets('SOC_USB_SERIAL_JTAG_SUPPORTED == 1'), indirect=['target'])
|
|
@pytest.mark.parametrize('config', [pytest.param('sysview_usj')], indirect=True)
|
|
@idf_parametrize('port', ['/dev/serial_ports/ttyUSB-esp32'], indirect=['port'])
|
|
def test_sysview_tracing_usj_serial(dut: IdfDut) -> None:
|
|
time.sleep(1) # wait for USJ port to be ready
|
|
usj_port = '/dev/serial_ports/ttyACM-esp32'
|
|
trace_log = os.path.join(dut.logdir, 'sys_log_usj.svdat') # pylint: disable=protected-access
|
|
with serial.Serial(usj_port, baudrate=1000000, timeout=10) as ser:
|
|
_capture_sysview_trace(ser, trace_log)
|
|
_validate_trace_data(trace_log, dut.target, is_uart=True)
|