github/esphome
2
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-09-15 17:18:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'rp2040-upload-improvements' into integration

Browse Source
This commit is contained in:
J. Nick Koston
2026-03-04 14:18:29 -10:00
parent c725152431 d856046830
commit d30175d84d
20 changed files with 721 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/codeowner-approved-label.yml
+2 -1
View File
@@ -12,7 +12,8 @@ on:
types: [submitted, dismissed]
permissions:
pull-requests: write
issues: write
pull-requests: read
contents: read
jobs:
esphome/__main__.py
+170 -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,107 @@ 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
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()
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,
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. If known_ports is
given, wait for a new port that wasn't in the set. Otherwise wait
for any serial port to appear.
"""
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)
else:
_LOGGER.info("Waiting for device to reboot...")
start = time.monotonic()
while time.monotonic() - start < timeout:
time.sleep(0.05)
if _port_found():
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 +874,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 +938,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)
@@ -925,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.")
@@ -935,6 +1090,19 @@ 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
# 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(known_ports=pre_upload_ports)
# If exactly one new serial port appeared, use it directly
serial_ports = get_serial_ports()
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(
default=successful_device,
esphome/components/display/display.cpp
+3
View File
@@ -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<DisplayPage *> pages) {
if (pages.empty())
return;
for (auto *page : pages)
page->set_parent(this);
esphome/components/esp32_ble_server/ble_characteristic.cpp
+5 -1
View File
@@ -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_;
esphome/components/esp32_improv/esp32_improv_component.cpp
+2
View File
@@ -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
esphome/components/ezo/ezo.cpp
+2 -1
View File
@@ -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;
}
esphome/components/ezo_pmp/ezo_pmp.cpp
+19 -17
View File
@@ -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<float>(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<char> &command) {
this->arbitrary_command_ = command.c_str();
this->arbitrary_command_ = command;
this->queue_command_(EZO_PMP_COMMAND_EXEC_ARBITRARY_COMMAND_ADDRESS, 0, 0, true);
}
esphome/components/ezo_pmp/ezo_pmp.h
+1 -1
View File
@@ -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_();
esphome/components/inkplate/inkplate.cpp
+15 -5
View File
@@ -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) {
esphome/components/lvgl/lvgl_esphome.cpp
+1 -1
View File
@@ -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;
esphome/components/packet_transport/packet_transport.cpp
+1 -1
View File
@@ -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);
}
esphome/components/packet_transport/packet_transport.h
+1 -1
View File
@@ -150,7 +150,7 @@ class PacketTransport : public PollingComponent {
std::vector<uint8_t> ping_header_{};
std::vector<uint8_t> header_{};
std::vector<uint8_t> data_{};
std::map<const char *, uint32_t> ping_keys_{};
std::map<std::string, uint32_t> ping_keys_{};
const char *platform_name_{""};
void add_key_(const char *name, uint32_t key);
void send_ping_pong_request_();
esphome/components/pipsolar/pipsolar.cpp
+3 -1
View File
@@ -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] &&
esphome/components/rf_bridge/rf_bridge.cpp
+1 -1
View File
@@ -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;
}
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
+253
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,141 @@ 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
vol_path = str(Path("/Volumes/RPI-RP2"))
mock_choose_prompt.assert_called_once_with(
[(f"{vol_path} (RP2040 BOOTSEL)", f"MS:{vol_path}")],
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,
)
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}"),
],
purpose=Purpose.UPLOADING,
)
@dataclass
class MockArgs:
"""Mock args for testing."""
@@ -1082,6 +1225,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 +1855,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
+137
View File
@@ -3,6 +3,7 @@
from __future__ import annotations
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
@@ -402,3 +403,139 @@ 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_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"