mirror of
https://github.com/esphome/esphome.git
synced 2026-09-01 10:36:01 +00:00
[rp2040] Use picotool for BOOTSEL upload instead of mass storage copy
Replace UF2 file copy to mass storage volume with direct picotool upload. This avoids macOS "disk not ejected properly" warnings caused by the RP2040 resetting immediately after receiving the firmware. - Use picotool (already installed by PlatformIO) for BOOTSEL detection and firmware upload via USB - Upload ELF directly with `picotool load -v -x` for real-time progress - Remove platform-specific mass storage volume detection (macOS/Linux/Windows) - Show helpful udev rules message on Linux permission errors
This commit is contained in:
+71
-59
@@ -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
|
||||
|
||||
+30
-56
@@ -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/<user>/RPI-RP2, /run/media/<user>/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:
|
||||
|
||||
+110
-76
@@ -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:
|
||||
|
||||
+91
-132
@@ -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/<user>/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
|
||||
|
||||
Reference in New Issue
Block a user