From 0c883b80c4376906eb5dc67f4b77d1249ee67b1d Mon Sep 17 00:00:00 2001 From: Jonathan Swoboda <154711427+swoboda1337@users.noreply.github.com> Date: Wed, 4 Mar 2026 16:05:49 -0500 Subject: [PATCH 1/7] [inkplate][ezo_pmp][ezo][packet_transport] Fix use-after-free bugs (#14467) Co-authored-by: Claude Opus 4.6 --- esphome/components/ezo/ezo.cpp | 3 +- esphome/components/ezo_pmp/ezo_pmp.cpp | 36 ++++++++++--------- esphome/components/ezo_pmp/ezo_pmp.h | 2 +- esphome/components/inkplate/inkplate.cpp | 20 ++++++++--- .../packet_transport/packet_transport.cpp | 2 +- .../packet_transport/packet_transport.h | 2 +- 6 files changed, 39 insertions(+), 26 deletions(-) diff --git a/esphome/components/ezo/ezo.cpp b/esphome/components/ezo/ezo.cpp index e4036021df..2dc65b7d14 100644 --- a/esphome/components/ezo/ezo.cpp +++ b/esphome/components/ezo/ezo.cpp @@ -66,8 +66,9 @@ void EZOSensor::loop() { if (to_run->command_type == EzoCommandType::EZO_SLEEP || to_run->command_type == EzoCommandType::EZO_I2C) { // Commands with no return data + bool update_address = to_run->command_type == EzoCommandType::EZO_I2C; this->commands_.pop_front(); - if (to_run->command_type == EzoCommandType::EZO_I2C) + if (update_address) this->address_ = this->new_address_; return; } diff --git a/esphome/components/ezo_pmp/ezo_pmp.cpp b/esphome/components/ezo_pmp/ezo_pmp.cpp index bf6e3926b8..4ce4da57ff 100644 --- a/esphome/components/ezo_pmp/ezo_pmp.cpp +++ b/esphome/components/ezo_pmp/ezo_pmp.cpp @@ -165,22 +165,23 @@ void EzoPMP::read_command_result_() { continue; } - switch (current_parameter) { - case 1: - first_parameter_buffer[position_in_parameter_buffer] = current_char; - first_parameter_buffer[position_in_parameter_buffer + 1] = '\0'; - break; - case 2: - second_parameter_buffer[position_in_parameter_buffer] = current_char; - second_parameter_buffer[position_in_parameter_buffer + 1] = '\0'; - break; - case 3: - third_parameter_buffer[position_in_parameter_buffer] = current_char; - third_parameter_buffer[position_in_parameter_buffer + 1] = '\0'; - break; + if (position_in_parameter_buffer < sizeof(first_parameter_buffer) - 1) { + switch (current_parameter) { + case 1: + first_parameter_buffer[position_in_parameter_buffer] = current_char; + first_parameter_buffer[position_in_parameter_buffer + 1] = '\0'; + break; + case 2: + second_parameter_buffer[position_in_parameter_buffer] = current_char; + second_parameter_buffer[position_in_parameter_buffer + 1] = '\0'; + break; + case 3: + third_parameter_buffer[position_in_parameter_buffer] = current_char; + third_parameter_buffer[position_in_parameter_buffer + 1] = '\0'; + break; + } + position_in_parameter_buffer++; } - - position_in_parameter_buffer++; } auto parsed_first_parameter = parse_number(first_parameter_buffer); @@ -404,7 +405,8 @@ void EzoPMP::send_next_command_() { break; case EZO_PMP_COMMAND_EXEC_ARBITRARY_COMMAND_ADDRESS: // Run an arbitrary command - command_buffer_length = snprintf((char *) command_buffer, sizeof(command_buffer), "%s", this->arbitrary_command_); + command_buffer_length = + snprintf((char *) command_buffer, sizeof(command_buffer), "%s", this->arbitrary_command_.c_str()); ESP_LOGI(TAG, "Sending arbitrary command: %s", (char *) command_buffer); break; @@ -541,7 +543,7 @@ void EzoPMP::change_i2c_address(int address) { } void EzoPMP::exec_arbitrary_command(const std::basic_string &command) { - this->arbitrary_command_ = command.c_str(); + this->arbitrary_command_ = command; this->queue_command_(EZO_PMP_COMMAND_EXEC_ARBITRARY_COMMAND_ADDRESS, 0, 0, true); } diff --git a/esphome/components/ezo_pmp/ezo_pmp.h b/esphome/components/ezo_pmp/ezo_pmp.h index d4917e7f4b..bbfd899170 100644 --- a/esphome/components/ezo_pmp/ezo_pmp.h +++ b/esphome/components/ezo_pmp/ezo_pmp.h @@ -85,7 +85,7 @@ class EzoPMP : public PollingComponent, public i2c::I2CDevice { bool is_paused_flag_ = false; bool is_dosing_flag_ = false; - const char *arbitrary_command_{nullptr}; + std::string arbitrary_command_{}; void send_next_command_(); void read_command_result_(); diff --git a/esphome/components/inkplate/inkplate.cpp b/esphome/components/inkplate/inkplate.cpp index c921c643fa..df9c2b29c7 100644 --- a/esphome/components/inkplate/inkplate.cpp +++ b/esphome/components/inkplate/inkplate.cpp @@ -63,16 +63,26 @@ void Inkplate::initialize_() { if (buffer_size == 0) return; - if (this->partial_buffer_ != nullptr) + if (this->partial_buffer_ != nullptr) { allocator.deallocate(this->partial_buffer_, buffer_size); - if (this->partial_buffer_2_ != nullptr) + this->partial_buffer_ = nullptr; + } + if (this->partial_buffer_2_ != nullptr) { allocator.deallocate(this->partial_buffer_2_, buffer_size * 2); - if (this->buffer_ != nullptr) + this->partial_buffer_2_ = nullptr; + } + if (this->buffer_ != nullptr) { allocator.deallocate(this->buffer_, buffer_size); - if (this->glut_ != nullptr) + this->buffer_ = nullptr; + } + if (this->glut_ != nullptr) { allocator32.deallocate(this->glut_, 256 * 9); - if (this->glut2_ != nullptr) + this->glut_ = nullptr; + } + if (this->glut2_ != nullptr) { allocator32.deallocate(this->glut2_, 256 * 9); + this->glut2_ = nullptr; + } this->buffer_ = allocator.allocate(buffer_size); if (this->buffer_ == nullptr) { diff --git a/esphome/components/packet_transport/packet_transport.cpp b/esphome/components/packet_transport/packet_transport.cpp index 7b7a852398..d2c5920001 100644 --- a/esphome/components/packet_transport/packet_transport.cpp +++ b/esphome/components/packet_transport/packet_transport.cpp @@ -249,7 +249,7 @@ void PacketTransport::init_data_() { } else { add(this->data_, DATA_KEY); } - for (auto pkey : this->ping_keys_) { + for (const auto &pkey : this->ping_keys_) { add(this->data_, PING_KEY); add(this->data_, pkey.second); } diff --git a/esphome/components/packet_transport/packet_transport.h b/esphome/components/packet_transport/packet_transport.h index 57f40874b5..a236744231 100644 --- a/esphome/components/packet_transport/packet_transport.h +++ b/esphome/components/packet_transport/packet_transport.h @@ -150,7 +150,7 @@ class PacketTransport : public PollingComponent { std::vector ping_header_{}; std::vector header_{}; std::vector data_{}; - std::map ping_keys_{}; + std::map ping_keys_{}; const char *platform_name_{""}; void add_key_(const char *name, uint32_t key); void send_ping_pong_request_(); From e11a91411b800497b812a9baf374fca36af5e085 Mon Sep 17 00:00:00 2001 From: Jonathan Swoboda <154711427+swoboda1337@users.noreply.github.com> Date: Wed, 4 Mar 2026 16:36:52 -0500 Subject: [PATCH 2/7] [esp32_improv][rf_bridge][esp32_ble_server][display][lvgl][pipsolar] Fix unsigned integer underflows (#14466) Co-authored-by: Claude Opus 4.6 Co-authored-by: Clyde Stubbs <2366188+clydebarrow@users.noreply.github.com> --- esphome/components/display/display.cpp | 3 +++ esphome/components/esp32_ble_server/ble_characteristic.cpp | 6 +++++- esphome/components/esp32_improv/esp32_improv_component.cpp | 2 ++ esphome/components/lvgl/lvgl_esphome.cpp | 2 +- esphome/components/pipsolar/pipsolar.cpp | 4 +++- esphome/components/rf_bridge/rf_bridge.cpp | 2 +- 6 files changed, 15 insertions(+), 4 deletions(-) diff --git a/esphome/components/display/display.cpp b/esphome/components/display/display.cpp index 2bd7d03600..f8569b6e7c 100644 --- a/esphome/components/display/display.cpp +++ b/esphome/components/display/display.cpp @@ -661,6 +661,9 @@ void Display::printf(int x, int y, BaseFont *font, const char *format, ...) { void Display::set_writer(display_writer_t &&writer) { this->writer_ = writer; } void Display::set_pages(std::vector pages) { + if (pages.empty()) + return; + for (auto *page : pages) page->set_parent(this); diff --git a/esphome/components/esp32_ble_server/ble_characteristic.cpp b/esphome/components/esp32_ble_server/ble_characteristic.cpp index a1b1ff94bb..d4ccefd9b2 100644 --- a/esphome/components/esp32_ble_server/ble_characteristic.cpp +++ b/esphome/components/esp32_ble_server/ble_characteristic.cpp @@ -209,7 +209,11 @@ void BLECharacteristic::gatts_event_handler(esp_gatts_cb_event_t event, esp_gatt esp_gatt_rsp_t response; if (param->read.is_long) { - if (this->value_.size() - this->value_read_offset_ < max_offset) { + if (this->value_read_offset_ >= this->value_.size()) { + response.attr_value.len = 0; + response.attr_value.offset = this->value_read_offset_; + this->value_read_offset_ = 0; + } else if (this->value_.size() - this->value_read_offset_ < max_offset) { // Last message in the chain response.attr_value.len = this->value_.size() - this->value_read_offset_; response.attr_value.offset = this->value_read_offset_; diff --git a/esphome/components/esp32_improv/esp32_improv_component.cpp b/esphome/components/esp32_improv/esp32_improv_component.cpp index 83bc842a3d..e4ae49f235 100644 --- a/esphome/components/esp32_improv/esp32_improv_component.cpp +++ b/esphome/components/esp32_improv/esp32_improv_component.cpp @@ -314,6 +314,8 @@ void ESP32ImprovComponent::dump_config() { } void ESP32ImprovComponent::process_incoming_data_() { + if (this->incoming_data_.size() < 3) + return; uint8_t length = this->incoming_data_[1]; #if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_VERBOSE diff --git a/esphome/components/lvgl/lvgl_esphome.cpp b/esphome/components/lvgl/lvgl_esphome.cpp index bb373abb88..3e447e9169 100644 --- a/esphome/components/lvgl/lvgl_esphome.cpp +++ b/esphome/components/lvgl/lvgl_esphome.cpp @@ -421,7 +421,7 @@ void LvglComponent::write_random_() { col = col / this->draw_rounding * this->draw_rounding; auto row = random_uint32() % this->disp_drv_.ver_res; row = row / this->draw_rounding * this->draw_rounding; - auto size = (random_uint32() % 32) / this->draw_rounding * this->draw_rounding - 1; + auto size = ((random_uint32() % 32) / this->draw_rounding + 2) * this->draw_rounding - 1; lv_area_t area; area.x1 = col; area.y1 = row; diff --git a/esphome/components/pipsolar/pipsolar.cpp b/esphome/components/pipsolar/pipsolar.cpp index f95bf4aedb..9c5caec775 100644 --- a/esphome/components/pipsolar/pipsolar.cpp +++ b/esphome/components/pipsolar/pipsolar.cpp @@ -162,13 +162,15 @@ void Pipsolar::loop() { } uint8_t Pipsolar::check_incoming_length_(uint8_t length) { - if (this->read_pos_ - 3 == length) { + if (this->read_pos_ >= 3 && this->read_pos_ - 3 == length) { return 1; } return 0; } uint8_t Pipsolar::check_incoming_crc_() { + if (this->read_pos_ < 3) + return 0; uint16_t crc16; crc16 = this->pipsolar_crc_(read_buffer_, read_pos_ - 3); if (((uint8_t) ((crc16) >> 8)) == read_buffer_[read_pos_ - 3] && diff --git a/esphome/components/rf_bridge/rf_bridge.cpp b/esphome/components/rf_bridge/rf_bridge.cpp index d8c148145c..700e2ba162 100644 --- a/esphome/components/rf_bridge/rf_bridge.cpp +++ b/esphome/components/rf_bridge/rf_bridge.cpp @@ -74,7 +74,7 @@ bool RFBridgeComponent::parse_bridge_byte_(uint8_t byte) { data.length = raw[2]; data.protocol = raw[3]; char next_byte[3]; // 2 hex chars + null - for (uint8_t i = 0; i < data.length - 1; i++) { + for (uint8_t i = 0; i + 1 < data.length; i++) { buf_append_printf(next_byte, sizeof(next_byte), 0, "%02X", raw[4 + i]); data.code += next_byte; } From 61ea6c3b2f759c325c1428b371e4e9f8a00425fb Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Wed, 4 Mar 2026 12:46:26 -1000 Subject: [PATCH 3/7] [ci] Add missing issues: write permission to codeowner approval workflow (#14477) Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- .github/workflows/codeowner-approved-label.yml | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/.github/workflows/codeowner-approved-label.yml b/.github/workflows/codeowner-approved-label.yml index 217ae06419..200f18f544 100644 --- a/.github/workflows/codeowner-approved-label.yml +++ b/.github/workflows/codeowner-approved-label.yml @@ -12,7 +12,8 @@ on: types: [submitted, dismissed] permissions: - pull-requests: write + issues: write + pull-requests: read contents: read jobs: From 74dd61442aa82f7da65ada2b47d3fdbd1bb93318 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Wed, 4 Mar 2026 13:50:52 -1000 Subject: [PATCH 4/7] [rp2040] Improve upload experience with mass storage and BOOTSEL support 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 --- esphome/__main__.py | 155 ++++++++++- .../components/rp2040/post_build.py.script | 15 ++ esphome/espota2.py | 26 +- esphome/helpers.py | 27 ++ esphome/util.py | 62 +++++ tests/unit_tests/test_main.py | 251 ++++++++++++++++++ tests/unit_tests/test_util.py | 49 ++++ 7 files changed, 558 insertions(+), 27 deletions(-) diff --git a/esphome/__main__.py b/esphome/__main__.py index 0164e2eeb3..656bc2a2df 100644 --- a/esphome/__main__.py +++ b/esphome/__main__.py @@ -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, diff --git a/esphome/components/rp2040/post_build.py.script b/esphome/components/rp2040/post_build.py.script index 7dcd7e52a6..1f000ac78e 100644 --- a/esphome/components/rp2040/post_build.py.script +++ b/esphome/components/rp2040/post_build.py.script @@ -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 diff --git a/esphome/espota2.py b/esphome/espota2.py index c342eb4463..c412bb51ff 100644 --- a/esphome/espota2.py +++ b/esphome/espota2.py @@ -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 diff --git a/esphome/helpers.py b/esphome/helpers.py index 145ebd4096..f41bec357d 100644 --- a/esphome/helpers.py +++ b/esphome/helpers.py @@ -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 diff --git a/esphome/util.py b/esphome/util.py index 686aa74306..b1314f0518 100644 --- a/esphome/util.py +++ b/esphome/util.py @@ -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//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 + + 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 diff --git a/tests/unit_tests/test_main.py b/tests/unit_tests/test_main.py index cef561c54b..e6ab5494e1 100644 --- a/tests/unit_tests/test_main.py +++ b/tests/unit_tests/test_main.py @@ -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.""" diff --git a/tests/unit_tests/test_util.py b/tests/unit_tests/test_util.py index 85873caea8..853c48be23 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 +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" From 5f79e3e0c250c4031d2993f5781dbc1b9cb1ea6f Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Wed, 4 Mar 2026 14:00:35 -1000 Subject: [PATCH 5/7] [rp2040] Fix test assertions for Windows path separators Use str(Path(...)) in test assertions so paths match platform-specific separators (forward slashes on Unix, backslashes on Windows). --- tests/unit_tests/test_main.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/tests/unit_tests/test_main.py b/tests/unit_tests/test_main.py index e6ab5494e1..115cdfe9d9 100644 --- a/tests/unit_tests/test_main.py +++ b/tests/unit_tests/test_main.py @@ -879,8 +879,9 @@ 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( - [("/Volumes/RPI-RP2 (RP2040 BOOTSEL)", "MS:/Volumes/RPI-RP2")], + [(f"{vol_path} (RP2040 BOOTSEL)", f"MS:{vol_path}")], purpose=Purpose.UPLOADING, ) @@ -983,10 +984,11 @@ def test_choose_upload_log_host_rp2040_serial_and_mass_storage( 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"), - ("/Volumes/RPI-RP2 (RP2040 BOOTSEL)", "MS:/Volumes/RPI-RP2"), + (f"{vol_path} (RP2040 BOOTSEL)", f"MS:{vol_path}"), ], purpose=Purpose.UPLOADING, ) From c6c0dc62ebe42fcc77405181784f7afcd0c55026 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Wed, 4 Mar 2026 14:03:28 -1000 Subject: [PATCH 6/7] [rp2040] Address review feedback for upload improvements - Fix _wait_for_serial_port to use get_serial_ports() instead of os.access() so it works on Windows COM ports - Snapshot serial ports before upload and wait for new ports to appear, preventing false matches on pre-existing serial devices - Auto-select only newly appeared ports after mass storage upload - Guard against ZeroDivisionError if UF2 file is empty --- esphome/__main__.py | 45 +++++++++++++++++++++++++++++++-------------- 1 file changed, 31 insertions(+), 14 deletions(-) diff --git a/esphome/__main__.py b/esphome/__main__.py index 656bc2a2df..228b46aaa2 100644 --- a/esphome/__main__.py +++ b/esphome/__main__.py @@ -775,6 +775,9 @@ def upload_using_uf2_copy(config: ConfigType, mount_path: str) -> int: 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() @@ -805,16 +808,30 @@ def upload_using_uf2_copy(config: ConfigType, mount_path: str) -> int: return 0 -def _wait_for_serial_port(port: str | None = None, timeout: float = 30.0) -> None: +def _wait_for_serial_port( + port: str | None = None, + timeout: float = 30.0, + known_ports: set[str] | None = None, +) -> 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 given, wait for that specific path. If known_ports is + given, wait for a new port that wasn't in the set. Otherwise wait + for any serial port to appear. """ - if port is not None and os.access(port, os.F_OK): + + def _port_found() -> bool: + ports = get_serial_ports() + if port is not None: + return any(p.path == port for p in ports) + if known_ports is not None: + return any(p.path not in known_ports for p in ports) + return bool(ports) + + if _port_found(): return if port is not None: _LOGGER.info("Waiting for %s to come online...", port) @@ -823,11 +840,7 @@ def _wait_for_serial_port(port: str | None = None, timeout: float = 30.0) -> Non 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(): + if _port_found(): time.sleep(0.05) return @@ -1064,6 +1077,9 @@ def command_run(args: ArgsProtocol, config: ConfigType) -> int | None: purpose=Purpose.UPLOADING, ) + # Snapshot current serial ports before upload so we can detect new ones + pre_upload_ports = {p.path for p in get_serial_ports()} + exit_code, successful_device = upload_program(config, args, devices) if exit_code == 0: _LOGGER.info("Successfully uploaded program.") @@ -1074,17 +1090,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 + # After mass storage upload, wait for a new serial port to appear # 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 + _wait_for_serial_port(known_ports=pre_upload_ports) + # If exactly one new serial port appeared, use it directly serial_ports = get_serial_ports() - if len(serial_ports) == 1: - successful_device = serial_ports[0].path + new_ports = [p for p in serial_ports if p.path not in pre_upload_ports] + if len(new_ports) == 1: + successful_device = new_ports[0].path # For logs, prefer the device we successfully uploaded to devices = choose_upload_log_host( From d8560468309bf84d9ff7c6c70cc82195c7a9f8d1 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Wed, 4 Mar 2026 14:12:39 -1000 Subject: [PATCH 7/7] [rp2040] Add test coverage for Linux and Windows mass storage detection Add tests for get_rp2040_mass_storage_volumes() on Linux, Windows, and unsupported platforms to improve test coverage. --- tests/unit_tests/test_util.py | 90 ++++++++++++++++++++++++++++++++++- 1 file changed, 89 insertions(+), 1 deletion(-) diff --git a/tests/unit_tests/test_util.py b/tests/unit_tests/test_util.py index 853c48be23..7fd3d4b851 100644 --- a/tests/unit_tests/test_util.py +++ b/tests/unit_tests/test_util.py @@ -3,7 +3,7 @@ from __future__ import annotations from pathlib import Path -from unittest.mock import patch +from unittest.mock import MagicMock, patch import pytest @@ -446,6 +446,94 @@ def test_get_rp2040_mass_storage_volumes_none_found(tmp_path: Path) -> None: 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")