Fix v5.x parsers, clean up CLI, bump to 2026.02.25

- Rewrite _parse_tools_list for ESP-IDF v5.x compact format
  (handles both v5.x and older verbose output)
- Archive detection runs before v5.x version matching to avoid
  false positives on filenames like *.tar.gz
- Remove dead --config and --port CLI parameters
- Add 21 new tests: v5.x parser coverage, Tier 2 tool invocations,
  resource/prompt tests (193 total)
This commit is contained in:
Ryan Malloy 2026-03-02 02:16:40 -07:00
parent 44553ebcdb
commit 76ff1ad46a
5 changed files with 493 additions and 30 deletions

View File

@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "mcesptool"
version = "2026.02.23"
version = "2026.02.25"
description = "FastMCP server for ESP32/ESP8266 development with esptool integration"
readme = "README.md"
requires-python = ">=3.10"

View File

@ -68,7 +68,14 @@ def _validate_tool_names(names: list[str]) -> list[str]:
def _parse_tools_list(output: str) -> list[dict[str, Any]]:
"""Parse ``idf_tools.py list`` text output into structured records.
The output format is roughly::
Handles two output formats:
**ESP-IDF v5.x** (compact)::
* xtensa-esp-elf-gdb: GDB for Xtensa
- 14.2_20240403 (recommended, installed)
**Older IDF** (verbose)::
* xtensa-esp-elf-gdb
- Version 14.2_20240403
@ -83,20 +90,32 @@ def _parse_tools_list(output: str) -> list[dict[str, Any]]:
if not stripped:
continue
# Tool header: "* tool-name"
# Tool header: "* tool-name" or "* tool-name: Description"
if stripped.startswith("* "):
if current_tool is not None:
if current_version:
current_tool["versions"].append(current_version)
tools.append(current_tool)
current_tool = {"name": stripped[2:].strip(), "versions": []}
header = stripped[2:].strip()
# v5.x format: "tool-name: Description"
if ": " in header:
name, description = header.split(": ", 1)
else:
name, description = header, ""
current_tool = {
"name": name.strip(),
"description": description.strip(),
"supported_targets": [],
"versions": [],
}
current_version = None
continue
if current_tool is None:
continue
# Version line: "- Version X.Y.Z"
# Version line: "- Version X.Y.Z" (older format)
version_match = re.match(r"^-\s+Version\s+(.+)", stripped)
if version_match:
if current_version:
@ -108,17 +127,47 @@ def _parse_tools_list(output: str) -> list[dict[str, Any]]:
}
continue
# Archive line: "- filename.tar.gz (installed)" or just "- filename"
# Archive line: "- filename.tar.gz (installed)" (older format)
# Must check BEFORE v5.x version lines — archive filenames contain
# file extensions (.tar.gz, .zip, etc.) that distinguish them.
if current_version and stripped.startswith("- "):
archive_part = stripped[2:].strip()
installed = "(installed)" in archive_part
if installed:
current_version["installed"] = True
archive_part = archive_part.replace("(installed)", "").strip()
current_version["archives"].append({
"file": archive_part,
"installed": installed,
})
item_text = stripped[2:].strip()
# Detect archive by file extension in the first token
first_token = item_text.split()[0] if item_text.split() else ""
if re.search(r"\.(tar\.gz|tar\.xz|tar\.bz2|zip|dmg|exe)$", first_token):
installed = "(installed)" in item_text
if installed:
current_version["installed"] = True
item_text = item_text.replace("(installed)", "").strip()
current_version["archives"].append({
"file": item_text,
"installed": installed,
})
continue
# v5.x version line: "- 14.2_20240403 (recommended, installed)"
v5_match = re.match(r"^-\s+(\S+)\s*\(([^)]+)\)", stripped)
if v5_match:
if current_version:
current_tool["versions"].append(current_version)
status_text = v5_match.group(2)
current_version = {
"version": v5_match.group(1),
"installed": "installed" in status_text,
"status": status_text.strip(),
"archives": [],
}
continue
# v5.x version line without status: "- 14.2_20240403"
v5_bare = re.match(r"^-\s+(\S+)\s*$", stripped)
if v5_bare and current_version is None:
current_version = {
"version": v5_bare.group(1),
"installed": False,
"archives": [],
}
continue
# Flush final tool
if current_tool is not None:

View File

@ -126,7 +126,7 @@ class ESPToolServer:
return {
"server_name": "MCP ESPTool Server",
"version": "2026.02.23",
"version": "2026.02.25",
"uptime_seconds": round(uptime, 2),
"configuration": self.config.to_dict(),
"components": list(self.components.keys()),
@ -402,29 +402,23 @@ class ESPToolServer:
# CLI interface
@click.command()
@click.option("--config", "-c", help="Configuration file path")
@click.option("--debug", "-d", is_flag=True, help="Enable debug logging")
@click.option("--production", "-p", is_flag=True, help="Run in production mode")
@click.option("--port", default=8080, help="Server port (for future HTTP interface)")
@click.version_option(version="2026.02.23")
def main(config: str | None, debug: bool, production: bool, port: int) -> None:
@click.version_option(version="2026.02.25")
def main(debug: bool, production: bool) -> None:
"""
FastMCP ESP Development Server
Provides AI-powered ESP32/ESP8266 development workflows through natural language.
Provides ESP32/ESP8266 development workflows through MCP.
Configure via environment variables (see reference docs).
"""
# Configure logging level
if debug:
logging.getLogger().setLevel(logging.DEBUG)
logger.info("🐛 Debug logging enabled")
# Load configuration
if config:
logger.info(f"📁 Loading configuration from: {config}")
# TODO: Implement configuration file loading
server_config = ESPToolServerConfig.from_environment()
else:
server_config = ESPToolServerConfig.from_environment()
# Load configuration from environment
server_config = ESPToolServerConfig.from_environment()
# Override production mode if specified
if production:
@ -434,7 +428,7 @@ def main(config: str | None, debug: bool, production: bool, port: int) -> None:
# Display startup banner
console.print("\n[bold blue]🚀 FastMCP ESP Development Server[/bold blue]")
console.print("[dim]AI-powered ESP32/ESP8266 development workflows[/dim]")
console.print("[dim]Version: 2026.02.23[/dim]")
console.print("[dim]Version: 2026.02.25[/dim]")
console.print()
# Create and run server

View File

@ -166,6 +166,63 @@ class TestParseToolsList:
assert tools[0]["name"] == "lonely-tool"
assert tools[0]["versions"][0]["archives"] == []
# v5.x compact format tests (real ESP-IDF v5.3 output)
SAMPLE_V5 = textwrap.dedent("""\
* xtensa-esp-elf-gdb: GDB for Xtensa
- 14.2_20240403 (recommended, installed)
* riscv32-esp-elf-gdb: GDB for RISC-V
- 14.2_20240403 (recommended, installed)
* xtensa-esp-elf: Toolchain for 32-bit Xtensa based on GCC
- esp-13.2.0_20240530 (recommended, installed)
* cmake: CMake build system (optional)
- 3.24.0 (recommended, installed)
- 3.16.3 (supported)
* qemu-riscv32: QEMU for RISC-V (optional)
- esp_develop_8.2.0_20240122 (recommended)
""")
def test_v5_extracts_tool_names(self):
tools = _parse_tools_list(self.SAMPLE_V5)
names = [t["name"] for t in tools]
assert "xtensa-esp-elf-gdb" in names
assert "riscv32-esp-elf-gdb" in names
assert "cmake" in names
assert "qemu-riscv32" in names
def test_v5_extracts_descriptions(self):
tools = _parse_tools_list(self.SAMPLE_V5)
gdb = next(t for t in tools if t["name"] == "xtensa-esp-elf-gdb")
assert gdb["description"] == "GDB for Xtensa"
def test_v5_tool_count(self):
tools = _parse_tools_list(self.SAMPLE_V5)
assert len(tools) == 5
def test_v5_installed_status(self):
tools = _parse_tools_list(self.SAMPLE_V5)
gdb = next(t for t in tools if t["name"] == "xtensa-esp-elf-gdb")
assert gdb["versions"][0]["installed"] is True
qemu = next(t for t in tools if t["name"] == "qemu-riscv32")
assert qemu["versions"][0]["installed"] is False
def test_v5_version_extracted(self):
tools = _parse_tools_list(self.SAMPLE_V5)
xtensa = next(t for t in tools if t["name"] == "xtensa-esp-elf")
assert xtensa["versions"][0]["version"] == "esp-13.2.0_20240530"
def test_v5_multiple_versions(self):
tools = _parse_tools_list(self.SAMPLE_V5)
cmake = next(t for t in tools if t["name"] == "cmake")
assert len(cmake["versions"]) == 2
assert cmake["versions"][0]["version"] == "3.24.0"
assert cmake["versions"][1]["version"] == "3.16.3"
def test_v5_status_field(self):
tools = _parse_tools_list(self.SAMPLE_V5)
cmake = next(t for t in tools if t["name"] == "cmake")
assert cmake["versions"][0]["status"] == "recommended, installed"
assert cmake["versions"][1]["status"] == "supported"
class TestParseToolsCheck:
"""Tests for _parse_tools_check parser."""
@ -826,3 +883,366 @@ class TestTargetArch:
riscv_targets = {k for k, v in TARGET_ARCH.items() if v == "riscv"}
expected = {"esp32c2", "esp32c3", "esp32c5", "esp32c6", "esp32c61", "esp32h2", "esp32p4"}
assert riscv_targets == expected
# ------------------------------------------------------------------ #
# 5. Mocked invocation tests (tool / resource / prompt functions)
# ------------------------------------------------------------------ #
class TestIdfBuildProject:
"""Test idf_build_project tool function execution (mocked subprocess)."""
@pytest.fixture
def project_dir(self, tmp_path):
"""Create a minimal ESP-IDF project directory with CMakeLists.txt."""
proj = tmp_path / "my_project"
proj.mkdir()
(proj / "CMakeLists.txt").write_text("cmake_minimum_required(VERSION 3.16)\n")
return proj
@pytest.mark.asyncio
async def test_build_validates_target(self, mock_app, config, mock_context, project_dir):
"""Invalid target returns error before any subprocess is spawned."""
IDFIntegration(mock_app, config)
build_fn = mock_app._registered_tools["idf_build_project"]
result = await build_fn(mock_context, project_path=str(project_dir), target="esp8266")
assert result["success"] is False
assert "Unknown target" in result["error"]
@pytest.mark.asyncio
async def test_build_checks_cmakelists_exists(self, mock_app, config, mock_context, tmp_path):
"""Missing CMakeLists.txt returns error."""
empty_dir = tmp_path / "no_cmake"
empty_dir.mkdir()
# Put the directory inside a project root so path validation passes
config.project_roots = [tmp_path]
IDFIntegration(mock_app, config)
build_fn = mock_app._registered_tools["idf_build_project"]
result = await build_fn(mock_context, project_path=str(empty_dir), target="esp32")
assert result["success"] is False
assert "CMakeLists.txt" in result["error"]
@pytest.mark.asyncio
async def test_build_calls_set_target_and_build(
self, mock_app, config, mock_context, project_dir
):
"""Mocks _run_idf_py and verifies set-target + build are called."""
config.project_roots = [project_dir.parent]
integration = IDFIntegration(mock_app, config)
build_fn = mock_app._registered_tools["idf_build_project"]
calls = []
async def fake_run_idf_py(args, timeout=300.0):
calls.append(args)
return {"success": True, "output": "ok", "stderr": ""}
integration._run_idf_py = fake_run_idf_py
result = await build_fn(mock_context, project_path=str(project_dir), target="esp32s3")
assert result["success"] is True
# Should have called set-target then build
assert len(calls) == 2
assert "set-target" in calls[0]
assert "esp32s3" in calls[0]
assert "build" in calls[1]
@pytest.mark.asyncio
async def test_build_clean_runs_fullclean(
self, mock_app, config, mock_context, project_dir
):
"""When clean=True, verifies fullclean is called before set-target and build."""
config.project_roots = [project_dir.parent]
integration = IDFIntegration(mock_app, config)
build_fn = mock_app._registered_tools["idf_build_project"]
calls = []
async def fake_run_idf_py(args, timeout=300.0):
calls.append(args)
return {"success": True, "output": "ok", "stderr": ""}
integration._run_idf_py = fake_run_idf_py
result = await build_fn(
mock_context, project_path=str(project_dir), target="esp32", clean=True,
)
assert result["success"] is True
# First call should be fullclean, then set-target, then build
assert len(calls) == 3
assert "fullclean" in calls[0]
assert "set-target" in calls[1]
assert "build" in calls[2]
@pytest.mark.asyncio
async def test_build_project_path_validation(self, mock_app, config, mock_context, tmp_path):
"""Path outside project_roots is rejected."""
allowed_root = tmp_path / "allowed"
allowed_root.mkdir()
outside_dir = tmp_path / "outside"
outside_dir.mkdir()
(outside_dir / "CMakeLists.txt").write_text("# stub")
config.project_roots = [allowed_root]
# Clear idf_path so it can't be used as fallback
config.esp_idf_path = None
IDFIntegration(mock_app, config)
build_fn = mock_app._registered_tools["idf_build_project"]
result = await build_fn(mock_context, project_path=str(outside_dir), target="esp32")
assert result["success"] is False
assert "outside" in result["error"].lower() or "roots" in result["error"].lower()
class TestIdfFlashProject:
"""Test idf_flash_project tool function execution (mocked subprocess)."""
@pytest.fixture
def built_project(self, tmp_path):
"""Create a project directory that looks like it has been built."""
proj = tmp_path / "built_project"
proj.mkdir()
(proj / "CMakeLists.txt").write_text("# stub")
(proj / "build").mkdir()
return proj
@pytest.mark.asyncio
async def test_flash_validates_baud_rate(self, mock_app, config, mock_context, built_project):
"""Invalid baud rate returns error."""
config.project_roots = [built_project.parent]
IDFIntegration(mock_app, config)
flash_fn = mock_app._registered_tools["idf_flash_project"]
result = await flash_fn(
mock_context,
project_path=str(built_project),
port="/dev/ttyUSB0",
baud=12345,
)
assert result["success"] is False
assert "Invalid baud rate" in result["error"]
@pytest.mark.asyncio
async def test_flash_calls_idf_py_flash(self, mock_app, config, mock_context, built_project):
"""Mocks _run_idf_py and verifies args include --port and flash."""
config.project_roots = [built_project.parent]
integration = IDFIntegration(mock_app, config)
flash_fn = mock_app._registered_tools["idf_flash_project"]
calls = []
async def fake_run_idf_py(args, timeout=300.0):
calls.append(args)
return {"success": True, "output": "flash ok", "stderr": ""}
integration._run_idf_py = fake_run_idf_py
result = await flash_fn(
mock_context,
project_path=str(built_project),
port="/dev/ttyUSB0",
baud=460800,
)
assert result["success"] is True
assert len(calls) == 1
cmd_args = calls[0]
assert "flash" in cmd_args
assert "-p" in cmd_args
assert "/dev/ttyUSB0" in cmd_args
assert "-b" in cmd_args
assert "460800" in cmd_args
@pytest.mark.asyncio
async def test_flash_project_path_validation(self, mock_app, config, mock_context, tmp_path):
"""Path outside project_roots is rejected."""
allowed_root = tmp_path / "allowed"
allowed_root.mkdir()
outside_dir = tmp_path / "outside"
outside_dir.mkdir()
(outside_dir / "build").mkdir()
config.project_roots = [allowed_root]
config.esp_idf_path = None
IDFIntegration(mock_app, config)
flash_fn = mock_app._registered_tools["idf_flash_project"]
result = await flash_fn(
mock_context,
project_path=str(outside_dir),
port="/dev/ttyUSB0",
)
assert result["success"] is False
assert "outside" in result["error"].lower() or "roots" in result["error"].lower()
class TestIdfMonitor:
"""Test idf_monitor tool function execution (mocked subprocess)."""
@pytest.mark.asyncio
async def test_monitor_calls_idf_py(self, mock_app, config, mock_context):
"""Mocks subprocess and verifies monitor command is issued."""
config.get_idf_available = MagicMock(return_value=True)
integration = IDFIntegration(mock_app, config)
monitor_fn = mock_app._registered_tools["idf_monitor"]
# Mock _build_idf_env to return a valid env
integration._build_idf_env = AsyncMock(
return_value={"PATH": "/usr/bin", "IDF_PATH": str(config.esp_idf_path)}
)
mock_proc = AsyncMock()
mock_proc.communicate = AsyncMock(return_value=(b"serial output here", b""))
mock_proc.returncode = 0
with patch("asyncio.create_subprocess_exec", return_value=mock_proc) as mock_exec:
result = await monitor_fn(mock_context, port="/dev/ttyUSB0", duration=5)
assert result["success"] is True
assert result["port"] == "/dev/ttyUSB0"
assert result["duration"] == 5
# Verify the subprocess was called with monitor args
exec_args = mock_exec.call_args[0]
assert "monitor" in exec_args
assert "--no-reset" in exec_args
assert "-p" in exec_args
@pytest.mark.asyncio
async def test_monitor_timeout_cleanup(self, mock_app, config, mock_context):
"""Process is terminated/killed on timeout."""
config.get_idf_available = MagicMock(return_value=True)
integration = IDFIntegration(mock_app, config)
monitor_fn = mock_app._registered_tools["idf_monitor"]
integration._build_idf_env = AsyncMock(
return_value={"PATH": "/usr/bin", "IDF_PATH": str(config.esp_idf_path)}
)
mock_proc = AsyncMock()
# First communicate raises TimeoutError, then terminate+communicate also times out
mock_proc.communicate = AsyncMock(side_effect=asyncio.TimeoutError)
mock_proc.returncode = None
mock_proc.terminate = MagicMock()
mock_proc.kill = MagicMock()
mock_proc.wait = AsyncMock()
with patch("asyncio.create_subprocess_exec", return_value=mock_proc):
with patch("asyncio.wait_for", side_effect=asyncio.TimeoutError):
result = await monitor_fn(mock_context, port="/dev/ttyUSB0", duration=1)
assert result["success"] is True
# Process should have been killed (terminate or kill called)
assert mock_proc.kill.called or mock_proc.terminate.called
class TestIdfStatusResource:
"""Test esp://idf/status resource function."""
@pytest.mark.asyncio
async def test_status_resource_returns_valid_json(self, mock_app, config):
"""Call the registered resource function; verify it returns dict with expected keys."""
integration = IDFIntegration(mock_app, config)
status_fn = mock_app._registered_resources["esp://idf/status"]
# Mock _run_idf_tools so it doesn't actually run subprocess
integration._run_idf_tools = AsyncMock(
return_value={
"success": True,
"output": "xtensa-esp-elf 14.2.0: found\ncmake 3.24.0: not found\n",
"stderr": "",
}
)
# Mock _load_tools_json to skip disk I/O
integration._load_tools_json = AsyncMock(return_value=None)
# Write a version.txt so the resource can read it
version_file = config.esp_idf_path / "version.txt"
version_file.write_text("v5.3\n")
result = await status_fn()
assert isinstance(result, dict)
assert result["available"] is True
assert "idf_path" in result
assert "idf_version" in result
assert result["idf_version"] == "v5.3"
assert "installed_tools" in result
assert "missing_tools" in result
assert "missing_tool_names" in result
@pytest.mark.asyncio
async def test_status_with_no_idf_path(self, mock_app, config):
"""config.esp_idf_path = None returns graceful response."""
config.esp_idf_path = None
IDFIntegration(mock_app, config)
status_fn = mock_app._registered_resources["esp://idf/status"]
result = await status_fn()
assert isinstance(result, dict)
assert result["available"] is False
assert "error" in result
class TestIdfSetupTargetPrompt:
"""Test idf_setup_target prompt function."""
@pytest.mark.asyncio
async def test_prompt_returns_markdown(self, mock_app, config):
"""Call registered prompt with target='esp32p4'; verify non-empty output."""
integration = IDFIntegration(mock_app, config)
prompt_fn = mock_app._registered_prompts["idf_setup_target"]
# Mock _run_idf_tools to return something parseable
integration._run_idf_tools = AsyncMock(
return_value={
"success": True,
"output": "riscv32-esp-elf 14.2.0: found\ncmake 3.24.0: found\n",
"stderr": "",
}
)
integration._load_tools_json = AsyncMock(return_value={
"tools": [
{
"name": "riscv32-esp-elf",
"description": "RISC-V compiler",
"supported_targets": ["esp32c3", "esp32c6", "esp32h2", "esp32p4"],
"versions": [{"name": "14.2.0", "status": "recommended"}],
},
]
})
result = await prompt_fn(target="esp32p4")
assert isinstance(result, str)
assert len(result) > 0
assert "esp32p4" in result
assert "riscv" in result.lower()
@pytest.mark.asyncio
async def test_prompt_invalid_target(self, mock_app, config):
"""Call with bad target; verify architecture shows 'unknown'."""
integration = IDFIntegration(mock_app, config)
prompt_fn = mock_app._registered_prompts["idf_setup_target"]
# Mock _run_idf_tools -- the prompt still runs check even for unknown targets
integration._run_idf_tools = AsyncMock(
return_value={
"success": True,
"output": "",
"stderr": "",
}
)
integration._load_tools_json = AsyncMock(return_value={"tools": []})
result = await prompt_fn(target="badchip")
assert isinstance(result, str)
assert "badchip" in result
assert "unknown" in result.lower()

2
uv.lock generated
View File

@ -888,7 +888,7 @@ wheels = [
[[package]]
name = "mcesptool"
version = "2026.2.23"
version = "2026.2.25"
source = { editable = "." }
dependencies = [
{ name = "click" },