diff --git a/esphome/__main__.py b/esphome/__main__.py index 225a18cc1c..72f0ff34e2 100644 --- a/esphome/__main__.py +++ b/esphome/__main__.py @@ -58,7 +58,8 @@ from esphome.helpers import get_bool_env, indent, is_ip_address from esphome.log import AnsiFore, color, setup_log from esphome.types import ConfigType from esphome.util import ( - get_rp2040_mass_storage_volumes, + detect_rp2040_bootsel, + get_picotool_path, get_serial_ports, list_yaml_files, run_external_command, @@ -175,7 +176,7 @@ class PortType(StrEnum): NETWORK = "NETWORK" MQTT = "MQTT" MQTTIP = "MQTTIP" - MASS_STORAGE = "MASS_STORAGE" + BOOTSEL = "BOOTSEL" # Magic MQTT port types that require special handling @@ -254,14 +255,14 @@ def choose_upload_log_host( (f"{port.path} ({port.description})", port.path) for port in get_serial_ports() ] - # Add RP2040 mass storage volumes when uploading + # Add RP2040 BOOTSEL device option when uploading if ( purpose == Purpose.UPLOADING and CORE.data.get(KEY_CORE, {}).get(KEY_TARGET_PLATFORM) == PLATFORM_RP2040 ): - for vol in get_rp2040_mass_storage_volumes(): - # Use MS: prefix so get_port_type() identifies as MASS_STORAGE - options.append((f"{vol.path} ({vol.description})", f"MS:{vol.path}")) + picotool = _find_picotool() + if picotool is not None and detect_rp2040_bootsel(picotool) > 0: + options.append(("RP2040 BOOTSEL (via picotool)", "BOOTSEL")) if purpose == Purpose.LOGGING: if has_mqtt_logging(): @@ -285,7 +286,7 @@ def choose_upload_log_host( purpose == Purpose.UPLOADING and CORE.data.get(KEY_CORE, {}).get(KEY_TARGET_PLATFORM) == PLATFORM_RP2040 and not any( - get_port_type(opt[1]) in (PortType.SERIAL, PortType.MASS_STORAGE) + get_port_type(opt[1]) in (PortType.SERIAL, PortType.BOOTSEL) for opt in options ) ): @@ -441,13 +442,13 @@ def get_port_type(port: str) -> PortType: Returns: PortType.SERIAL for serial ports (/dev/ttyUSB0, COM1, etc.) - PortType.MASS_STORAGE for RP2040 BOOTSEL mass storage volumes + PortType.BOOTSEL for RP2040 BOOTSEL upload via picotool PortType.MQTT for MQTT logging PortType.MQTTIP for MQTT IP lookup PortType.NETWORK for IP addresses, hostnames, or mDNS names """ - if port.startswith("MS:"): - return PortType.MASS_STORAGE + if port == "BOOTSEL": + return PortType.BOOTSEL if port.startswith("/") or port.startswith("COM"): return PortType.SERIAL if port == "MQTT": @@ -757,67 +758,80 @@ def upload_using_platformio(config: ConfigType, port: str) -> int: return platformio_api.run_platformio_cli_run(config, CORE.verbose, *upload_args) -def upload_using_uf2_copy(config: ConfigType, mount_path: str) -> int: - """Upload firmware to RP2040 by copying UF2 file to mass storage volume. +def _find_picotool() -> Path | None: + """Find the picotool binary from PlatformIO packages.""" + from esphome import platformio_api - When an RP2040 is in BOOTSEL mode, it appears as a USB mass storage device. - Firmware can be uploaded by simply copying the .uf2 file to the volume. + try: + idedata = platformio_api.get_idedata(CORE.config) + except Exception: # noqa: BLE001 + return None + return get_picotool_path(idedata.cc_path) + + +def upload_using_picotool(config: ConfigType) -> int: + """Upload firmware to RP2040 in BOOTSEL mode using picotool. + + Uses picotool to load the ELF firmware directly via USB, avoiding + the mass storage copy approach that causes "disk not ejected properly" + warnings on macOS. """ + import subprocess + from esphome import platformio_api - from esphome.helpers import ProgressBar idedata = platformio_api.get_idedata(config) - build_dir = Path(idedata.firmware_elf_path).parent - uf2_file = build_dir / "firmware.uf2" + firmware_elf = Path(idedata.firmware_elf_path) - if not uf2_file.exists(): + if not firmware_elf.is_file(): _LOGGER.error( - "UF2 firmware file not found at %s. Make sure the project has been compiled first.", - uf2_file, + "Firmware ELF file not found at %s. " + "Make sure the project has been compiled first.", + firmware_elf, ) return 1 - dest_dir = Path(mount_path) - if not dest_dir.is_dir(): + picotool = get_picotool_path(idedata.cc_path) + if picotool is None: _LOGGER.error( - "Mass storage volume %s is no longer available. " - "Is the RP2040 still in BOOTSEL mode?", - mount_path, + "picotool not found. Ensure the RP2040 PlatformIO platform " + "is installed (tool-picotool-rp2040-earlephilhower)." ) return 1 - dest_file = dest_dir / uf2_file.name - file_size = uf2_file.stat().st_size - if file_size == 0: - _LOGGER.error("UF2 firmware file is empty: %s", uf2_file) - return 1 - _LOGGER.info("Uploading UF2 firmware to %s (%s bytes)", mount_path, file_size) - - progress = ProgressBar() + _LOGGER.info("Uploading firmware to RP2040 via picotool...") try: - chunk_size = 65536 - bytes_written = 0 - with open(uf2_file, "rb") as src, open(dest_file, "wb") as dst: - while True: - chunk = src.read(chunk_size) - if not chunk: - break - dst.write(chunk) - dst.flush() - os.fsync(dst.fileno()) - bytes_written += len(chunk) - progress.update(bytes_written / file_size) - progress.done() + # Don't capture stdout — let picotool write directly to the terminal + # so progress bars display in real-time with \r updates. + # Capture stderr only so we can detect permission errors. + result = subprocess.run( + [str(picotool), "load", "-v", "-x", str(firmware_elf)], + stderr=subprocess.PIPE, + timeout=60, + check=False, + ) + except subprocess.TimeoutExpired: + _LOGGER.error("picotool upload timed out after 60 seconds.") + return 1 except OSError as err: - progress.done() - _LOGGER.error("Failed to copy UF2 file to %s: %s", mount_path, err) + _LOGGER.error("Failed to run picotool: %s", err) + return 1 + + if result.returncode != 0: + stderr = result.stderr.decode("utf-8", errors="replace").strip() + if stderr: + for line in stderr.splitlines(): + safe_print(line) + if "LIBUSB_ERROR_ACCESS" in stderr or "Permission denied" in stderr: + _LOGGER.error( + "Permission denied accessing USB device. " + "On Linux, you may need to add udev rules for RP2040 devices. " + "See: https://github.com/raspberrypi/picotool#linux-permissions" + ) + else: + _LOGGER.error("picotool upload failed (exit code %d).", result.returncode) return 1 - _LOGGER.info( - "Successfully copied firmware to %s. " - "The device will automatically reset and run the new firmware.", - mount_path, - ) return 0 @@ -889,11 +903,9 @@ def upload_program( port_type = get_port_type(host) - if port_type == PortType.MASS_STORAGE: - # Strip the MS: prefix to get the actual mount path - mount_path = host[3:] - exit_code = upload_using_uf2_copy(config, mount_path) - # Return None for device - mass storage can't be used for logging, + if port_type == PortType.BOOTSEL: + exit_code = upload_using_picotool(config) + # Return None for device - BOOTSEL can't be used for logging, # so command_run will show the interactive chooser for log source return exit_code, None @@ -1103,7 +1115,7 @@ def command_run(args: ArgsProtocol, config: ConfigType) -> int | None: if args.no_logs: return 0 - # After mass storage upload, wait for a new serial port to appear + # After BOOTSEL upload, wait for a new serial port to appear # so it shows up in the log chooser if ( successful_device is None diff --git a/esphome/util.py b/esphome/util.py index b1314f0518..695c83cab3 100644 --- a/esphome/util.py +++ b/esphome/util.py @@ -355,66 +355,40 @@ def get_serial_ports() -> list[SerialPort]: return result -class MassStorageVolume: - """Represents a mass storage volume for RP2040 BOOTSEL upload.""" +def get_picotool_path(cc_path: str) -> Path | None: + """Derive the picotool binary path from the PlatformIO toolchain cc_path. - def __init__(self, path: Path, description: str) -> None: - self.path = path - self.description = description - - -def get_rp2040_mass_storage_volumes() -> list[MassStorageVolume]: - """Detect mounted RP2040 BOOTSEL mass storage volumes. - - When an RP2040 is in BOOTSEL mode, it appears as a USB mass storage - device named 'RPI-RP2'. This function finds those mount points. + The cc_path from IDEData points to the toolchain package, e.g.: + ~/.platformio/packages/toolchain-rp2040-earlephilhower/bin/arm-none-eabi-gcc + Picotool is in a sibling package: + ~/.platformio/packages/tool-picotool-rp2040-earlephilhower/picotool """ - result: list[MassStorageVolume] = [] + cc = Path(cc_path) + # Go from .../packages/toolchain-.../bin/gcc up to .../packages/ + packages_dir = cc.parent.parent.parent + binary_name = "picotool.exe" if sys.platform == "win32" else "picotool" + picotool = packages_dir / "tool-picotool-rp2040-earlephilhower" / binary_name + if picotool.is_file(): + return picotool + return None - if sys.platform == "darwin": - # macOS: /Volumes/RPI-RP2 - result.extend( - MassStorageVolume(path, "RP2040 BOOTSEL") - for path in Path("/Volumes").glob("RPI-RP2*") - if path.is_dir() + +def detect_rp2040_bootsel(picotool_path: str | Path) -> int: + """Detect RP2040/RP2350 devices in BOOTSEL mode using picotool. + + Returns the number of devices found (by counting 'type:' lines in output), + matching PlatformIO's detection approach. + """ + try: + result = subprocess.run( + [str(picotool_path), "info", "-d"], + capture_output=True, + timeout=10, + check=False, ) - - elif sys.platform.startswith("linux"): - # Linux: /media//RPI-RP2, /run/media//RPI-RP2, /mnt/RPI-RP2 - search_patterns = [ - Path("/media").glob("*/RPI-RP2*"), - Path("/run/media").glob("*/RPI-RP2*"), - Path("/mnt").glob("RPI-RP2*"), - ] - for pattern in search_patterns: - try: - result.extend( - MassStorageVolume(path, "RP2040 BOOTSEL") - for path in pattern - if path.is_dir() - ) - except OSError: - continue - - elif sys.platform == "win32": - # Windows: Check drive letters for RPI-RP2 volume label - import ctypes - - for letter in "DEFGHIJKLMNOPQRSTUVWXYZ": - drive = f"{letter}:\\" - if not Path(drive).exists(): - continue - try: - volume_name = ctypes.create_unicode_buffer(1024) - ctypes.windll.kernel32.GetVolumeInformationW( - drive, volume_name, 1024, None, None, None, None, 0 - ) - if volume_name.value.startswith("RPI-RP2"): - result.append(MassStorageVolume(Path(drive), "RP2040 BOOTSEL")) - except OSError: - continue - - return result + return result.stdout.count(b"type:") + except (OSError, subprocess.TimeoutExpired): + return 0 def get_esp32_arduino_flash_error_help() -> str | None: diff --git a/tests/unit_tests/test_main.py b/tests/unit_tests/test_main.py index 483dc4a23e..32ebeaacf2 100644 --- a/tests/unit_tests/test_main.py +++ b/tests/unit_tests/test_main.py @@ -40,8 +40,8 @@ from esphome.__main__ import ( show_logs, upload_program, upload_using_esptool, + upload_using_picotool, upload_using_platformio, - upload_using_uf2_copy, ) from esphome.components.esp32 import KEY_ESP32, KEY_VARIANT, VARIANT_ESP32 from esphome.const import ( @@ -177,9 +177,9 @@ def mock_upload_using_platformio() -> Generator[Mock]: @pytest.fixture -def mock_upload_using_uf2_copy() -> Generator[Mock]: - """Mock upload_using_uf2_copy for testing.""" - with patch("esphome.__main__.upload_using_uf2_copy") as mock: +def mock_upload_using_picotool() -> Generator[Mock]: + """Mock upload_using_picotool for testing.""" + with patch("esphome.__main__.upload_using_picotool") as mock: yield mock @@ -861,18 +861,17 @@ def test_choose_upload_log_host_no_address_with_ota_config() -> None: @pytest.mark.usefixtures("mock_no_serial_ports") -def test_choose_upload_log_host_no_defaults_with_rp2040_mass_storage( +def test_choose_upload_log_host_no_defaults_with_rp2040_bootsel( mock_choose_prompt: Mock, ) -> None: - """Test interactive mode shows RP2040 mass storage volumes.""" + """Test interactive mode shows RP2040 BOOTSEL option via picotool.""" setup_core(platform=PLATFORM_RP2040) - mock_volumes = [ - MagicMock(path=Path("/Volumes/RPI-RP2"), description="RP2040 BOOTSEL"), - ] - with patch( - "esphome.__main__.get_rp2040_mass_storage_volumes", - return_value=mock_volumes, + with ( + patch( + "esphome.__main__._find_picotool", return_value=Path("/usr/bin/picotool") + ), + patch("esphome.__main__.detect_rp2040_bootsel", return_value=1), ): result = choose_upload_log_host( default=None, @@ -880,9 +879,8 @@ def test_choose_upload_log_host_no_defaults_with_rp2040_mass_storage( purpose=Purpose.UPLOADING, ) assert result == ["/dev/ttyUSB0"] # mock_choose_prompt default - vol_path = str(Path("/Volumes/RPI-RP2")) mock_choose_prompt.assert_called_once_with( - [(f"{vol_path} (RP2040 BOOTSEL)", f"MS:{vol_path}")], + [("RP2040 BOOTSEL (via picotool)", "BOOTSEL")], purpose=Purpose.UPLOADING, ) @@ -894,9 +892,9 @@ def test_choose_upload_log_host_rp2040_no_device_shows_bootsel_help() -> None: with ( patch( - "esphome.__main__.get_rp2040_mass_storage_volumes", - return_value=[], + "esphome.__main__._find_picotool", return_value=Path("/usr/bin/picotool") ), + patch("esphome.__main__.detect_rp2040_bootsel", return_value=0), pytest.raises(EsphomeError, match="BOOTSEL"), ): choose_upload_log_host( @@ -919,9 +917,9 @@ def test_choose_upload_log_host_rp2040_bootsel_tip_with_ota( with ( patch( - "esphome.__main__.get_rp2040_mass_storage_volumes", - return_value=[], + "esphome.__main__._find_picotool", return_value=Path("/usr/bin/picotool") ), + patch("esphome.__main__.detect_rp2040_bootsel", return_value=0), patch( "esphome.__main__.choose_prompt", return_value="192.168.1.100", @@ -936,10 +934,10 @@ def test_choose_upload_log_host_rp2040_bootsel_tip_with_ota( assert "BOOTSEL" in caplog.text -def test_choose_upload_log_host_no_mass_storage_for_non_rp2040( +def test_choose_upload_log_host_no_bootsel_for_non_rp2040( mock_no_serial_ports: Mock, ) -> None: - """Test that mass storage detection is not run for non-RP2040 platforms.""" + """Test that BOOTSEL detection is not run for non-RP2040 platforms.""" setup_core( platform=PLATFORM_ESP32, config={CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME}]}, @@ -947,9 +945,7 @@ def test_choose_upload_log_host_no_mass_storage_for_non_rp2040( ) with ( - patch( - "esphome.__main__.get_rp2040_mass_storage_volumes", - ) as mock_get_volumes, + patch("esphome.__main__._find_picotool") as mock_find_picotool, patch( "esphome.__main__.choose_prompt", return_value="192.168.1.100", @@ -960,36 +956,32 @@ def test_choose_upload_log_host_no_mass_storage_for_non_rp2040( check_default=None, purpose=Purpose.UPLOADING, ) - mock_get_volumes.assert_not_called() + mock_find_picotool.assert_not_called() -def test_choose_upload_log_host_rp2040_serial_and_mass_storage( +def test_choose_upload_log_host_rp2040_serial_and_bootsel( mock_choose_prompt: Mock, ) -> None: - """Test both serial ports and mass storage volumes shown for RP2040.""" + """Test both serial ports and BOOTSEL option shown for RP2040.""" setup_core(platform=PLATFORM_RP2040) mock_ports = [MockSerialPort("/dev/ttyACM0", "RP2040 Serial")] - mock_volumes = [ - MagicMock(path=Path("/Volumes/RPI-RP2"), description="RP2040 BOOTSEL"), - ] with ( patch("esphome.__main__.get_serial_ports", return_value=mock_ports), patch( - "esphome.__main__.get_rp2040_mass_storage_volumes", - return_value=mock_volumes, + "esphome.__main__._find_picotool", return_value=Path("/usr/bin/picotool") ), + patch("esphome.__main__.detect_rp2040_bootsel", return_value=1), ): choose_upload_log_host( default=None, check_default=None, purpose=Purpose.UPLOADING, ) - vol_path = str(Path("/Volumes/RPI-RP2")) mock_choose_prompt.assert_called_once_with( [ ("/dev/ttyACM0 (RP2040 Serial)", "/dev/ttyACM0"), - (f"{vol_path} (RP2040 BOOTSEL)", f"MS:{vol_path}"), + ("RP2040 BOOTSEL (via picotool)", "BOOTSEL"), ], purpose=Purpose.UPLOADING, ) @@ -1266,108 +1258,152 @@ def test_upload_program_serial_upload_failed( mock_upload_using_esptool.assert_called_once() -def test_upload_program_mass_storage( - mock_upload_using_uf2_copy: Mock, +def test_upload_program_bootsel( + mock_upload_using_picotool: Mock, mock_get_port_type: Mock, ) -> None: - """Test upload_program with mass storage for RP2040.""" + """Test upload_program with BOOTSEL for RP2040.""" setup_core(platform=PLATFORM_RP2040) - mock_get_port_type.return_value = "MASS_STORAGE" - mock_upload_using_uf2_copy.return_value = 0 + mock_get_port_type.return_value = "BOOTSEL" + mock_upload_using_picotool.return_value = 0 config = {} args = MockArgs() - devices = ["MS:/Volumes/RPI-RP2"] + devices = ["BOOTSEL"] exit_code, host = upload_program(config, args, devices) assert exit_code == 0 - # Mass storage device can't be used for logging, so host should be None + # BOOTSEL device can't be used for logging, so host should be None assert host is None - mock_upload_using_uf2_copy.assert_called_once_with(config, "/Volumes/RPI-RP2") + mock_upload_using_picotool.assert_called_once_with(config) -def test_upload_program_mass_storage_failed( - mock_upload_using_uf2_copy: Mock, +def test_upload_program_bootsel_failed( + mock_upload_using_picotool: Mock, mock_get_port_type: Mock, ) -> None: - """Test upload_program when mass storage upload fails.""" + """Test upload_program when BOOTSEL upload fails.""" setup_core(platform=PLATFORM_RP2040) - mock_get_port_type.return_value = "MASS_STORAGE" - mock_upload_using_uf2_copy.return_value = 1 + mock_get_port_type.return_value = "BOOTSEL" + mock_upload_using_picotool.return_value = 1 config = {} args = MockArgs() - devices = ["MS:/Volumes/RPI-RP2"] + devices = ["BOOTSEL"] exit_code, host = upload_program(config, args, devices) assert exit_code == 1 assert host is None - mock_upload_using_uf2_copy.assert_called_once_with(config, "/Volumes/RPI-RP2") + mock_upload_using_picotool.assert_called_once_with(config) -def test_upload_using_uf2_copy_success(tmp_path: Path) -> None: - """Test upload_using_uf2_copy copies UF2 file with progress.""" +def test_upload_using_picotool_success(tmp_path: Path) -> None: + """Test upload_using_picotool succeeds.""" setup_core(platform=PLATFORM_RP2040, tmp_path=tmp_path) - # Create a mock UF2 file build_dir = tmp_path / "build" build_dir.mkdir() - uf2_file = build_dir / "firmware.uf2" - uf2_file.write_bytes(b"\x00" * 1024) + firmware_elf = build_dir / "firmware.elf" + firmware_elf.write_bytes(b"\x00" * 1024) - # Create a mock mount point - mount_dir = tmp_path / "RPI-RP2" - mount_dir.mkdir() + # Create picotool binary + packages_dir = tmp_path / "packages" + toolchain_bin = packages_dir / "toolchain-rp2040-earlephilhower" / "bin" + toolchain_bin.mkdir(parents=True) + picotool_dir = packages_dir / "tool-picotool-rp2040-earlephilhower" + picotool_dir.mkdir(parents=True) + picotool = picotool_dir / "picotool" + picotool.touch() mock_idedata = MagicMock() - mock_idedata.firmware_elf_path = str(build_dir / "firmware.elf") + mock_idedata.firmware_elf_path = str(firmware_elf) + mock_idedata.cc_path = str(toolchain_bin / "arm-none-eabi-gcc") + + mock_result = MagicMock() + mock_result.returncode = 0 + mock_result.stderr = b"" config = {} - with patch("esphome.platformio_api.get_idedata", return_value=mock_idedata): - exit_code = upload_using_uf2_copy(config, str(mount_dir)) + with ( + patch("esphome.platformio_api.get_idedata", return_value=mock_idedata), + patch("subprocess.run", return_value=mock_result), + ): + exit_code = upload_using_picotool(config) assert exit_code == 0 - assert (mount_dir / "firmware.uf2").exists() - assert (mount_dir / "firmware.uf2").read_bytes() == b"\x00" * 1024 -def test_upload_using_uf2_copy_no_uf2_file(tmp_path: Path) -> None: - """Test upload_using_uf2_copy when UF2 file is missing.""" +def test_upload_using_picotool_no_elf(tmp_path: Path) -> None: + """Test upload_using_picotool when ELF file is missing.""" setup_core(platform=PLATFORM_RP2040, tmp_path=tmp_path) build_dir = tmp_path / "build" build_dir.mkdir() - mount_dir = tmp_path / "RPI-RP2" - mount_dir.mkdir() - mock_idedata = MagicMock() mock_idedata.firmware_elf_path = str(build_dir / "firmware.elf") + mock_idedata.cc_path = "/fake/path/gcc" config = {} with patch("esphome.platformio_api.get_idedata", return_value=mock_idedata): - exit_code = upload_using_uf2_copy(config, str(mount_dir)) + exit_code = upload_using_picotool(config) assert exit_code == 1 -def test_upload_using_uf2_copy_mount_gone(tmp_path: Path) -> None: - """Test upload_using_uf2_copy when mount point disappeared.""" +def test_upload_using_picotool_not_found(tmp_path: Path) -> None: + """Test upload_using_picotool when picotool binary not found.""" setup_core(platform=PLATFORM_RP2040, tmp_path=tmp_path) build_dir = tmp_path / "build" build_dir.mkdir() - uf2_file = build_dir / "firmware.uf2" - uf2_file.write_bytes(b"\x00" * 512) + firmware_elf = build_dir / "firmware.elf" + firmware_elf.write_bytes(b"\x00" * 512) mock_idedata = MagicMock() - mock_idedata.firmware_elf_path = str(build_dir / "firmware.elf") + mock_idedata.firmware_elf_path = str(firmware_elf) + mock_idedata.cc_path = "/fake/path/gcc" config = {} with patch("esphome.platformio_api.get_idedata", return_value=mock_idedata): - exit_code = upload_using_uf2_copy(config, str(tmp_path / "nonexistent")) + exit_code = upload_using_picotool(config) + + assert exit_code == 1 + + +def test_upload_using_picotool_permission_error(tmp_path: Path) -> None: + """Test upload_using_picotool shows helpful message on permission error.""" + setup_core(platform=PLATFORM_RP2040, tmp_path=tmp_path) + + build_dir = tmp_path / "build" + build_dir.mkdir() + firmware_elf = build_dir / "firmware.elf" + firmware_elf.write_bytes(b"\x00" * 512) + + packages_dir = tmp_path / "packages" + toolchain_bin = packages_dir / "toolchain-rp2040-earlephilhower" / "bin" + toolchain_bin.mkdir(parents=True) + picotool_dir = packages_dir / "tool-picotool-rp2040-earlephilhower" + picotool_dir.mkdir(parents=True) + picotool = picotool_dir / "picotool" + picotool.touch() + + mock_idedata = MagicMock() + mock_idedata.firmware_elf_path = str(firmware_elf) + mock_idedata.cc_path = str(toolchain_bin / "arm-none-eabi-gcc") + + mock_result = MagicMock() + mock_result.returncode = 1 + mock_result.stderr = b"LIBUSB_ERROR_ACCESS" + + config = {} + with ( + patch("esphome.platformio_api.get_idedata", return_value=mock_idedata), + patch("subprocess.run", return_value=mock_result), + ): + exit_code = upload_using_picotool(config) assert exit_code == 1 @@ -1896,9 +1932,7 @@ def test_get_port_type() -> None: assert get_port_type("esphome-device.local") == "NETWORK" assert get_port_type("10.0.0.1") == "NETWORK" - assert get_port_type("MS:/Volumes/RPI-RP2") == "MASS_STORAGE" - assert get_port_type("MS:/media/user/RPI-RP2") == "MASS_STORAGE" - assert get_port_type("MS:D:\\") == "MASS_STORAGE" + assert get_port_type("BOOTSEL") == "BOOTSEL" def test_has_mqtt_ip_lookup() -> None: diff --git a/tests/unit_tests/test_util.py b/tests/unit_tests/test_util.py index 7fd3d4b851..8a2529a6f6 100644 --- a/tests/unit_tests/test_util.py +++ b/tests/unit_tests/test_util.py @@ -3,6 +3,7 @@ from __future__ import annotations from pathlib import Path +import subprocess from unittest.mock import MagicMock, patch import pytest @@ -405,137 +406,95 @@ def test_shlex_quote_edge_cases() -> None: assert util.shlex_quote(" ") == "' '" -def test_get_rp2040_mass_storage_volumes_macos(tmp_path: Path) -> None: - """Test RP2040 mass storage detection on macOS.""" - volumes_dir = tmp_path / "Volumes" - volumes_dir.mkdir() - rpi_vol = volumes_dir / "RPI-RP2" - rpi_vol.mkdir() +def test_get_picotool_path_found(tmp_path: Path) -> None: + """Test picotool path derivation from cc_path.""" + # Create the expected directory structure + packages_dir = tmp_path / "packages" + toolchain_dir = packages_dir / "toolchain-rp2040-earlephilhower" / "bin" + toolchain_dir.mkdir(parents=True) + gcc = toolchain_dir / "arm-none-eabi-gcc" + gcc.touch() - with ( - patch("esphome.util.sys") as mock_sys, - patch("esphome.util.Path") as mock_path_cls, + picotool_dir = packages_dir / "tool-picotool-rp2040-earlephilhower" + picotool_dir.mkdir(parents=True) + picotool = picotool_dir / "picotool" + picotool.touch() + + result = util.get_picotool_path(str(gcc)) + assert result == picotool + + +def test_get_picotool_path_not_found(tmp_path: Path) -> None: + """Test picotool path returns None when not installed.""" + packages_dir = tmp_path / "packages" + toolchain_dir = packages_dir / "toolchain-rp2040-earlephilhower" / "bin" + toolchain_dir.mkdir(parents=True) + gcc = toolchain_dir / "arm-none-eabi-gcc" + gcc.touch() + + result = util.get_picotool_path(str(gcc)) + assert result is None + + +def test_get_picotool_path_windows(tmp_path: Path) -> None: + """Test picotool path uses .exe on Windows.""" + packages_dir = tmp_path / "packages" + toolchain_dir = packages_dir / "toolchain-rp2040-earlephilhower" / "bin" + toolchain_dir.mkdir(parents=True) + gcc = toolchain_dir / "arm-none-eabi-gcc.exe" + gcc.touch() + + picotool_dir = packages_dir / "tool-picotool-rp2040-earlephilhower" + picotool_dir.mkdir(parents=True) + picotool = picotool_dir / "picotool.exe" + picotool.touch() + + with patch("esphome.util.sys.platform", "win32"): + result = util.get_picotool_path(str(gcc)) + assert result == picotool + + +def test_detect_rp2040_bootsel_found() -> None: + """Test BOOTSEL device detection when device is present.""" + mock_result = MagicMock() + mock_result.stdout = b"Device Information\n type: RP2040\n" + with patch("esphome.util.subprocess.run", return_value=mock_result): + count = util.detect_rp2040_bootsel("/usr/bin/picotool") + assert count == 1 + + +def test_detect_rp2040_bootsel_multiple() -> None: + """Test BOOTSEL detection with multiple devices.""" + mock_result = MagicMock() + mock_result.stdout = b"type: RP2040\ntype: RP2350\n" + with patch("esphome.util.subprocess.run", return_value=mock_result): + count = util.detect_rp2040_bootsel("/usr/bin/picotool") + assert count == 2 + + +def test_detect_rp2040_bootsel_none() -> None: + """Test BOOTSEL detection when no device found.""" + mock_result = MagicMock() + mock_result.stdout = ( + b"No accessible RP2040/RP2350 devices in BOOTSEL mode were found.\n" + ) + with patch("esphome.util.subprocess.run", return_value=mock_result): + count = util.detect_rp2040_bootsel("/usr/bin/picotool") + assert count == 0 + + +def test_detect_rp2040_bootsel_oserror() -> None: + """Test BOOTSEL detection handles OSError.""" + with patch("esphome.util.subprocess.run", side_effect=OSError("not found")): + count = util.detect_rp2040_bootsel("/usr/bin/picotool") + assert count == 0 + + +def test_detect_rp2040_bootsel_timeout() -> None: + """Test BOOTSEL detection handles timeout.""" + with patch( + "esphome.util.subprocess.run", + side_effect=subprocess.TimeoutExpired("picotool", 10), ): - mock_sys.platform = "darwin" - # Make Path("/Volumes") return our tmp_path version - mock_path_cls.side_effect = lambda p: ( - volumes_dir if p == "/Volumes" else Path(p) - ) - - result = util.get_rp2040_mass_storage_volumes() - - assert len(result) == 1 - assert result[0].description == "RP2040 BOOTSEL" - - -def test_get_rp2040_mass_storage_volumes_none_found(tmp_path: Path) -> None: - """Test RP2040 mass storage detection when no volumes found.""" - # Point at an empty directory so no RPI-RP2* matches - empty_dir = tmp_path / "Volumes" - empty_dir.mkdir() - - with ( - patch("esphome.util.sys.platform", "darwin"), - patch( - "esphome.util.Path", - side_effect=lambda p: empty_dir if p == "/Volumes" else Path(p), - ), - ): - result = util.get_rp2040_mass_storage_volumes() - - assert result == [] - - -def test_get_rp2040_mass_storage_volumes_linux(tmp_path: Path) -> None: - """Test RP2040 mass storage detection on Linux.""" - # Create /media//RPI-RP2 structure - media_dir = tmp_path / "media" - media_dir.mkdir() - user_dir = media_dir / "testuser" - user_dir.mkdir() - rp2_dir = user_dir / "RPI-RP2" - rp2_dir.mkdir() - - # Create /run/media and /mnt as empty dirs - run_media_dir = tmp_path / "run_media" - run_media_dir.mkdir() - mnt_dir = tmp_path / "mnt" - mnt_dir.mkdir() - - def mock_path_side_effect(p: str) -> Path: - if p == "/media": - return media_dir - if p == "/run/media": - return run_media_dir - if p == "/mnt": - return mnt_dir - return Path(p) - - with ( - patch("esphome.util.sys.platform", "linux"), - patch("esphome.util.Path", side_effect=mock_path_side_effect), - ): - result = util.get_rp2040_mass_storage_volumes() - - assert len(result) == 1 - assert result[0].description == "RP2040 BOOTSEL" - - -def test_get_rp2040_mass_storage_volumes_linux_oserror(tmp_path: Path) -> None: - """Test RP2040 mass storage detection on Linux handles OSError.""" - media_dir = tmp_path / "media" - media_dir.mkdir() - - def mock_path_side_effect(p: str) -> Path: - if p == "/media": - return media_dir - if p in ("/run/media", "/mnt"): - # Return a path that will raise OSError when globbed - return tmp_path / "nonexistent" - return Path(p) - - with ( - patch("esphome.util.sys.platform", "linux"), - patch("esphome.util.Path", side_effect=mock_path_side_effect), - ): - result = util.get_rp2040_mass_storage_volumes() - - assert result == [] - - -def test_get_rp2040_mass_storage_volumes_windows() -> None: - """Test RP2040 mass storage detection on Windows.""" - mock_ctypes = MagicMock() - mock_volume_name = MagicMock() - mock_volume_name.value = "RPI-RP2" - mock_ctypes.create_unicode_buffer.return_value = mock_volume_name - - def path_side_effect(p: str) -> MagicMock: - inst = MagicMock() - inst.exists.return_value = p == "D:\\" - return inst - - with ( - patch("esphome.util.sys.platform", "win32"), - patch.dict("sys.modules", {"ctypes": mock_ctypes}), - patch("esphome.util.Path", side_effect=path_side_effect), - ): - result = util.get_rp2040_mass_storage_volumes() - - assert len(result) >= 1 - assert result[0].description == "RP2040 BOOTSEL" - - -def test_get_rp2040_mass_storage_volumes_unsupported_platform() -> None: - """Test RP2040 mass storage detection on unsupported platform returns empty.""" - with patch("esphome.util.sys.platform", "freebsd"): - result = util.get_rp2040_mass_storage_volumes() - - assert result == [] - - -def test_mass_storage_volume_attributes() -> None: - """Test MassStorageVolume class attributes.""" - vol = util.MassStorageVolume(Path("/Volumes/RPI-RP2"), "RP2040 BOOTSEL") - assert vol.path == Path("/Volumes/RPI-RP2") - assert vol.description == "RP2040 BOOTSEL" + count = util.detect_rp2040_bootsel("/usr/bin/picotool") + assert count == 0