fix(ble_log_console): exclude from idf-ci pytest collection to fix CI

idf-ci discovers pyproject.toml pytest config and runs a separate
collection pass with --target all, which finds 0 target tests and
raises RuntimeError. Add to .idf_ci.toml exclude_dirs instead of
deleting tests. Restore all 227 unit tests with lint/mypy fixes.
This commit is contained in:
Zhou Xiao
2026-03-24 12:13:30 +08:00
parent a1f610ce9b
commit 37dc0826e6
12 changed files with 76 additions and 70 deletions

View File

@@ -50,7 +50,7 @@ menu "BT Logs"
bool "Enable Controller logs"
default n
choice
choice BT_LE_CONTROLLER_LOG_OUTPUT_MODE
depends on BT_LE_CONTROLLER_LOG_ENABLED
prompt "Controller log output mode"
default BT_LE_CONTROLLER_LOG_MODE_BLE_LOG_V2 if BLE_LOG_ENABLED

View File

@@ -232,6 +232,24 @@ if BLE_LOG_ENABLED
menu "Settings of BLE Log Compression"
source "$IDF_PATH/components/bt/common/ble_log/extension/log_compression/Kconfig.in"
endmenu
# Deprecated options -- retained for backward sdkconfig compatibility.
# These symbols have no prompt so they never appear in menuconfig.
config BLE_LOG_LBM_TRANS_SIZE
int
default 512
config BLE_LOG_ENH_STAT_ENABLED
bool
default y
config BLE_LOG_PAYLOAD_CHECKSUM_ENABLED
bool
default y
config BLE_LOG_LBM_LL_TRANS_SIZE
int
default 512
endif
menu "Legacy SPI Log Output (Deprecated - use BT Log Async Output instead)"

View File

@@ -1,8 +1,6 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
"""Thin composition of stats sub-modules into a single accumulator."""
from __future__ import annotations
from src.backend.models import BleLogSource
@@ -17,13 +15,13 @@ from src.backend.models import ThroughputInfo
from src.backend.stats.buf_util import BufUtilTracker
from src.backend.stats.firmware_loss import FirmwareLossTracker
from src.backend.stats.firmware_written import FirmwareWrittenTracker
from src.backend.stats.peak_burst import WRITE_RATE_WINDOW_MS
from src.backend.stats.peak_burst import PeakBurstTracker
from src.backend.stats.peak_burst import WRITE_RATE_WINDOW_MS
from src.backend.stats.sn_gap import SNGapTracker
from src.backend.stats.traffic_spike import TrafficSpikeDetector
from src.backend.stats.traffic_spike import TrafficSpikeResult
from src.backend.stats.transport import UART_BITS_PER_BYTE
from src.backend.stats.transport import TransportMetrics
from src.backend.stats.transport import UART_BITS_PER_BYTE
_ZERO = FrameByteCount(frames=0, bytes=0)
@@ -124,7 +122,7 @@ class StatsAccumulator:
self._enh_stat_prev[src_code] = (written_frames, lost_frames, written_bytes, lost_bytes)
self._fw_written.record(src_code, written_frames, written_bytes)
return self._fw_loss.record(src_code, lost_frames, lost_bytes)
return self._fw_loss.record(src_code, lost_frames, lost_bytes) # type: ignore[no-any-return]
# -- Reset -------------------------------------------------------------------

View File

