diff --git a/pyproject.toml b/pyproject.toml index 4de5438..4f27221 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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" diff --git a/src/mcesptool/components/idf_integration.py b/src/mcesptool/components/idf_integration.py index eaa5a73..f8c26c4 100644 --- a/src/mcesptool/components/idf_integration.py +++ b/src/mcesptool/components/idf_integration.py @@ -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: diff --git a/src/mcesptool/server.py b/src/mcesptool/server.py index b220d25..b52d7f3 100644 --- a/src/mcesptool/server.py +++ b/src/mcesptool/server.py @@ -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 diff --git a/tests/test_idf_integration.py b/tests/test_idf_integration.py index c9572a4..cb55938 100644 --- a/tests/test_idf_integration.py +++ b/tests/test_idf_integration.py @@ -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() diff --git a/uv.lock b/uv.lock index 40eb50b..d25f179 100644 --- a/uv.lock +++ b/uv.lock @@ -888,7 +888,7 @@ wheels = [ [[package]] name = "mcesptool" -version = "2026.2.23" +version = "2026.2.25" source = { editable = "." } dependencies = [ { name = "click" },