Trunk and semgrep fixes

This commit is contained in:
Ben Meadors
2026-04-17 18:20:29 -05:00
parent 405c8b951b
commit 0c112a9ea6
20 changed files with 85 additions and 50 deletions
@@ -21,7 +21,6 @@ def test_owner_survives_reboot(
) -> None:
"""Runs once per connected role — proves the reboot-persistence
round-trip works on each device independently, not just one."""
role = baked_single["role"]
port = baked_single["port"]
pre = info.device_info(port=port, timeout_s=8.0)
+10 -16
View File
@@ -35,7 +35,10 @@ import pytest
# Ensure the MCP server is on `sys.path` without requiring installation in
# development mode for every checkout (we DO install in .venv but this makes
# `pytest tests/` work from a fresh clone too).
# `pytest tests/` work from a fresh clone too). The path mutation must
# happen before `meshtastic_mcp.*` imports below — hence the `noqa: E402`
# markers on those imports (ruff's "module-level import not at top of file"
# rule doesn't understand path-bootstrapping patterns).
_HERE = pathlib.Path(__file__).resolve().parent
_MCP_SRC = _HERE.parent / "src"
if str(_MCP_SRC) not in sys.path:
@@ -44,18 +47,9 @@ if str(_MCP_SRC) not in sys.path:
# Default firmware root: the repo this mcp-server/ lives inside.
os.environ.setdefault("MESHTASTIC_FIRMWARE_ROOT", str(_HERE.parent.parent))
from meshtastic_mcp import (
admin,
)
from meshtastic_mcp import ( # noqa: E402 (import after path setup)
devices as devices_module,
)
from meshtastic_mcp import (
flash,
info,
serial_session,
userprefs,
)
from meshtastic_mcp import admin # noqa: E402
from meshtastic_mcp import devices as devices_module # noqa: E402
from meshtastic_mcp import info, serial_session, userprefs # noqa: E402
from . import tool_coverage # noqa: E402
@@ -407,7 +401,7 @@ def baked_mesh(
session_seed: str,
request: pytest.FixtureRequest,
) -> dict[str, Any]:
"""Verifies that both roles are baked with the session `test_profile`.
"""Verify that both roles are baked with the session `test_profile`.
Does NOT reflash. `test_00_bake.py` is responsible for applying the bake;
this fixture just checks the result by connecting to each device and
@@ -696,7 +690,7 @@ class _SerialCapture:
@pytest.fixture
def serial_capture(hub_devices: dict[str, str], request: pytest.FixtureRequest) -> Any:
"""Returns a `_SerialCapture` factory.
"""Return a `_SerialCapture` factory.
Usage:
cap = serial_capture("esp32s3")
@@ -929,7 +923,7 @@ def pytest_runtest_makereport(item: pytest.Item, call: pytest.CallInfo[Any]) ->
# Attach serial captures (if the test used `serial_capture`)
caps = getattr(item, "_serial_captures", None)
if caps:
for i, cap in enumerate(caps):
for cap in caps:
try:
lines = _run_with_timeout(lambda c=cap: c.snapshot(max_lines=2000), 5.0)
except Exception as exc:
+6 -3
View File
@@ -101,13 +101,16 @@ class ReceiveCollector:
pass
def snapshot(self) -> list[dict[str, Any]]:
"""Current list of collected packets (thread-safe copy)."""
"""Return a thread-safe copy of the list of collected packets."""
with self._lock:
return list(self._packets)
def log_snapshot(self) -> list[str]:
"""Captured firmware log lines (only populated if `capture_logs=True`
AND the device has `security.debug_log_api_enabled=True`)."""
"""Return captured firmware log lines.
Only populated if `capture_logs=True` AND the device has
`security.debug_log_api_enabled=True`.
"""
with self._lock:
return list(self._log_lines)
@@ -86,7 +86,7 @@ def test_direct_with_ack_roundtrip(
# sufficient on its own with fresh keys; the retry is purely
# an airtime-collision safety net.
got = None
for attempt in range(2):
for _attempt in range(2):
packet = tx_iface.sendText(
unique,
destinationId=rx_node_num,
@@ -11,7 +11,6 @@ from __future__ import annotations
from typing import Any
import pytest
from meshtastic_mcp import info
from meshtastic_mcp.connection import connect
@@ -7,8 +7,6 @@ misconfig), this test fails within a minute of reboot.
from __future__ import annotations
import os
import re
import time
from typing import Any
+8 -5
View File
@@ -14,7 +14,7 @@ from __future__ import annotations
import json
import pathlib
from typing import Any, Callable
from typing import Any
_counts: dict[str, int] = {}
_installed = False
@@ -108,14 +108,17 @@ def install() -> None:
for tool_name, (module_path, attr) in _TOOL_MAP.items():
# Defense in depth: if someone mutates `_TOOL_MAP` at runtime
# (shouldn't happen; it's module-level) the whitelist catches it.
# `module_path` is a key from the hardcoded `_TOOL_MAP` dict and
# is gated above by membership in `_allowed_modules` (itself
# derived from the same literal values). There is no path for
# untrusted input to reach the `import_module` call below; the
# Semgrep suppression must sit on the line immediately preceding
# the call (multi-line comment blocks between comment and call
# break the rule's scope detection).
if module_path not in _allowed_modules:
continue
try:
# nosemgrep: python.lang.security.audit.non-literal-import.non-literal-import
# `module_path` is a key from the hardcoded `_TOOL_MAP` dict
# and is gated above by membership in `_allowed_modules`
# (itself derived from the same literal values). There is no
# path for untrusted input to reach this call.
mod = importlib.import_module(module_path)
except ImportError:
continue