@@ -1,24 +1,21 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
"""Status panel widget — docked to bottom, shows live stats.
See Spec Section 11.
"""
from rich.text import Text
from src.backend.models import format_bytes
from src.backend.models import format_throughput
from src.backend.models import FrameStats
from src.backend.models import SyncState
from src.backend.stats import UART_BITS_PER_BYTE
from textual.reactive import reactive
from textual.widget import Widget
from src.backend.models import FrameStats
from src.backend.models import SyncState
from src.backend.models import format_bytes
from src.backend.models import format_throughput
from src.backend.stats import UART_BITS_PER_BYTE
def _format_speed(bps: float) -> str:
return format_throughput(bps / UART_BITS_PER_BYTE)
return format_throughput(bps / UART_BITS_PER_BYTE) # type: ignore[no-any-return]
_SYNC_COLORS = {

View File

@@ -1,22 +1,20 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
from src.backend.checksum import sum_checksum
from src.backend.checksum import xor_checksum
from src.backend.frame_parser import FrameParser
from src.backend.models import ChecksumAlgorithm
from src.backend.models import ChecksumScope
from src.backend.models import SyncState
from tests.helpers import build_frame
def _make_sum_frame(payload: bytes, src: int, sn: int) -> bytes:
return build_frame(payload, src, sn, sum_checksum, checksum_scope_full=True)
return build_frame(payload, src, sn, sum_checksum, checksum_scope_full=True) # type: ignore[no-any-return]
def _make_xor_frame(payload: bytes, src: int, sn: int) -> bytes:
return build_frame(payload, src, sn, xor_checksum, checksum_scope_full=True)
return build_frame(payload, src, sn, xor_checksum, checksum_scope_full=True) # type: ignore[no-any-return]
class TestFrameParserStateTransitions:
@@ -138,11 +136,11 @@ class TestFrameParserOutput:
def _make_sum_header_only_frame(payload: bytes, src: int, sn: int) -> bytes:
return build_frame(payload, src, sn, sum_checksum, checksum_scope_full=False)
return build_frame(payload, src, sn, sum_checksum, checksum_scope_full=False) # type: ignore[no-any-return]
def _make_xor_header_only_frame(payload: bytes, src: int, sn: int) -> bytes:
return build_frame(payload, src, sn, xor_checksum, checksum_scope_full=False)
return build_frame(payload, src, sn, xor_checksum, checksum_scope_full=False) # type: ignore[no-any-return]
class TestChecksumAutoDetection:

View File

@@ -1,24 +1,23 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
from src.backend.models import FrameByteCount
from src.backend.models import FunnelSnapshot
from src.backend.models import LossType
from src.backend.models import ThroughputInfo
def test_frame_byte_count():
def test_frame_byte_count() -> None:
fbc = FrameByteCount(frames=100, bytes=5000)
assert fbc.frames == 100
assert fbc.bytes == 5000
def test_loss_type_enum():
def test_loss_type_enum() -> None:
assert LossType.BUFFER == 'buffer'
assert LossType.TRANSPORT == 'transport'
def test_funnel_snapshot_structure():
def test_funnel_snapshot_structure() -> None:
zero = FrameByteCount(frames=0, bytes=0)
tp = ThroughputInfo(
throughput_fps=0.0, throughput_bps=0.0, peak_write_frames=0, peak_write_bytes=0, peak_window_ms=10

View File

@@ -1,18 +1,16 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
"""Reset propagation matrix tests.
Verifies that reset("init") and reset("flush") dispatch correctly per the spec:
| Group | Components | INIT_DONE | FLUSH |
|------------------|---------------------------------------------|--------------------|------------------------------------|
| SN-coupled | SNGapTracker | full reset | full reset |
| ENH_STAT-coupled | FirmwareLossTracker, FirmwareWrittenTracker | full reset | reset baselines, preserve accumulators |
| Console-local | TransportMetrics, PeakBurstTracker, | preserve | preserve |
| | per_source_received, throughput cache | | |
| Group | Components | INIT_DONE | FLUSH |
|------------------|-----------------------------------------|--------------|------------------------------------|
| SN-coupled | SNGapTracker | full reset | full reset |
| ENH_STAT-coupled | FirmwareLossTracker, FirmwareWritten | full reset | reset baselines, keep accumulators |
| Console-local | TransportMetrics, PeakBurstTracker, | preserve | preserve |
| | per_source_received, throughput cache | | |
"""
from src.backend.stats import StatsAccumulator

View File

@@ -1,25 +1,24 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
from src.backend.stats.sn_gap import SNGapTracker
class TestSNGapTracker:
def setup_method(self):
def setup_method(self) -> None:
self.tracker = SNGapTracker()
# --- Baseline ---
def test_first_frame_establishes_baseline(self):
def test_first_frame_establishes_baseline(self) -> None:
assert self.tracker.record(src_code=1, frame_sn=42) == 0
# --- In-order ---
def test_sequential_no_gap(self):
def test_sequential_no_gap(self) -> None:
self.tracker.record(1, 0)
assert self.tracker.record(1, 1) == 0
assert self.tracker.record(1, 2) == 0
# --- Simple reorder (within window) ---
def test_reorder_no_false_gap(self):
def test_reorder_no_false_gap(self) -> None:
"""SN=8 arrives before SN=5,6,7 — no gaps should be counted."""
self.tracker.record(1, 5) # baseline → window_base=6
assert self.tracker.record(1, 8) == 0 # within window, NOT a gap
@@ -28,7 +27,7 @@ class TestSNGapTracker:
assert self.tracker.totals().get(1, 0) == 0
# --- Confirmed loss ---
def test_loss_confirmed_when_window_expires(self):
def test_loss_confirmed_when_window_expires(self) -> None:
"""Frame beyond window forces expiry of unreceived SNs."""
self.tracker.record(1, 0) # baseline → base=1
# SN=1 never arrives; jump to SN=257 (beyond window of 256)
@@ -37,13 +36,13 @@ class TestSNGapTracker:
assert self.tracker.totals()[1] > 0
# --- Late arrival behind window ---
def test_late_arrival_ignored(self):
def test_late_arrival_ignored(self) -> None:
self.tracker.record(1, 0)
self.tracker.record(1, 257) # force window advance past 0
assert self.tracker.record(1, 1) == 0 # too late, ignored
# --- Reset detection ---
def test_large_backward_jump_resets_baseline(self):
def test_large_backward_jump_resets_baseline(self) -> None:
self.tracker.record(1, 1000)
# SN jumps back to 5 (far beyond REORDER_WINDOW backward)
assert self.tracker.record(1, 5) == 0
@@ -51,14 +50,14 @@ class TestSNGapTracker:
assert self.tracker.record(1, 6) == 0
# --- Multi-source independence ---
def test_sources_independent(self):
def test_sources_independent(self) -> None:
self.tracker.record(1, 10)
self.tracker.record(2, 20)
assert self.tracker.record(1, 11) == 0
assert self.tracker.record(2, 21) == 0
# --- 24-bit wraparound ---
def test_wraparound(self):
def test_wraparound(self) -> None:
SN_MAX = 1 << 24
self.tracker.record(1, SN_MAX - 2) # base = SN_MAX-1
assert self.tracker.record(1, SN_MAX - 1) == 0
@@ -66,13 +65,13 @@ class TestSNGapTracker:
assert self.tracker.record(1, 1) == 0
# --- Reset method ---
def test_reset_clears_all(self):
def test_reset_clears_all(self) -> None:
self.tracker.record(1, 10)
self.tracker.reset()
# After reset, next frame establishes new baseline
assert self.tracker.record(1, 0) == 0
def test_reset_single_source(self):
def test_reset_single_source(self) -> None:
self.tracker.record(1, 10)
self.tracker.record(2, 20)
self.tracker.reset(src_code=1)

View File

@@ -1,16 +1,16 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
from typing import Optional
from unittest.mock import patch
from src.backend.models import BleLogSource
from src.backend.models import has_os_ts
from src.backend.stats import StatsAccumulator
from src.backend.stats import TRAFFIC_ALERT_COOLDOWN_SEC
from src.backend.stats import TRAFFIC_THRESHOLD_PCT
from src.backend.stats import TRAFFIC_WINDOW_SEC
from src.backend.stats import WRITE_RATE_WINDOW_MS
from src.backend.stats import StatsAccumulator
from src.backend.stats import TrafficSpikeResult
from src.backend.stats import WRITE_RATE_WINDOW_MS
from src.backend.stats.peak_burst import _ts_delta_ms
# Convenience: default frame size used in peak write tests (arbitrary but consistent)
@@ -55,7 +55,7 @@ class TestStatsAccumulator:
self, stats: StatsAccumulator, src_code: int, lost_frames: int, lost_bytes: int
) -> tuple[int, int]:
"""Helper: call record_enh_stat with dummy written/baudrate, return loss delta."""
return stats.record_enh_stat(
return stats.record_enh_stat( # type: ignore[no-any-return]
src_code=src_code,
written_frames=0,
lost_frames=lost_frames,
@@ -839,7 +839,7 @@ class TestTrafficSpikeDetection:
def _trigger_spike(
self, stats: StatsAccumulator, mock_time: object, t: float, hot_bytes: int, src: int = 1
) -> TrafficSpikeResult | None:
) -> Optional[TrafficSpikeResult]:
"""Helper: inject traffic, enter spike, then exit and return result."""
mock_time.perf_counter.return_value = t # type: ignore[attr-defined]
stats.record_frame_traffic(hot_bytes, src)

View File

@@ -1,10 +1,9 @@
# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
# SPDX-License-Identifier: Apache-2.0
from src.backend.models import format_throughput
from src.backend.models import FrameByteCount
from src.backend.models import FunnelSnapshot
from src.backend.models import ThroughputInfo
from src.backend.models import format_throughput
from src.backend.stats import StatsAccumulator
from src.frontend.stats_screen import _build_console_table
from src.frontend.stats_screen import _build_firmware_table
@@ -19,15 +18,15 @@ _ZERO_TP = ThroughputInfo(
def _snap(
src,
produced=(0, 0),
written=(0, 0),
received=(0, 0),
buf_loss=(0, 0),
tx_loss=(0, 0),
tp_fps=0.0,
peak_frames=0,
):
src: int,
produced: tuple[int, int] = (0, 0),
written: tuple[int, int] = (0, 0),
received: tuple[int, int] = (0, 0),
buf_loss: tuple[int, int] = (0, 0),
tx_loss: tuple[int, int] = (0, 0),
tp_fps: float = 0.0,
peak_frames: int = 0,
) -> FunnelSnapshot:
return FunnelSnapshot(
source=src,
produced=FrameByteCount(*produced),
@@ -85,29 +84,29 @@ class TestFormatThroughput:
class TestBuildFirmwareTable:
def test_empty_returns_no_rows(self):
def test_empty_returns_no_rows(self) -> None:
table = _build_firmware_table([])
assert table.row_count == 0
def test_column_headers(self):
def test_column_headers(self) -> None:
table = _build_firmware_table([])
headers = [str(col.header) for col in table.columns]
assert 'Source' in headers
assert any('Written' in h for h in headers)
assert any('Loss' in h for h in headers)
def test_single_source(self):
def test_single_source(self) -> None:
snap = _snap(_SRC_HOST, written=(120, 6000))
table = _build_firmware_table([snap])
assert table.row_count == 1
assert len(table.columns) == 5
def test_with_loss_shows_red(self):
def test_with_loss_shows_red(self) -> None:
snap = _snap(_SRC_HOST, written=(110, 5500), buf_loss=(10, 500))
table = _build_firmware_table([snap])
assert table.row_count == 1
def test_multiple_sources(self):
def test_multiple_sources(self) -> None:
snaps = [
_snap(_SRC_HOST, written=(100, 5000)),
_snap(_SRC_LL_TASK, written=(200, 10000)),
@@ -117,11 +116,11 @@ class TestBuildFirmwareTable:
class TestBuildConsoleTable:
def test_empty_returns_no_rows(self):
def test_empty_returns_no_rows(self) -> None:
table = _build_console_table([])
assert table.row_count == 0
def test_column_headers(self):
def test_column_headers(self) -> None:
table = _build_console_table([])
headers = [str(col.header) for col in table.columns]
assert 'Source' in headers
@@ -129,13 +128,13 @@ class TestBuildConsoleTable:
assert any('Average' in h for h in headers)
assert any('Peak' in h for h in headers)
def test_single_source(self):
def test_single_source(self) -> None:
snap = _snap(_SRC_HOST, tp_fps=850.0, peak_frames=12)
table = _build_console_table([snap])
assert table.row_count == 1
assert len(table.columns) == 7
def test_zero_throughput_shows_dash(self):
def test_zero_throughput_shows_dash(self) -> None:
snap = _snap(_SRC_HOST, tp_fps=0.0, peak_frames=0)
table = _build_console_table([snap])
assert table.row_count == 1