github/esphome
2
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-07-11 09:25:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

[rp2040] Improve upload experience with mass storage and BOOTSEL support

Browse Source 
Add auto-detection of RP2040 BOOTSEL mass storage volumes (RPI-RP2) on
macOS, Linux, and Windows. Show detected volumes as upload targets with
a progress bar for UF2 file copy. Display helpful BOOTSEL instructions
when no RP2040 device is found.

- Add get_rp2040_mass_storage_volumes() to detect mounted RPI-RP2 volumes
- Add PortType.MASS_STORAGE and upload_using_uf2_copy() with progress bar
- Move ProgressBar to helpers.py for shared use
- Wait for USB-CDC serial port after upload for log output
- Auto-select single serial port for logs after mass storage upload
- Create firmware.bin.signed in post_build to fix nobuild upload target
- Show BOOTSEL tip when only OTA options are available
This commit is contained in:
J. Nick Koston
2026-03-04 13:50:52 -10:00
parent 61ea6c3b2f
commit 74dd61442a
7 changed files with 558 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
esphome/__main__.py
+153 -2
View File
@@ -44,7 +44,9 @@ from esphome.const import (
CONF_SUBSTITUTIONS,
CONF_TOPIC,
ENV_NOGITIGNORE,
KEY_CORE,
KEY_NATIVE_IDF,
KEY_TARGET_PLATFORM,
PLATFORM_ESP32,
PLATFORM_ESP8266,
PLATFORM_RP2040,
@@ -56,6 +58,7 @@ 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,
get_serial_ports,
list_yaml_files,
run_external_command,
@@ -68,6 +71,15 @@ _LOGGER = logging.getLogger(__name__)
# Maximum buffer size for serial log reading to prevent unbounded memory growth
SERIAL_BUFFER_MAX_SIZE = 65536
_RP2040_BOOTSEL_INSTRUCTIONS = (
"To enter BOOTSEL mode:\n"
" 1. Unplug the device\n"
" 2. Hold the BOOT/BOOTSEL button\n"
" 3. Plug in the USB cable while holding the button\n"
" 4. Release the button - the device should appear as a USB drive (RPI-RP2)\n"
"Then run the upload command again."
)
# Special non-component keys that appear in configs
_NON_COMPONENT_KEYS = frozenset(
{
@@ -163,6 +175,7 @@ class PortType(StrEnum):
NETWORK = "NETWORK"
MQTT = "MQTT"
MQTTIP = "MQTTIP"
MASS_STORAGE = "MASS_STORAGE"
# Magic MQTT port types that require special handling
@@ -241,6 +254,15 @@ 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
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}"))
if purpose == Purpose.LOGGING:
if has_mqtt_logging():
mqtt_config = CORE.config[CONF_MQTT]
@@ -258,6 +280,21 @@ def choose_upload_log_host(
if has_mqtt_ip_lookup():
options.append(("Over The Air (MQTT IP lookup)", "MQTTIP"))
# Show helpful BOOTSEL instructions for RP2040 when no USB device is found
if (
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)
for opt in options
)
):
if not options:
raise EsphomeError(
f"No RP2040 device found. {_RP2040_BOOTSEL_INSTRUCTIONS}"
)
_LOGGER.info("Tip: %s", _RP2040_BOOTSEL_INSTRUCTIONS)
if check_default is not None and check_default in [opt[1] for opt in options]:
return [check_default]
return [choose_prompt(options, purpose=purpose)]
@@ -404,10 +441,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.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.startswith("/") or port.startswith("COM"):
return PortType.SERIAL
if port == "MQTT":
@@ -695,7 +735,7 @@ def upload_using_esptool(
return run_esptool(115200)
def upload_using_platformio(config: ConfigType, port: str):
def upload_using_platformio(config: ConfigType, port: str) -> int:
from esphome import platformio_api
upload_args = ["-t", "upload", "-t", "nobuild"]
@@ -704,6 +744,94 @@ def upload_using_platformio(config: ConfigType, port: str):
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.
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.
"""
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"
if not uf2_file.exists():
_LOGGER.error(
"UF2 firmware file not found at %s. Make sure the project has been compiled first.",
uf2_file,
)
return 1
dest_dir = Path(mount_path)
if not dest_dir.is_dir():
_LOGGER.error(
"Mass storage volume %s is no longer available. "
"Is the RP2040 still in BOOTSEL mode?",
mount_path,
)
return 1
dest_file = dest_dir / uf2_file.name
file_size = uf2_file.stat().st_size
_LOGGER.info("Uploading UF2 firmware to %s (%s bytes)", mount_path, file_size)
progress = ProgressBar()
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()
except OSError as err:
progress.done()
_LOGGER.error("Failed to copy UF2 file to %s: %s", mount_path, err)
return 1
_LOGGER.info(
"Successfully copied firmware to %s. "
"The device will automatically reset and run the new firmware.",
mount_path,
)
return 0
def _wait_for_serial_port(port: str | None = None, timeout: float = 30.0) -> None:
"""Wait for a serial port to appear, e.g. after a device reboot.
USB-CDC devices disappear briefly after flashing while the device
reboots and re-enumerates on the USB bus.
If port is given, wait for that specific path. Otherwise wait for
any serial port to appear.
"""
if port is not None and os.access(port, os.F_OK):
return
if port is not None:
_LOGGER.info("Waiting for %s to come online...", port)
else:
_LOGGER.info("Waiting for device to reboot...")
start = time.monotonic()
while time.monotonic() - start < timeout:
time.sleep(0.05)
if port is not None:
if os.access(port, os.F_OK):
time.sleep(0.05)
return
elif get_serial_ports():
time.sleep(0.05)
return
def check_permissions(port: str):
if os.name == "posix" and get_port_type(port) == PortType.SERIAL:
# Check if we can open selected serial port
@@ -733,7 +861,17 @@ def upload_program(
except AttributeError:
pass
if get_port_type(host) == PortType.SERIAL:
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,
# so command_run will show the interactive chooser for log source
return exit_code, None
if port_type == PortType.SERIAL:
check_permissions(host)
exit_code = 1
@@ -787,6 +925,7 @@ def show_logs(config: ConfigType, args: ArgsProtocol, devices: list[str]) -> int
port_type = get_port_type(port)
if port_type == PortType.SERIAL:
_wait_for_serial_port(port)
check_permissions(port)
return run_miniterm(config, port, args)
@@ -935,6 +1074,18 @@ def command_run(args: ArgsProtocol, config: ConfigType) -> int | None:
if args.no_logs:
return 0
# After mass storage upload, wait for the serial port to reappear
# so it shows up in the log chooser
if (
successful_device is None
and CORE.data.get(KEY_CORE, {}).get(KEY_TARGET_PLATFORM) == PLATFORM_RP2040
):
_wait_for_serial_port()
# If exactly one serial port appeared, use it directly
serial_ports = get_serial_ports()
if len(serial_ports) == 1:
successful_device = serial_ports[0].path
# For logs, prefer the device we successfully uploaded to
devices = choose_upload_log_host(
default=successful_device,
esphome/components/rp2040/post_build.py.script
+15
View File
@@ -18,6 +18,21 @@ def rp2040_copy_ota_bin(source, target, env):
shutil.copyfile(firmware_name, new_file_name)
def rp2040_copy_signed_bin(source, target, env):
"""Create firmware.bin.signed so that 'nobuild' upload target can find it.
The platform-raspberrypi build recipe creates firmware.bin.signed as a build
target, but the 'nobuild' upload flag skips the build phase. Without this
file, the upload fails with 'firmware.bin.signed not found'.
ESPHome does not use signing for RP2040, so this is just a copy.
"""
firmware_name = env.subst("$BUILD_DIR/${PROGNAME}.bin")
signed_name = env.subst("$BUILD_DIR/${PROGNAME}.bin.signed")
shutil.copyfile(firmware_name, signed_name)
# pylint: disable=E0602
env.AddPostAction("$BUILD_DIR/${PROGNAME}.bin", rp2040_copy_factory_uf2) # noqa
env.AddPostAction("$BUILD_DIR/${PROGNAME}.bin", rp2040_copy_ota_bin) # noqa
env.AddPostAction("$BUILD_DIR/${PROGNAME}.bin", rp2040_copy_signed_bin) # noqa
esphome/espota2.py
+1 -25
View File
@@ -13,7 +13,7 @@ import time
from typing import Any
from esphome.core import EsphomeError
from esphome.helpers import resolve_ip_address
from esphome.helpers import ProgressBar, resolve_ip_address
RESPONSE_OK = 0x00
RESPONSE_REQUEST_AUTH = 0x01
@@ -63,30 +63,6 @@ _AUTH_METHODS: dict[int, tuple[Callable[..., Any], int, str]] = {
}
class ProgressBar:
def __init__(self):
self.last_progress = None
def update(self, progress):
bar_length = 60
status = ""
if progress >= 1:
progress = 1
status = "Done...\r\n"
new_progress = int(progress * 100)
if new_progress == self.last_progress:
return
self.last_progress = new_progress
block = int(round(bar_length * progress))
text = f"\rUploading: [{'=' * block + ' ' * (bar_length - block)}] {new_progress}% {status}"
sys.stderr.write(text)
sys.stderr.flush()
def done(self):
sys.stderr.write("\n")
sys.stderr.flush()
class OTAError(EsphomeError):
pass
esphome/helpers.py
+27
View File
@@ -9,6 +9,7 @@ import platform
import re
import shutil
import stat
import sys
import tempfile
from typing import TYPE_CHECKING
from urllib.parse import urlparse
@@ -585,6 +586,32 @@ def sanitize(value):
return _DISALLOWED_CHARS.sub("_", value)
class ProgressBar:
"""A simple terminal progress bar for upload operations."""
def __init__(self) -> None:
self.last_progress: int | None = None
def update(self, progress: float) -> None:
bar_length = 60
status = ""
if progress >= 1:
progress = 1
status = "Done...\r\n"
new_progress = int(progress * 100)
if new_progress == self.last_progress:
return
self.last_progress = new_progress
block = int(round(bar_length * progress))
text = f"\rUploading: [{'=' * block + ' ' * (bar_length - block)}] {new_progress}% {status}"
sys.stderr.write(text)
sys.stderr.flush()
def done(self) -> None:
sys.stderr.write("\n")
sys.stderr.flush()
def docs_url(path: str) -> str:
"""Return the URL to the documentation for a given path."""
# Local import to avoid circular import
esphome/util.py
+62
View File
@@ -355,6 +355,68 @@ def get_serial_ports() -> list[SerialPort]:
return result
class MassStorageVolume:
"""Represents a mass storage volume for RP2040 BOOTSEL upload."""
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.
"""
result: list[MassStorageVolume] = []
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()
)
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
def get_esp32_arduino_flash_error_help() -> str | None:
"""Returns helpful message when ESP32 with Arduino runs out of flash space."""
from esphome.core import CORE
tests/unit_tests/test_main.py
+251
View File
@@ -40,6 +40,7 @@ from esphome.__main__ import (
show_logs,
upload_program,
upload_using_esptool,
upload_using_uf2_copy,
)
from esphome.components.esp32 import KEY_ESP32, KEY_VARIANT, VARIANT_ESP32
from esphome.const import (
@@ -174,6 +175,13 @@ def mock_upload_using_platformio() -> Generator[Mock]:
yield 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:
yield mock
@pytest.fixture
def mock_run_ota() -> Generator[Mock]:
"""Mock espota2.run_ota for testing."""
@@ -851,6 +859,139 @@ 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(
mock_choose_prompt: Mock,
) -> None:
"""Test interactive mode shows RP2040 mass storage volumes."""
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,
):
result = choose_upload_log_host(
default=None,
check_default=None,
purpose=Purpose.UPLOADING,
)
assert result == ["/dev/ttyUSB0"] # mock_choose_prompt default
mock_choose_prompt.assert_called_once_with(
[("/Volumes/RPI-RP2 (RP2040 BOOTSEL)", "MS:/Volumes/RPI-RP2")],
purpose=Purpose.UPLOADING,
)
@pytest.mark.usefixtures("mock_no_serial_ports")
def test_choose_upload_log_host_rp2040_no_device_shows_bootsel_help() -> None:
"""Test BOOTSEL instructions shown when no RP2040 device found."""
setup_core(platform=PLATFORM_RP2040)
with (
patch(
"esphome.__main__.get_rp2040_mass_storage_volumes",
return_value=[],
),
pytest.raises(EsphomeError, match="BOOTSEL"),
):
choose_upload_log_host(
default=None,
check_default=None,
purpose=Purpose.UPLOADING,
)
@pytest.mark.usefixtures("mock_no_serial_ports")
def test_choose_upload_log_host_rp2040_bootsel_tip_with_ota(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test BOOTSEL tip shown when only OTA options exist for RP2040."""
setup_core(
platform=PLATFORM_RP2040,
config={CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME}]},
address="192.168.1.100",
)
with (
patch(
"esphome.__main__.get_rp2040_mass_storage_volumes",
return_value=[],
),
patch(
"esphome.__main__.choose_prompt",
return_value="192.168.1.100",
),
caplog.at_level(logging.INFO, logger="esphome.__main__"),
):
choose_upload_log_host(
default=None,
check_default=None,
purpose=Purpose.UPLOADING,
)
assert "BOOTSEL" in caplog.text
def test_choose_upload_log_host_no_mass_storage_for_non_rp2040(
mock_no_serial_ports: Mock,
) -> None:
"""Test that mass storage detection is not run for non-RP2040 platforms."""
setup_core(
platform=PLATFORM_ESP32,
config={CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME}]},
address="192.168.1.100",
)
with (
patch(
"esphome.__main__.get_rp2040_mass_storage_volumes",
) as mock_get_volumes,
patch(
"esphome.__main__.choose_prompt",
return_value="192.168.1.100",
),
):
choose_upload_log_host(
default=None,
check_default=None,
purpose=Purpose.UPLOADING,
)
mock_get_volumes.assert_not_called()
def test_choose_upload_log_host_rp2040_serial_and_mass_storage(
mock_choose_prompt: Mock,
) -> None:
"""Test both serial ports and mass storage volumes 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,
),
):
choose_upload_log_host(
default=None,
check_default=None,
purpose=Purpose.UPLOADING,
)
mock_choose_prompt.assert_called_once_with(
[
("/dev/ttyACM0 (RP2040 Serial)", "/dev/ttyACM0"),
("/Volumes/RPI-RP2 (RP2040 BOOTSEL)", "MS:/Volumes/RPI-RP2"),
],
purpose=Purpose.UPLOADING,
)
@dataclass
class MockArgs:
"""Mock args for testing."""
@@ -1082,6 +1223,112 @@ 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,
mock_get_port_type: Mock,
) -> None:
"""Test upload_program with mass storage for RP2040."""
setup_core(platform=PLATFORM_RP2040)
mock_get_port_type.return_value = "MASS_STORAGE"
mock_upload_using_uf2_copy.return_value = 0
config = {}
args = MockArgs()
devices = ["MS:/Volumes/RPI-RP2"]
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
assert host is None
mock_upload_using_uf2_copy.assert_called_once_with(config, "/Volumes/RPI-RP2")
def test_upload_program_mass_storage_failed(
mock_upload_using_uf2_copy: Mock,
mock_get_port_type: Mock,
) -> None:
"""Test upload_program when mass storage upload fails."""
setup_core(platform=PLATFORM_RP2040)
mock_get_port_type.return_value = "MASS_STORAGE"
mock_upload_using_uf2_copy.return_value = 1
config = {}
args = MockArgs()
devices = ["MS:/Volumes/RPI-RP2"]
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")
def test_upload_using_uf2_copy_success(tmp_path: Path) -> None:
"""Test upload_using_uf2_copy copies UF2 file with progress."""
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)
# Create a mock mount point
mount_dir = tmp_path / "RPI-RP2"
mount_dir.mkdir()
mock_idedata = MagicMock()
mock_idedata.firmware_elf_path = str(build_dir / "firmware.elf")
config = {}
with patch("esphome.platformio_api.get_idedata", return_value=mock_idedata):
exit_code = upload_using_uf2_copy(config, str(mount_dir))
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."""
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")
config = {}
with patch("esphome.platformio_api.get_idedata", return_value=mock_idedata):
exit_code = upload_using_uf2_copy(config, str(mount_dir))
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."""
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)
mock_idedata = MagicMock()
mock_idedata.firmware_elf_path = str(build_dir / "firmware.elf")
config = {}
with patch("esphome.platformio_api.get_idedata", return_value=mock_idedata):
exit_code = upload_using_uf2_copy(config, str(tmp_path / "nonexistent"))
assert exit_code == 1
def test_upload_program_ota_success(
mock_run_ota: Mock,
mock_get_port_type: Mock,
@@ -1606,6 +1853,10 @@ 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"
def test_has_mqtt_ip_lookup() -> None:
"""Test has_mqtt_ip_lookup function."""
tests/unit_tests/test_util.py
+49
View File
@@ -3,6 +3,7 @@
from __future__ import annotations
from pathlib import Path
from unittest.mock import patch
import pytest
@@ -402,3 +403,51 @@ def test_shlex_quote_edge_cases() -> None:
assert util.shlex_quote("\t") == "'\t'"
assert util.shlex_quote("\n") == "'\n'"
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()
with (
patch("esphome.util.sys") as mock_sys,
patch("esphome.util.Path") as mock_path_cls,
):
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_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"