mirror of
https://github.com/esphome/esphome.git
synced 2026-10-07 11:26:39 +00:00
Merge branch 'dev' into configure_entity
This commit is contained in:
@@ -3,3 +3,7 @@ audio_file:
|
||||
file:
|
||||
type: local
|
||||
path: $component_dir/test.wav
|
||||
|
||||
media_source:
|
||||
- platform: audio_file
|
||||
id: audio_file_source
|
||||
|
||||
@@ -0,0 +1,4 @@
|
||||
ble_nus:
|
||||
type: uart
|
||||
tx_buffer_size: 160
|
||||
rx_buffer_size: 160
|
||||
@@ -1,3 +1,5 @@
|
||||
modbus:
|
||||
id: mod_bus1
|
||||
flow_control_pin: ${flow_control_pin}
|
||||
send_wait_time: 500ms
|
||||
turnaround_time: 100ms
|
||||
|
||||
@@ -12,7 +12,9 @@
|
||||
using namespace esphome;
|
||||
|
||||
void setup() {
|
||||
App.pre_setup("livingroom", "LivingRoom", false);
|
||||
static char name[] = "livingroom";
|
||||
static char friendly_name[] = "LivingRoom";
|
||||
App.pre_setup(name, sizeof(name) - 1, friendly_name, sizeof(friendly_name) - 1);
|
||||
auto *log = new logger::Logger(115200); // NOLINT
|
||||
log->pre_setup();
|
||||
log->set_uart_selection(logger::UART_SELECTION_UART0);
|
||||
|
||||
@@ -71,6 +71,7 @@ RESPONSE_SCHEMA = cv.Schema(
|
||||
{
|
||||
cv.Required(CONF_EXPECT_TX): [cv.hex_uint8_t],
|
||||
cv.Required(CONF_INJECT_RX): [cv.hex_uint8_t],
|
||||
cv.Optional(CONF_DELAY, default="0ms"): cv.positive_time_period_milliseconds,
|
||||
}
|
||||
)
|
||||
|
||||
@@ -151,7 +152,8 @@ async def to_code(config):
|
||||
for response in config[CONF_RESPONSES]:
|
||||
tx_data = response[CONF_EXPECT_TX]
|
||||
rx_data = response[CONF_INJECT_RX]
|
||||
cg.add(var.add_response(tx_data, rx_data))
|
||||
delay_ms = response[CONF_DELAY]
|
||||
cg.add(var.add_response(tx_data, rx_data, delay_ms))
|
||||
|
||||
for periodic in config[CONF_PERIODIC_RX]:
|
||||
data = periodic[CONF_DATA]
|
||||
|
||||
@@ -36,8 +36,8 @@ void MockUartComponent::loop() {
|
||||
// component (e.g., LD2410) a chance to process each batch independently.
|
||||
if (this->injection_index_ < this->injections_.size()) {
|
||||
auto &injection = this->injections_[this->injection_index_];
|
||||
uint32_t target_time = this->scenario_start_ms_ + this->cumulative_delay_ms_ + injection.delay_ms;
|
||||
if (now >= target_time) {
|
||||
uint32_t total_delay = this->cumulative_delay_ms_ + injection.delay_ms;
|
||||
if (now - this->scenario_start_ms_ >= total_delay) {
|
||||
ESP_LOGD(TAG, "Injecting %zu RX bytes (injection %u)", injection.rx_data.size(), this->injection_index_);
|
||||
this->inject_to_rx_buffer(injection.rx_data);
|
||||
this->cumulative_delay_ms_ += injection.delay_ms;
|
||||
@@ -52,6 +52,15 @@ void MockUartComponent::loop() {
|
||||
periodic.last_inject_ms = now;
|
||||
}
|
||||
}
|
||||
|
||||
// Process delayed responses
|
||||
for (auto &response : this->responses_) {
|
||||
if (response.delay_ms > 0 && response.last_match_ms > 0 && now - response.last_match_ms >= response.delay_ms) {
|
||||
ESP_LOGD(TAG, "Injecting %zu RX bytes for delayed response", response.inject_rx.size());
|
||||
this->inject_to_rx_buffer(response.inject_rx);
|
||||
response.last_match_ms = 0; // Reset to prevent repeated injection
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void MockUartComponent::start_scenario() {
|
||||
@@ -149,8 +158,9 @@ void MockUartComponent::add_injection(const std::vector<uint8_t> &rx_data, uint3
|
||||
this->injections_.push_back({rx_data, delay_ms});
|
||||
}
|
||||
|
||||
void MockUartComponent::add_response(const std::vector<uint8_t> &expect_tx, const std::vector<uint8_t> &inject_rx) {
|
||||
this->responses_.push_back({expect_tx, inject_rx});
|
||||
void MockUartComponent::add_response(const std::vector<uint8_t> &expect_tx, const std::vector<uint8_t> &inject_rx,
|
||||
uint32_t delay_ms) {
|
||||
this->responses_.push_back({expect_tx, inject_rx, delay_ms, 0});
|
||||
}
|
||||
|
||||
void MockUartComponent::add_periodic_rx(const std::vector<uint8_t> &data, uint32_t interval_ms) {
|
||||
@@ -166,7 +176,13 @@ void MockUartComponent::try_match_response_() {
|
||||
size_t offset = this->tx_buffer_.size() - response.expect_tx.size();
|
||||
if (std::equal(response.expect_tx.begin(), response.expect_tx.end(), this->tx_buffer_.begin() + offset)) {
|
||||
ESP_LOGD(TAG, "TX match found, injecting %zu RX bytes", response.inject_rx.size());
|
||||
this->inject_to_rx_buffer(response.inject_rx);
|
||||
if (response.delay_ms > 0) {
|
||||
ESP_LOGD(TAG, "Delaying response by %u ms", response.delay_ms);
|
||||
// Schedule the response injection as a future injection
|
||||
response.last_match_ms = App.get_loop_component_start_time();
|
||||
} else {
|
||||
this->inject_to_rx_buffer(response.inject_rx);
|
||||
}
|
||||
this->tx_buffer_.clear();
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -34,7 +34,8 @@ class MockUartComponent : public uart::UARTComponent, public Component {
|
||||
|
||||
// Scenario configuration - called from generated code
|
||||
void add_injection(const std::vector<uint8_t> &rx_data, uint32_t delay_ms);
|
||||
void add_response(const std::vector<uint8_t> &expect_tx, const std::vector<uint8_t> &inject_rx);
|
||||
void add_response(const std::vector<uint8_t> &expect_tx, const std::vector<uint8_t> &inject_rx,
|
||||
uint32_t delay_ms = 0);
|
||||
void add_periodic_rx(const std::vector<uint8_t> &data, uint32_t interval_ms);
|
||||
|
||||
void start_scenario();
|
||||
@@ -64,6 +65,8 @@ class MockUartComponent : public uart::UARTComponent, public Component {
|
||||
struct Response {
|
||||
std::vector<uint8_t> expect_tx;
|
||||
std::vector<uint8_t> inject_rx;
|
||||
uint32_t delay_ms;
|
||||
uint32_t last_match_ms{0};
|
||||
};
|
||||
std::vector<Response> responses_;
|
||||
std::vector<uint8_t> tx_buffer_;
|
||||
|
||||
@@ -25,20 +25,64 @@ uart_mock:
|
||||
auto_start: false
|
||||
debug:
|
||||
responses:
|
||||
- expect_tx: [0x01, 0x03, 0x00, 0x03, 0x00, 0x01, 0x74, 0x0A] # Read holding register 1 on device 1
|
||||
- expect_tx: [0x01, 0x03, 0x00, 0x03, 0x00, 0x01, 0x74, 0x0A] # Read holding register 3 on device 1 (basic_register)
|
||||
inject_rx: [0x01, 0x03, 0x02, 0x01, 0x03, 0xF9, 0xD5] # Return value 0x0103 (hex) = 259 (dec)
|
||||
- expect_tx: [0x01, 0x03, 0x00, 0x05, 0x00, 0x01, 0x94, 0x0B] # Read holding register 5 on device 1 (delayed_response)
|
||||
delay: 100ms # Shorter than modbus send_wait_time of 200ms, should succeed
|
||||
inject_rx: [0x01, 0x03, 0x02, 0x00, 0xFF, 0xF8, 0x04] # Return value 0x00FF (hex) = 255 (dec)
|
||||
- expect_tx: [0x02, 0x03, 0x00, 0x07, 0x00, 0x01, 0x35, 0xF8] # Read holding register 7 on device 2 (late_response)
|
||||
delay: 300ms # Longer than modbus send_wait_time of 200ms, should cause timeout
|
||||
inject_rx: [0x02, 0x03, 0x02, 0x00, 0xF0, 0xFC, 0x00] # Return value 0x00F0 (hex) = 240 (dec)
|
||||
- expect_tx: [0x03, 0x03, 0x00, 0x09, 0x00, 0x01, 0x55, 0xEA] # Read holding register 9 on device 3 (no_response)
|
||||
inject_rx: [] # No response, should cause timeout
|
||||
- expect_tx: [0x01, 0x03, 0x00, 0x0A, 0x00, 0x01, 0xA4, 0x08] # Read holding register A on device 1 (exception_response)
|
||||
inject_rx: [0x01, 0x83, 0x02, 0xC0, 0xF1] # Exception response with code 2 (illegal data address)
|
||||
|
||||
modbus:
|
||||
uart_id: virtual_uart_dev
|
||||
send_wait_time: 200ms
|
||||
turnaround_time: 10ms
|
||||
|
||||
modbus_controller:
|
||||
address: 1
|
||||
- address: 1
|
||||
id: modbus_controller_ok
|
||||
max_cmd_retries: 0
|
||||
update_interval: 1s
|
||||
- address: 2
|
||||
id: modbus_controller_slow
|
||||
max_cmd_retries: 0
|
||||
update_interval: 1s
|
||||
- address: 3
|
||||
id: modbus_controller_offline
|
||||
max_cmd_retries: 0
|
||||
update_interval: 1s
|
||||
|
||||
sensor:
|
||||
- platform: modbus_controller
|
||||
name: "basic_register"
|
||||
address: 0x03
|
||||
register_type: holding
|
||||
modbus_controller_id: modbus_controller_ok
|
||||
- platform: modbus_controller
|
||||
name: "delayed_response"
|
||||
address: 0x05
|
||||
register_type: holding
|
||||
modbus_controller_id: modbus_controller_ok
|
||||
- platform: modbus_controller
|
||||
name: "late_response"
|
||||
address: 0x07
|
||||
register_type: holding
|
||||
modbus_controller_id: modbus_controller_slow
|
||||
- platform: modbus_controller
|
||||
name: "no_response"
|
||||
address: 0x09
|
||||
register_type: holding
|
||||
modbus_controller_id: modbus_controller_offline
|
||||
- platform: modbus_controller
|
||||
name: "exception_response"
|
||||
address: 0x0A
|
||||
register_type: holding
|
||||
modbus_controller_id: modbus_controller_ok
|
||||
|
||||
button:
|
||||
- platform: template
|
||||
|
||||
@@ -46,10 +46,12 @@ uart_mock:
|
||||
|
||||
modbus:
|
||||
uart_id: virtual_uart_dev
|
||||
turnaround_time: 10ms
|
||||
|
||||
sensor:
|
||||
- platform: sdm_meter
|
||||
address: 2
|
||||
update_interval: 1s
|
||||
phase_a:
|
||||
voltage:
|
||||
name: sdm_voltage
|
||||
|
||||
@@ -39,9 +39,17 @@ async def test_uart_mock_modbus(
|
||||
# Track sensor state updates (after initial state is swallowed)
|
||||
sensor_states: dict[str, list[float]] = {
|
||||
"basic_register": [],
|
||||
"delayed_response": [],
|
||||
"late_response": [],
|
||||
"no_response": [],
|
||||
"exception_response": [],
|
||||
}
|
||||
|
||||
basic_register_changed = loop.create_future()
|
||||
delayed_response_changed = loop.create_future()
|
||||
late_response_changed = loop.create_future()
|
||||
no_response_changed = loop.create_future()
|
||||
exception_response_changed = loop.create_future()
|
||||
|
||||
def on_state(state: EntityState) -> None:
|
||||
if isinstance(state, SensorState) and not state.missing_state:
|
||||
@@ -54,6 +62,23 @@ async def test_uart_mock_modbus(
|
||||
and not basic_register_changed.done()
|
||||
):
|
||||
basic_register_changed.set_result(True)
|
||||
elif (
|
||||
sensor_name == "delayed_response"
|
||||
and state.state == 255.0
|
||||
and not delayed_response_changed.done()
|
||||
):
|
||||
delayed_response_changed.set_result(True)
|
||||
elif (
|
||||
sensor_name == "late_response" and not late_response_changed.done()
|
||||
):
|
||||
late_response_changed.set_result(True)
|
||||
elif sensor_name == "no_response" and not no_response_changed.done():
|
||||
no_response_changed.set_result(True)
|
||||
elif (
|
||||
sensor_name == "exception_response"
|
||||
and not exception_response_changed.done()
|
||||
):
|
||||
exception_response_changed.set_result(True)
|
||||
|
||||
async with (
|
||||
run_compiled(yaml_config),
|
||||
@@ -79,20 +104,52 @@ async def test_uart_mock_modbus(
|
||||
assert start_btn is not None, "Start Scenario button not found"
|
||||
client.button_command(start_btn.key)
|
||||
|
||||
try:
|
||||
await asyncio.wait_for(delayed_response_changed, timeout=2.0)
|
||||
except TimeoutError:
|
||||
pytest.fail(
|
||||
f"Timeout waiting for delayed_response change. Received sensor states:\n"
|
||||
f" delayed_response: {sensor_states['delayed_response']}\n"
|
||||
)
|
||||
|
||||
try:
|
||||
await asyncio.wait_for(late_response_changed, timeout=2.0)
|
||||
pytest.fail(
|
||||
f"late_response change should not have been triggered, but was. Received sensor states:\n"
|
||||
f" late_response: {sensor_states['late_response']}\n"
|
||||
)
|
||||
except TimeoutError:
|
||||
pass # Expected timeout since we never inject a response for late_response
|
||||
|
||||
try:
|
||||
await asyncio.wait_for(no_response_changed, timeout=2.0)
|
||||
pytest.fail(
|
||||
f"no_response change should not have been triggered, but was. Received sensor states:\n"
|
||||
f" no_response: {sensor_states['no_response']}\n"
|
||||
)
|
||||
except TimeoutError:
|
||||
pass # Expected timeout since we never inject a response for no_response
|
||||
|
||||
# Wait for basic register to be updated with successful parse
|
||||
try:
|
||||
await asyncio.wait_for(basic_register_changed, timeout=15.0)
|
||||
await asyncio.wait_for(basic_register_changed, timeout=2.0)
|
||||
except TimeoutError:
|
||||
pytest.fail(
|
||||
f"Timeout waiting for Basic Register change. Received sensor states:\n"
|
||||
f" basic_register: {sensor_states['basic_register']}\n"
|
||||
)
|
||||
|
||||
try:
|
||||
await asyncio.wait_for(exception_response_changed, timeout=2.0)
|
||||
pytest.fail(
|
||||
f"exception_response change should not have been triggered, but was. Received sensor states:\n"
|
||||
f" exception_response: {sensor_states['exception_response']}\n"
|
||||
)
|
||||
except TimeoutError:
|
||||
pass
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.xfail(
|
||||
reason="There is a bug in UART which will timeout for long responses."
|
||||
)
|
||||
async def test_uart_mock_modbus_timing(
|
||||
yaml_config: str,
|
||||
run_compiled: RunCompiledFunction,
|
||||
@@ -155,7 +212,7 @@ async def test_uart_mock_modbus_timing(
|
||||
|
||||
# Wait for voltage to be updated with successful parse
|
||||
try:
|
||||
await asyncio.wait_for(voltage_changed, timeout=15.0)
|
||||
await asyncio.wait_for(voltage_changed, timeout=2.0)
|
||||
except TimeoutError:
|
||||
pytest.fail(
|
||||
f"Timeout waiting for SDM voltage change. Received sensor states:\n"
|
||||
|
||||
@@ -0,0 +1,273 @@
|
||||
"""Tests for rp2040 generate_boards.py."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
import textwrap
|
||||
|
||||
import pytest
|
||||
|
||||
from esphome.components.rp2040.generate_boards import load_boards, parse_variant_pins
|
||||
|
||||
PICO_PINS_HEADER = textwrap.dedent("""\
|
||||
#pragma once
|
||||
#define PIN_LED (25u)
|
||||
#define PIN_SERIAL1_TX (0u)
|
||||
#define PIN_SERIAL1_RX (1u)
|
||||
#define PIN_WIRE0_SDA (4u)
|
||||
#define PIN_WIRE0_SCL (5u)
|
||||
#define PIN_WIRE1_SDA (26u)
|
||||
#define PIN_WIRE1_SCL (27u)
|
||||
#define PIN_SPI0_MISO (16u)
|
||||
#define PIN_SPI0_MOSI (19u)
|
||||
#define PIN_SPI0_SCK (18u)
|
||||
#define PIN_SPI0_SS (17u)
|
||||
#include "../generic/common.h"
|
||||
""")
|
||||
|
||||
PICOW_PINS_HEADER = textwrap.dedent("""\
|
||||
#pragma once
|
||||
#include <cyw43_wrappers.h>
|
||||
#define PIN_LED (64u)
|
||||
#define PIN_WIRE0_SDA (4u)
|
||||
#define PIN_WIRE0_SCL (5u)
|
||||
#include "../generic/common.h"
|
||||
""")
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def arduino_pico(tmp_path: Path) -> Path:
|
||||
"""Create a minimal arduino-pico directory structure."""
|
||||
json_dir = tmp_path / "tools" / "json"
|
||||
json_dir.mkdir(parents=True)
|
||||
variants_dir = tmp_path / "variants"
|
||||
variants_dir.mkdir()
|
||||
|
||||
generic_dir = variants_dir / "generic"
|
||||
generic_dir.mkdir()
|
||||
(generic_dir / "common.h").write_text("#pragma once\n")
|
||||
|
||||
return tmp_path
|
||||
|
||||
|
||||
def _add_board(
|
||||
arduino_pico: Path,
|
||||
board_name: str,
|
||||
mcu: str = "rp2040",
|
||||
variant: str | None = None,
|
||||
vendor: str = "",
|
||||
name: str | None = None,
|
||||
pins_header: str | None = None,
|
||||
) -> None:
|
||||
"""Add a board JSON and variant to the fake arduino-pico tree."""
|
||||
if variant is None:
|
||||
variant = board_name
|
||||
if name is None:
|
||||
name = board_name
|
||||
|
||||
json_dir = arduino_pico / "tools" / "json"
|
||||
variants_dir = arduino_pico / "variants"
|
||||
|
||||
board_json = {
|
||||
"build": {
|
||||
"mcu": mcu,
|
||||
"variant": variant,
|
||||
},
|
||||
"name": name,
|
||||
"vendor": vendor,
|
||||
}
|
||||
(json_dir / f"{board_name}.json").write_text(json.dumps(board_json))
|
||||
|
||||
variant_dir = variants_dir / variant
|
||||
variant_dir.mkdir(exist_ok=True)
|
||||
if pins_header is not None:
|
||||
(variant_dir / "pins_arduino.h").write_text(pins_header)
|
||||
|
||||
|
||||
def test_parse_basic_pins(tmp_path: Path) -> None:
|
||||
variant_dir = tmp_path / "rpipico"
|
||||
variant_dir.mkdir()
|
||||
(variant_dir / "pins_arduino.h").write_text(PICO_PINS_HEADER)
|
||||
|
||||
pins = parse_variant_pins(variant_dir)
|
||||
assert pins["LED"] == 25
|
||||
assert pins["SDA"] == 4
|
||||
assert pins["SCL"] == 5
|
||||
assert pins["SDA1"] == 26
|
||||
assert pins["SCL1"] == 27
|
||||
assert pins["MISO"] == 16
|
||||
assert pins["MOSI"] == 19
|
||||
assert pins["SCK"] == 18
|
||||
assert pins["SS"] == 17
|
||||
assert pins["TX"] == 0
|
||||
assert pins["RX"] == 1
|
||||
|
||||
|
||||
def test_parse_cyw43_led_pin(tmp_path: Path) -> None:
|
||||
variant_dir = tmp_path / "rpipicow"
|
||||
variant_dir.mkdir()
|
||||
(variant_dir / "pins_arduino.h").write_text(PICOW_PINS_HEADER)
|
||||
|
||||
pins = parse_variant_pins(variant_dir)
|
||||
assert pins["LED"] == 64
|
||||
|
||||
|
||||
def test_parse_missing_header(tmp_path: Path) -> None:
|
||||
variant_dir = tmp_path / "noheader"
|
||||
variant_dir.mkdir()
|
||||
assert parse_variant_pins(variant_dir) == {}
|
||||
|
||||
|
||||
def test_parse_unmapped_defines_ignored(tmp_path: Path) -> None:
|
||||
variant_dir = tmp_path / "custom"
|
||||
variant_dir.mkdir()
|
||||
(variant_dir / "pins_arduino.h").write_text(
|
||||
"#define PIN_NEOPIXEL (16u)\n#define PIN_LED (25u)\n"
|
||||
)
|
||||
|
||||
pins = parse_variant_pins(variant_dir)
|
||||
assert "NEOPIXEL" not in pins
|
||||
assert pins["LED"] == 25
|
||||
|
||||
|
||||
def test_load_basic_board(arduino_pico: Path) -> None:
|
||||
_add_board(
|
||||
arduino_pico,
|
||||
"rpipico",
|
||||
vendor="Raspberry Pi",
|
||||
name="Pico",
|
||||
pins_header=PICO_PINS_HEADER,
|
||||
)
|
||||
|
||||
board_pins, boards = load_boards(arduino_pico)
|
||||
|
||||
assert "rpipico" in boards
|
||||
assert boards["rpipico"]["name"] == "Raspberry Pi Pico"
|
||||
assert boards["rpipico"]["mcu"] == "rp2040"
|
||||
assert boards["rpipico"]["max_pin"] == 29
|
||||
|
||||
assert "rpipico" in board_pins
|
||||
assert board_pins["rpipico"]["LED"] == 25
|
||||
assert board_pins["rpipico"]["SDA"] == 4
|
||||
|
||||
|
||||
def test_load_rp2350_board(arduino_pico: Path) -> None:
|
||||
_add_board(
|
||||
arduino_pico,
|
||||
"rpipico2",
|
||||
mcu="rp2350",
|
||||
vendor="Raspberry Pi",
|
||||
name="Pico 2",
|
||||
pins_header=PICO_PINS_HEADER,
|
||||
)
|
||||
|
||||
_, boards = load_boards(arduino_pico)
|
||||
|
||||
assert boards["rpipico2"]["mcu"] == "rp2350"
|
||||
assert boards["rpipico2"]["max_pin"] == 47
|
||||
|
||||
|
||||
def test_cyw43_board_has_max_virtual_pin(arduino_pico: Path) -> None:
|
||||
_add_board(
|
||||
arduino_pico,
|
||||
"rpipicow",
|
||||
vendor="Raspberry Pi",
|
||||
name="Pico W",
|
||||
pins_header=PICOW_PINS_HEADER,
|
||||
)
|
||||
|
||||
_, boards = load_boards(arduino_pico)
|
||||
|
||||
assert boards["rpipicow"]["max_virtual_pin"] == 64
|
||||
|
||||
|
||||
def test_non_cyw43_board_has_no_max_virtual_pin(arduino_pico: Path) -> None:
|
||||
_add_board(
|
||||
arduino_pico,
|
||||
"rpipico",
|
||||
vendor="Raspberry Pi",
|
||||
name="Pico",
|
||||
pins_header=PICO_PINS_HEADER,
|
||||
)
|
||||
|
||||
_, boards = load_boards(arduino_pico)
|
||||
|
||||
assert "max_virtual_pin" not in boards["rpipico"]
|
||||
|
||||
|
||||
def test_board_without_variant_header(arduino_pico: Path) -> None:
|
||||
_add_board(arduino_pico, "novariant", name="No Variant")
|
||||
|
||||
board_pins, boards = load_boards(arduino_pico)
|
||||
|
||||
assert "novariant" in boards
|
||||
assert "novariant" not in board_pins
|
||||
|
||||
|
||||
def test_shared_variant_deduplicates(arduino_pico: Path) -> None:
|
||||
"""Two boards sharing the same variant should alias."""
|
||||
_add_board(arduino_pico, "base_board", pins_header=PICO_PINS_HEADER)
|
||||
_add_board(arduino_pico, "alias_board", variant="base_board")
|
||||
|
||||
board_pins, _ = load_boards(arduino_pico)
|
||||
|
||||
assert board_pins["base_board"] == parse_variant_pins(
|
||||
arduino_pico / "variants" / "base_board"
|
||||
)
|
||||
assert board_pins["alias_board"] == "base_board"
|
||||
|
||||
|
||||
def test_display_name_with_vendor(arduino_pico: Path) -> None:
|
||||
_add_board(arduino_pico, "testboard", vendor="Acme", name="Widget")
|
||||
_, boards = load_boards(arduino_pico)
|
||||
assert boards["testboard"]["name"] == "Acme Widget"
|
||||
|
||||
|
||||
def test_display_name_without_vendor(arduino_pico: Path) -> None:
|
||||
_add_board(arduino_pico, "testboard", vendor="", name="Widget")
|
||||
_, boards = load_boards(arduino_pico)
|
||||
assert boards["testboard"]["name"] == "Widget"
|
||||
|
||||
|
||||
def test_unknown_mcu_gets_default_max_pin(arduino_pico: Path) -> None:
|
||||
_add_board(arduino_pico, "future", mcu="rp2450", pins_header=PICO_PINS_HEADER)
|
||||
_, boards = load_boards(arduino_pico)
|
||||
assert boards["future"]["max_pin"] == 29
|
||||
|
||||
|
||||
def test_placeholder_pins_filtered_out(arduino_pico: Path) -> None:
|
||||
"""Pins with placeholder values like 99 should be filtered out."""
|
||||
header = textwrap.dedent("""\
|
||||
#pragma once
|
||||
#define PIN_LED (25u)
|
||||
#define PIN_WIRE0_SDA (4u)
|
||||
#define PIN_WIRE0_SCL (5u)
|
||||
#define PIN_WIRE1_SDA (99u)
|
||||
#define PIN_WIRE1_SCL (99u)
|
||||
""")
|
||||
_add_board(arduino_pico, "placeholder", pins_header=header)
|
||||
|
||||
board_pins, boards = load_boards(arduino_pico)
|
||||
|
||||
assert "SDA1" not in board_pins["placeholder"]
|
||||
assert "SCL1" not in board_pins["placeholder"]
|
||||
assert board_pins["placeholder"]["LED"] == 25
|
||||
assert "max_virtual_pin" not in boards["placeholder"]
|
||||
|
||||
|
||||
def test_placeholder_pins_not_treated_as_virtual(arduino_pico: Path) -> None:
|
||||
"""Pin 99 should not cause max_virtual_pin to be set."""
|
||||
header = textwrap.dedent("""\
|
||||
#pragma once
|
||||
#define PIN_LED (64u)
|
||||
#define PIN_WIRE0_SDA (4u)
|
||||
#define PIN_WIRE0_SCL (5u)
|
||||
#define PIN_SPI0_MISO (99u)
|
||||
""")
|
||||
_add_board(arduino_pico, "badpin", pins_header=header)
|
||||
|
||||
board_pins, boards = load_boards(arduino_pico)
|
||||
|
||||
assert "MISO" not in board_pins["badpin"]
|
||||
assert boards["badpin"]["max_virtual_pin"] == 64
|
||||
@@ -23,6 +23,7 @@ from esphome.const import (
|
||||
from esphome.core import CORE, config
|
||||
from esphome.core.config import (
|
||||
Area,
|
||||
make_app_name_cpp,
|
||||
preload_core_config,
|
||||
valid_include,
|
||||
valid_project_name,
|
||||
@@ -969,3 +970,79 @@ def test_config_hash_different_for_different_configs() -> None:
|
||||
hash2 = CORE.config_hash
|
||||
|
||||
assert hash1 != hash2
|
||||
|
||||
|
||||
def test_make_app_name_cpp_no_mac_simple() -> None:
|
||||
"""Test simple name without MAC suffix returns string literal."""
|
||||
cpp_expr, global_decl, byte_len = make_app_name_cpp(
|
||||
"my-device", "buf", "-", add_mac_suffix=False
|
||||
)
|
||||
assert cpp_expr == '"my-device"'
|
||||
assert global_decl is None
|
||||
assert byte_len == 9
|
||||
|
||||
|
||||
def test_make_app_name_cpp_no_mac_empty() -> None:
|
||||
"""Test empty name without MAC suffix."""
|
||||
cpp_expr, global_decl, byte_len = make_app_name_cpp(
|
||||
"", "buf", "-", add_mac_suffix=False
|
||||
)
|
||||
assert cpp_expr == '""'
|
||||
assert global_decl is None
|
||||
assert byte_len == 0
|
||||
|
||||
|
||||
def test_make_app_name_cpp_mac_suffix() -> None:
|
||||
"""Test name with MAC suffix emits static buffer."""
|
||||
cpp_expr, global_decl, byte_len = make_app_name_cpp(
|
||||
"my-device", "esphome_app_name_buf", "-", add_mac_suffix=True
|
||||
)
|
||||
assert cpp_expr == "esphome_app_name_buf"
|
||||
assert global_decl is not None
|
||||
assert "static char esphome_app_name_buf[]" in global_decl
|
||||
assert "my-device-XXXXXX" in global_decl
|
||||
assert byte_len == len("my-device-XXXXXX")
|
||||
|
||||
|
||||
def test_make_app_name_cpp_mac_suffix_empty() -> None:
|
||||
"""Test empty name with MAC suffix emits empty static buffer."""
|
||||
cpp_expr, global_decl, byte_len = make_app_name_cpp(
|
||||
"", "esphome_app_name_buf", "-", add_mac_suffix=True
|
||||
)
|
||||
assert cpp_expr == "esphome_app_name_buf"
|
||||
assert global_decl is not None
|
||||
assert "static char esphome_app_name_buf[]" in global_decl
|
||||
assert byte_len == 0
|
||||
|
||||
|
||||
def test_make_app_name_cpp_mac_suffix_space_sep() -> None:
|
||||
"""Test friendly name uses space separator for MAC suffix."""
|
||||
cpp_expr, global_decl, byte_len = make_app_name_cpp(
|
||||
"My Device", "esphome_app_friendly_name_buf", " ", add_mac_suffix=True
|
||||
)
|
||||
assert cpp_expr == "esphome_app_friendly_name_buf"
|
||||
assert global_decl is not None
|
||||
assert "My Device XXXXXX" in global_decl
|
||||
assert byte_len == len("My Device XXXXXX")
|
||||
|
||||
|
||||
def test_make_app_name_cpp_non_ascii_utf8_length() -> None:
|
||||
"""Test non-ASCII characters use UTF-8 byte length."""
|
||||
_, global_decl, byte_len = make_app_name_cpp(
|
||||
"café", "buf", "-", add_mac_suffix=False
|
||||
)
|
||||
assert byte_len == len("café".encode()) # 5 bytes, not 4 chars
|
||||
assert global_decl is None
|
||||
|
||||
|
||||
def test_make_app_name_cpp_non_ascii_mac_suffix_utf8_length() -> None:
|
||||
"""Test non-ASCII with MAC suffix uses UTF-8 byte length."""
|
||||
_, _, byte_len = make_app_name_cpp("café", "buf", "-", add_mac_suffix=True)
|
||||
assert byte_len == len("café-XXXXXX".encode())
|
||||
|
||||
|
||||
def test_make_app_name_cpp_special_chars_escaped() -> None:
|
||||
"""Test special characters are properly escaped in C++ string."""
|
||||
cpp_expr, _, _ = make_app_name_cpp('my "device"', "buf", "-", add_mac_suffix=False)
|
||||
# cpp_string_escape uses octal escapes for quotes
|
||||
assert '"' not in cpp_expr[1:-1] # no unescaped quotes inside the outer quotes
|
||||
|
||||
@@ -23,6 +23,7 @@ from esphome.core.entity_helpers import (
|
||||
_setup_entity_impl,
|
||||
entity_duplicate_validator,
|
||||
get_base_entity_object_id,
|
||||
register_icon,
|
||||
setup_entity,
|
||||
)
|
||||
from esphome.cpp_generator import MockObj
|
||||
@@ -904,6 +905,22 @@ def test_register_string_overflow() -> None:
|
||||
_register_string("overflow", category, 3, "test")
|
||||
|
||||
|
||||
def test_register_icon_max_length() -> None:
|
||||
"""Test register_icon rejects icons exceeding 63 characters."""
|
||||
# 63 chars should succeed
|
||||
max_icon = "mdi:" + "a" * 59 # 63 total
|
||||
idx = register_icon(max_icon)
|
||||
assert idx > 0
|
||||
|
||||
# 64 chars should fail
|
||||
too_long = "mdi:" + "a" * 60 # 64 total
|
||||
with pytest.raises(ValueError, match="Icon string too long"):
|
||||
register_icon(too_long)
|
||||
|
||||
# Empty string returns 0
|
||||
assert register_icon("") == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_setup_entity_with_entity_category(
|
||||
setup_test_environment: list[str],
|
||||
|
||||
@@ -148,6 +148,18 @@ def test_icon__invalid():
|
||||
config_validation.icon("foo")
|
||||
|
||||
|
||||
def test_icon__max_length():
|
||||
"""Test that icons exceeding 63 characters are rejected."""
|
||||
# Exactly 63 chars should pass
|
||||
max_icon = "mdi:" + "a" * 59 # 63 chars total
|
||||
assert config_validation.icon(max_icon) == max_icon
|
||||
|
||||
# 64 chars should fail
|
||||
too_long = "mdi:" + "a" * 60 # 64 chars total
|
||||
with pytest.raises(Invalid, match="Icon string is too long"):
|
||||
config_validation.icon(too_long)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("value", ("True", "YES", "on", "enAblE", True))
|
||||
def test_boolean__valid_true(value):
|
||||
assert config_validation.boolean(value) is True
|
||||
|
||||
Reference in New Issue
Block a user