Compare commits

...
Author SHA1 Message Date
J. Nick Koston b984f63c49 Merge remote-tracking branch 'origin/dev' into api-reverse-connection
# Conflicts:
#	tests/integration/test_light_initial_state.py
2026-09-08 18:13:24 +02:00
J. Nick Koston 78f055299c Merge remote-tracking branch 'origin/dev' into api-reverse-connection
# Conflicts:
#	tests/integration/test_api_zero_psk_provisioning.py
2026-09-08 18:11:39 +02:00
J. Nick Koston 1ce0bed3f6 [core] Share compiled binaries across modbus integration tests (#18945) 2026-09-08 18:10:45 +02:00
J. Nick Koston 7f9812cef6 Terminate the loaded target before logging it, validate udp addresses on the raw lwip branch too, and resolve the socket default in the lwip_tcp test 2026-09-05 13:07:37 +02:00
J. Nick Koston a684e7e67d Decide the listener failure policy in setup, enforce the slot limit once, and reject lwip_tcp through the resolved socket implementation only 2026-09-05 11:23:59 +02:00
J. Nick Koston afc93a4dc8 Warn when any UDP address is dropped and print the boot delay with PRIu32 2026-09-04 21:21:50 +02:00
J. Nick Koston 0a74cb4e89 Apply the dial delay the same way at boot and after a disconnect, tell socket create and setblocking failures apart, and cap the delay so the wait always elapses 2026-09-04 12:42:07 +02:00
J. Nick Koston 9c4a08aaad Capture errno before it can be clobbered, warn when no UDP address is valid, and give the reboot test timing headroom 2026-09-04 07:25:50 +02:00
J. Nick Koston 68d5968405 Skip the dial delay when no inbound path waits, keep dial-out alive when the listener fails, check the socket results this PR made meaningful, and cover the clean reboot branch 2026-09-04 06:14:00 +02:00
J. Nick Koston 30da0e81fa Adopt the shared connect poll in async_tcp and name the refused implementation 2026-09-03 18:54:23 +02:00
J. Nick Koston 900638463a Move the connect poll into the socket component and dedupe target persistence 2026-09-03 17:41:48 +02:00
J. Nick Koston 1665d509d4 Bound the delay against uint32 wrap and print the effective boot delay 2026-09-03 12:16:42 +02:00
J. Nick Koston 29d8f9c7c6 Name the reboot cause and log the loaded target at setup 2026-09-03 11:28:32 +02:00
J. Nick Koston 18b5277254 Accept a bare outgoing_connection block and log the bound enforcement 2026-09-02 12:26:19 +02:00
J. Nick Koston b110cee973 Keep the platform level error message alongside the socket implementation check 2026-09-02 12:18:32 +02:00
J. Nick Koston 1171fbf892 Extract a reusable connect poll helper and enforce bounds at the write 2026-09-02 11:58:51 +02:00
J. Nick Koston ae549b25bb Report address parse failures from set_sockaddr and drop duplicated policy 2026-09-02 11:36:57 +02:00
J. Nick Koston 35ab455c10 Use the portable INADDR_NONE, self heal a bad remembered target, retry failed saves 2026-09-02 11:18:37 +02:00
J. Nick Koston dae9a5d1a6 Harden dial target validation and keep a fresh address on flash failure 2026-08-31 21:31:06 -04:00
J. Nick Koston bf380037a6 Refresh the watchdog on every authenticated removal 2026-08-31 21:01:03 -04:00
J. Nick Koston 493c7265ab Fix the IPv6 target guard so it actually compiles in 2026-08-31 20:44:29 -04:00
J. Nick Koston 238b6bb0b7 Simplify dial bookkeeping and settle the watchdog policy 2026-08-31 20:20:50 -04:00
J. Nick Koston 0211bdf2c0 Address review round: backoff fairness, watchdog integrity, silent failures 2026-08-31 19:26:01 -04:00
J. Nick Koston a95bfe0bcb Tighten comments and member ordering 2026-08-31 18:42:28 -04:00
J. Nick Koston 3a0bbc4c17 Merge remote-tracking branch 'origin/dev' into api-reverse-connection 2026-08-31 18:40:27 -04:00
J. Nick Koston cc9dc95cab Only dial immediately at boot for deep sleep devices 2026-08-31 18:24:40 -04:00
J. Nick Koston 79325fe59a Dial immediately when no client has ever connected and polish review nits 2026-08-31 17:36:57 -04:00
pre-commit-ci-lite[bot] 5fbe09b68c apply automatic formatting fixes 2026-08-31 21:28:10 +00:00
J. Nick Koston f41ff5aaeb Merge remote-tracking branch 'origin/dev' into api-reverse-connection 2026-08-31 17:26:02 -04:00
J. Nick Koston 7b57fdd35c Gate the dial-back flag on this connection's transport and tune redial policy 2026-08-31 17:19:17 -04:00
J. Nick Koston fc671d38e5 Address review findings and advertise support in device info 2026-08-31 16:44:49 -04:00
J. Nick Koston 4e0cda0287 Select the dial-back target with an explicit hello flag instead of guessing from client info 2026-08-31 15:40:15 -04:00
J. Nick Koston 676eac7686 [api] Add outgoing connections so the device can dial Home Assistant 2026-08-31 14:06:05 -05:00
58 changed files with 2112 additions and 1168 deletions
+96 -2
View File
@@ -5,7 +5,7 @@ from typing import Any
from esphome import automation
from esphome.automation import Condition
import esphome.codegen as cg
from esphome.components.const import CONF_DESCRIPTION
from esphome.components.const import CONF_DESCRIPTION, CONF_HOST
from esphome.components.logger import request_log_listener
# ENCRYPTION_SCHEMA and validate_encryption_key are re-exported for external
@@ -25,6 +25,8 @@ from esphome.const import (
CONF_CAPTURE_RESPONSE,
CONF_DATA,
CONF_DATA_TEMPLATE,
CONF_DELAY,
CONF_ENABLE_IPV6,
CONF_ENCRYPTION,
CONF_EVENT,
CONF_ID,
@@ -48,6 +50,7 @@ from esphome.const import (
)
from esphome.core import CORE, ID, CoroPriority, EsphomeError, coroutine_with_priority
from esphome.cpp_generator import MockObj, TemplateArgsType
import esphome.final_validate as fv
from esphome.helpers import fnv1_hash
from esphome.types import ConfigFragmentType, ConfigType
@@ -134,6 +137,7 @@ CONF_HOMEASSISTANT_SERVICES = "homeassistant_services"
CONF_HOMEASSISTANT_STATES = "homeassistant_states"
CONF_LISTEN_BACKLOG = "listen_backlog"
CONF_MAX_SEND_QUEUE = "max_send_queue"
CONF_OUTGOING_CONNECTION = "outgoing_connection"
CONF_STATE_SUBSCRIPTION_ONLY = "state_subscription_only"
@@ -285,9 +289,44 @@ def _consume_api_sockets(config: ConfigType) -> ConfigType:
# (not max_connections, which is the upper limit rarely reached)
socket.consume_sockets(3, "api")(config)
socket.consume_sockets(1, "api", socket.SocketType.TCP_LISTEN)(config)
if CONF_OUTGOING_CONNECTION in config:
socket.consume_sockets(1, "api_outgoing_connection")(config)
return config
def _validate_outgoing_connection(config: ConfigType) -> ConfigType:
if CONF_OUTGOING_CONNECTION not in config:
return config
if CONF_ENCRYPTION not in config:
raise cv.Invalid(
"outgoing_connection requires 'encryption' so the peer is verified by key",
path=[CONF_OUTGOING_CONNECTION],
)
return config
_OUTGOING_CONNECTION_SCHEMA = cv.Schema(
{
cv.Optional(CONF_HOST): cv.ipaddress,
cv.Optional(CONF_PORT, default=6054): cv.port,
# Bounded to half the device's uint32 millisecond range so the wait
# always elapses under a wrapping clock
cv.Optional(CONF_DELAY, default="60s"): cv.All(
cv.positive_time_period_milliseconds,
cv.Range(max=cv.TimePeriod(milliseconds=2147483647)),
),
}
)
def _outgoing_connection_schema(config: ConfigType | None) -> ConfigType:
# A bare `outgoing_connection:` block is valid; without a host the device
# dials the remembered last dial-back client
if config is None:
config = {}
return _OUTGOING_CONNECTION_SCHEMA(config)
CONFIG_SCHEMA = cv.All(
cv.Schema(
{
@@ -312,6 +351,7 @@ CONFIG_SCHEMA = cv.All(
): ACTIONS_SCHEMA,
cv.Exclusive(CONF_ACTIONS, group_of_exclusion=CONF_ACTIONS): ACTIONS_SCHEMA,
cv.Optional(CONF_ENCRYPTION): encryption_schema,
cv.Optional(CONF_OUTGOING_CONNECTION): _outgoing_connection_schema,
cv.Optional(CONF_BATCH_DELAY, default="100ms"): cv.All(
cv.positive_time_period_milliseconds,
cv.Range(max=cv.TimePeriod(milliseconds=65535)),
@@ -368,6 +408,7 @@ CONFIG_SCHEMA = cv.All(
}
).extend(cv.COMPONENT_SCHEMA),
cv.rename_key(CONF_SERVICES, CONF_ACTIONS),
_validate_outgoing_connection,
_consume_api_sockets,
_register_provisioning_source,
)
@@ -424,7 +465,52 @@ def _validate_esp8266_action_strings(config: ConfigType) -> ConfigType:
return config
FINAL_VALIDATE_SCHEMA = _validate_esp8266_action_strings
def _validate_outgoing_socket_implementation(config: ConfigType) -> ConfigType:
"""Reject the raw lwip_tcp socket, the only option on ESP8266 and RP2040.
Checked against the resolved implementation so an explicit selection on
another platform is caught the same way as the platform default.
"""
if CONF_OUTGOING_CONNECTION not in config:
return config
from esphome.components import socket
socket_conf = fv.full_config.get().get("socket") or {}
if (
impl := socket_conf.get(socket.CONF_IMPLEMENTATION)
) in socket.IMPLEMENTATIONS_WITHOUT_CONNECT:
raise cv.Invalid(
f"outgoing_connection is not supported with the {impl} socket "
"implementation (the only one on ESP8266 and RP2040) because it "
"cannot make outgoing connections",
path=[CONF_OUTGOING_CONNECTION],
)
return config
def _validate_outgoing_host_ipv6(config: ConfigType) -> ConfigType:
"""An IPv6 host can never be parsed, so never dialed, without IPv6."""
if (
(outgoing := config.get(CONF_OUTGOING_CONNECTION)) is None
or (host := outgoing.get(CONF_HOST)) is None
or host.version != 6
):
return config
network_conf = fv.full_config.get().get("network") or {}
if not network_conf.get(CONF_ENABLE_IPV6):
raise cv.Invalid(
"outgoing_connection host is an IPv6 address but IPv6 is not "
"enabled; set 'network: enable_ipv6: true'",
path=[CONF_OUTGOING_CONNECTION, CONF_HOST],
)
return config
FINAL_VALIDATE_SCHEMA = cv.All(
_validate_esp8266_action_strings,
_validate_outgoing_socket_implementation,
_validate_outgoing_host_ipv6,
)
def _add_action_strings(
@@ -606,6 +692,13 @@ async def to_code(config: ConfigType) -> None:
else:
cg.add_define("USE_API_PLAINTEXT")
if (outgoing := config.get(CONF_OUTGOING_CONNECTION)) is not None:
cg.add_define("USE_API_OUTGOING_CONNECTION")
if (host := outgoing.get(CONF_HOST)) is not None:
cg.add_define("API_OUTGOING_CONNECTION_HOST", str(host))
cg.add_define("API_OUTGOING_CONNECTION_PORT", outgoing[CONF_PORT])
cg.add_define("API_OUTGOING_CONNECTION_DELAY", outgoing[CONF_DELAY])
cg.add_define("USE_API")
cg.add_global(api_ns.using)
@@ -992,6 +1085,7 @@ _define_filter = filter_source_files_from_defines(
"user_services.cpp": "USE_API_USER_DEFINED_ACTIONS",
"api_frame_helper_noise.cpp": "USE_API_NOISE",
"api_frame_helper_plaintext.cpp": "USE_API_PLAINTEXT",
"api_outgoing_connection.cpp": "USE_API_OUTGOING_CONNECTION",
}
)
+9
View File
@@ -112,6 +112,11 @@ message HelloRequest {
string client_info = 1;
uint32 api_version_major = 2;
uint32 api_version_minor = 3;
// Set by clients that can accept connections the device opens to them
// (see api: outgoing_connection:). The device remembers this client's
// address as the target to dial when no such client is connected.
bool outgoing_connection_target = 4 [(field_ifdef) = "USE_API_OUTGOING_CONNECTION"];
}
// Confirmation of successful connection request.
@@ -331,6 +336,10 @@ message DeviceInfoResponse {
// all-zeros PSK, so the api encryption key can be provisioned without being
// sent in plaintext (protects against passive sniffing, not active MITM)
bool api_encryption_provisionable = 26 [(field_ifdef) = "USE_API_NOISE"];
// Device is built with the api outgoing_connection option and can open
// the TCP connection to a dial-back target itself
bool api_outgoing_connection_supported = 27 [(field_ifdef) = "USE_API_OUTGOING_CONNECTION"];
}
// ==================== DEVICE CAPABILITIES ====================
+16
View File
@@ -1822,6 +1822,19 @@ bool APIConnection::send_hello_response_(const HelloRequest &msg) {
// Auto-authenticate - password auth was removed in ESPHome 2026.1.0
this->complete_authentication_();
#ifdef USE_API_OUTGOING_CONNECTION
// With a PSK set only key-verified transports reach hello: plaintext and
// zero-PSK are rejected, and pre-activation sessions are force-closed
if (msg.outgoing_connection_target && !this->flags_.outgoing_connection_target) {
if (this->parent_->get_noise_ctx().has_psk()) {
this->flags_.outgoing_connection_target = true;
this->parent_->on_outgoing_target_client(this);
} else {
this->log_client_(ESPHOME_LOG_LEVEL_WARN, LOG_STR("Dial-back target refused; no key active"));
}
}
#endif
return this->send_message(resp);
}
@@ -1944,6 +1957,9 @@ bool APIConnection::send_device_info_response_() {
// one) so this advertisement survives the plaintext removal in 2027.2.0.
resp.api_encryption_provisionable = !this->parent_->get_noise_ctx().has_psk();
#endif
#ifdef USE_API_OUTGOING_CONNECTION
resp.api_outgoing_connection_supported = true;
#endif
#endif
#ifdef USE_DEVICES
size_t device_index = 0;
+18
View File
@@ -375,6 +375,21 @@ class APIConnection final : public APIServerConnectionBase {
return this->helper_->get_peername_to(buf);
}
#ifdef USE_API_OUTGOING_CONNECTION
/// Outgoing connection: send our server hello immediately so the peer can
/// pick the matching key. Outgoing connections are only dialed when a PSK
/// is set, so the helper is always the noise helper. Call after start().
void mark_outgoing() {
if (this->flags_.remove) {
return; // start() failed; the connection is already being torn down
}
APIError err = static_cast<APINoiseFrameHelper *>(this->helper_.get())->send_server_hello_first();
if (err != APIError::OK) {
this->fatal_error_with_log_(LOG_STR("Server hello failed"), err);
}
}
#endif
protected:
bool try_to_clear_buffer_slow_(bool log_out_of_space);
@@ -745,6 +760,9 @@ class APIConnection final : public APIServerConnectionBase {
uint8_t batch_first_message : 1; // For batch buffer allocation
uint8_t should_try_send_immediately : 1; // True after initial states are sent
uint8_t may_have_remaining_data : 1; // Read loop hit limit, retry without ready check
#ifdef USE_API_OUTGOING_CONNECTION
uint8_t outgoing_connection_target : 1; // Client declared itself a dial-back target in its hello
#endif
#ifdef HAS_PROTO_MESSAGE_DUMP
uint8_t log_only_mode : 1;
#endif
+2 -1
View File
@@ -282,7 +282,8 @@ class APIFrameHelper {
DATA = 5,
CLOSED = 6,
FAILED = 7,
EXPLICIT_REJECT = 8, // Noise only
EXPLICIT_REJECT = 8, // Noise only
CLIENT_HELLO_OUTGOING = 9, // Noise only: like CLIENT_HELLO but the server hello already went out (outgoing conn)
};
// Fast inline state check for read_packet/write_protobuf_messages hot path.
@@ -81,6 +81,13 @@ APIError APINoiseFrameHelper::init() {
state_ = State::CLIENT_HELLO;
return APIError::OK;
}
#ifdef USE_API_OUTGOING_CONNECTION
APIError APINoiseFrameHelper::send_server_hello_first() {
// The peer needs our name and MAC to pick the key before its first message
this->state_ = State::CLIENT_HELLO_OUTGOING;
return this->send_server_hello_frame_();
}
#endif
#ifdef USE_API_PLAINTEXT
APIError APINoiseFrameHelper::init_from_handoff(const uint8_t *header, uint8_t header_len) {
APIError err = this->init();
@@ -253,6 +260,9 @@ APIError APINoiseFrameHelper::state_action_() {
HELPER_LOG("Bad state for method: %d", (int) this->state_);
return APIError::BAD_STATE;
case State::CLIENT_HELLO:
#ifdef USE_API_OUTGOING_CONNECTION
case State::CLIENT_HELLO_OUTGOING:
#endif
return this->state_action_client_hello_();
case State::SERVER_HELLO:
return this->state_action_server_hello_();
@@ -285,11 +295,16 @@ APIError APINoiseFrameHelper::state_action_client_hello_() {
std::memcpy(this->prologue_.data() + old_size + 2, this->rx_buf_.data(), rx_size);
}
#ifdef USE_API_OUTGOING_CONNECTION
if (this->state_ == State::CLIENT_HELLO_OUTGOING) {
// Server hello already went out at handoff
return this->start_handshake_();
}
#endif
state_ = State::SERVER_HELLO;
return APIError::OK;
}
APIError APINoiseFrameHelper::state_action_server_hello_() {
// send server hello
APIError APINoiseFrameHelper::send_server_hello_frame_() {
const auto &name = App.get_name();
char mac[MAC_ADDRESS_BUFFER_SIZE];
get_mac_address_into_buffer(mac);
@@ -313,15 +328,18 @@ APIError APINoiseFrameHelper::state_action_server_hello_() {
// node mac, terminated by null byte
std::memcpy(msg + mac_offset, mac, MAC_ADDRESS_BUFFER_SIZE);
APIError aerr = write_frame_(msg, total_size);
return write_frame_(msg, total_size);
}
APIError APINoiseFrameHelper::state_action_server_hello_() {
APIError aerr = this->send_server_hello_frame_();
if (aerr != APIError::OK)
return aerr;
// start handshake
aerr = init_handshake_();
return this->start_handshake_();
}
APIError APINoiseFrameHelper::start_handshake_() {
APIError aerr = init_handshake_();
if (aerr != APIError::OK)
return aerr;
state_ = State::HANDSHAKE;
return APIError::OK;
}
@@ -28,6 +28,12 @@ class APINoiseFrameHelper final : public APIFrameHelper {
// Seeds the already-read header bytes and pumps the handshake state machine
// until it would block.
APIError init_from_handoff(const uint8_t *header, uint8_t header_len);
#endif
#ifdef USE_API_OUTGOING_CONNECTION
// Send the server hello immediately so the peer can pick the key before
// its PSK-mixed message. Call after init(); the mode is tracked in state_
// so the helper does not grow.
APIError send_server_hello_first();
#endif
APIError loop() override;
APIError read_packet(ReadPacketBuffer *buffer) override;
@@ -39,6 +45,8 @@ class APINoiseFrameHelper final : public APIFrameHelper {
APIError state_action_();
APIError state_action_client_hello_();
APIError state_action_server_hello_();
APIError send_server_hello_frame_();
APIError start_handshake_();
APIError state_action_handshake_();
APIError state_action_handshake_read_();
APIError state_action_handshake_write_();
@@ -0,0 +1,236 @@
#include "api_outgoing_connection.h"
#if defined(USE_API) && defined(USE_API_OUTGOING_CONNECTION)
#include "api_connection.h"
#include "api_server.h"
#include "esphome/components/network/util.h"
#include "esphome/core/application.h"
#include "esphome/core/helpers.h"
#include "esphome/core/log.h"
#include <cerrno>
#include <cinttypes>
#include <cstring>
namespace esphome::api {
static const char *const TAG = "api.outgoing";
void OutgoingConnectionManager::setup() {
#ifndef API_OUTGOING_CONNECTION_HOST
this->target_pref_ = global_preferences->make_preference<SavedOutgoingTarget>(629847102UL, true);
if (this->target_pref_.load(&this->saved_)) {
// Defend against a corrupt or truncated blob before the first read
this->saved_.host[sizeof(this->saved_.host) - 1] = '\0';
this->host_persisted_ = true;
ESP_LOGD(TAG, "Loaded target %s", this->saved_.host);
} else {
// Never saved, or the blob failed its size/CRC check
ESP_LOGD(TAG, "No saved target");
this->saved_ = {};
}
#endif
}
void OutgoingConnectionManager::loop(APIServer *server) {
if (server->has_outgoing_target_client_()) {
return; // on_target_client() already reset the dial state
}
if (this->dialed_conn_ != nullptr) {
// A live dialed session (flagged or not, e.g. a host: peer) is the
// target; a silent one dies on the handshake timeout
return;
}
const uint32_t now = App.get_loop_component_start_time();
switch (this->state_) {
case DialState::DIAL_STATE_IDLE:
#ifdef USE_DEEP_SLEEP
// A deep sleep wake window is too short to spend on the delay
this->schedule_wait_(now, BACKOFF_MIN_MS);
#else
// Target went away; give it the configured delay to reconnect first
this->schedule_wait_(now, API_OUTGOING_CONNECTION_DELAY);
#endif
break;
case DialState::DIAL_STATE_WAITING:
if (now - this->state_ts_ >= this->wait_) {
this->try_dial_(server, now);
}
break;
case DialState::DIAL_STATE_CONNECTING:
this->poll_connect_(server, now);
break;
}
}
void OutgoingConnectionManager::try_dial_(APIServer *server, uint32_t now) {
if (!network::is_connected()) {
// Flips within seconds of boot; recheck fast so a deep sleep wake
// window is not spent waiting
this->schedule_wait_(now, NETWORK_RETRY_MS);
return;
}
const char *host = this->target_host_();
if (host == nullptr) {
// The steady state until a dial-back client has ever connected
ESP_LOGV(TAG, "Not dialing: no target");
this->schedule_wait_(now, PRECONDITION_RETRY_MS);
return;
}
const bool at_limit = server->at_client_limit_();
if (at_limit || !server->noise_ctx_.has_psk()) {
ESP_LOGD(TAG, "Not dialing: %s", at_limit ? "max connections" : "no key");
// Not a dial failure; retry without escalating the backoff
this->schedule_wait_(now, PRECONDITION_RETRY_MS);
return;
}
struct sockaddr_storage addr;
socklen_t addr_len =
socket::set_sockaddr((struct sockaddr *) &addr, sizeof(addr), host, API_OUTGOING_CONNECTION_PORT);
if (addr_len == 0) {
ESP_LOGW(TAG, "Invalid target %s", host);
#ifndef API_OUTGOING_CONNECTION_HOST
// A corrupt remembered value can never become dialable; forget it
// (covers an IPv6 literal left by an earlier enable_ipv6 build too)
this->saved_ = {};
if (!this->persist_target_()) {
ESP_LOGW(TAG, "Failed to clear target");
}
#endif
this->schedule_retry_(now);
return;
}
this->dial_socket_ = socket::socket_loop_monitored(((struct sockaddr *) &addr)->sa_family, SOCK_STREAM, IPPROTO_TCP);
if (!this->dial_socket_ || this->dial_socket_->setblocking(false) != 0) {
ESP_LOGW(TAG, "Socket %s failed: errno %d", this->dial_socket_ ? "setblocking" : "create", errno);
this->schedule_retry_(now);
return;
}
ESP_LOGD(TAG, "Dialing %s:%u", host, API_OUTGOING_CONNECTION_PORT);
int err = this->dial_socket_->connect((struct sockaddr *) &addr, addr_len);
if (err == 0) {
// Immediate success (possible for localhost)
this->handoff_(server, now);
return;
}
if (errno != EINPROGRESS) {
ESP_LOGW(TAG, "Connect failed: errno %d", errno);
this->schedule_retry_(now);
return;
}
this->state_ = DialState::DIAL_STATE_CONNECTING;
this->state_ts_ = now;
this->last_poll_ = now;
}
void OutgoingConnectionManager::poll_connect_(APIServer *server, uint32_t now) {
if (now - this->state_ts_ >= CONNECT_TIMEOUT_MS) {
ESP_LOGW(TAG, "Connect timeout");
this->schedule_retry_(now);
return;
}
if (now - this->last_poll_ < CONNECT_POLL_INTERVAL_MS) {
return;
}
this->last_poll_ = now;
int err = 0;
switch (socket::poll_connect(*this->dial_socket_, err)) {
case socket::ConnectPollResult::CONNECT_POLL_PENDING:
break;
case socket::ConnectPollResult::CONNECT_POLL_CONNECTED:
this->handoff_(server, now);
break;
case socket::ConnectPollResult::CONNECT_POLL_ERROR:
ESP_LOGW(TAG, "Connect failed: %d", err);
this->schedule_retry_(now);
break;
}
}
void OutgoingConnectionManager::handoff_(APIServer *server, uint32_t now) {
this->dialed_conn_ = server->add_outgoing_client_(std::move(this->dial_socket_));
if (this->dialed_conn_ == nullptr) {
// Only preconditions (slot limit, key cleared) refuse the handoff; the
// peer is reachable, so do not escalate the backoff
this->schedule_wait_(now, PRECONDITION_RETRY_MS);
return;
}
// Connected; dialed_conn_ gates further dialing until the session settles
this->state_ = DialState::DIAL_STATE_IDLE;
}
void OutgoingConnectionManager::schedule_wait_(uint32_t now, uint32_t wait) {
this->dial_socket_.reset(); // no-op when the socket was handed off
this->state_ = DialState::DIAL_STATE_WAITING;
this->state_ts_ = now;
this->wait_ = wait;
}
void OutgoingConnectionManager::schedule_retry_(uint32_t now) {
// +/-20% jitter so a fleet of devices does not retry one server in lockstep
const uint32_t jitter_span = this->backoff_ / 5;
this->schedule_wait_(now, this->backoff_ - jitter_span + (random_uint32() % (2 * jitter_span + 1)));
this->backoff_ = std::min(this->backoff_ * 2, BACKOFF_MAX_MS);
}
void OutgoingConnectionManager::on_client_removed(APIConnection *conn, bool was_authenticated) {
if (conn != this->dialed_conn_) {
return;
}
this->dialed_conn_ = nullptr;
if (was_authenticated) {
// A working peer (e.g. a host: target that never sends the flag)
// disconnected normally; state is IDLE, so loop() applies the delay
this->backoff_ = BACKOFF_MIN_MS;
} else {
this->schedule_retry_(App.get_loop_component_start_time());
}
}
void OutgoingConnectionManager::on_target_client(APIConnection *conn) {
// The target is connected; stop any dial in flight and reset the backoff.
// A dialed connection stays tracked unless it is this one: an inbound
// target must not orphan a still-open dial.
this->dial_socket_.reset();
if (conn == this->dialed_conn_) {
this->dialed_conn_ = nullptr;
}
this->state_ = DialState::DIAL_STATE_IDLE;
this->backoff_ = BACKOFF_MIN_MS;
#ifndef API_OUTGOING_CONNECTION_HOST
SavedOutgoingTarget target{};
conn->get_peername_to(target.host);
if (target.host[0] == '\0') {
ESP_LOGW(TAG, "Could not read peer address; not remembering target");
return;
}
if (this->host_persisted_ && strcmp(target.host, this->saved_.host) == 0) {
return; // unchanged and already on flash; avoid flash wear
}
// Use the fresh address this boot even if the flash write fails; a failed
// write is retried on the next flagged hello via host_persisted_
this->saved_ = target;
if (!this->persist_target_()) {
ESP_LOGW(TAG, "Failed to save target");
return;
}
ESP_LOGD(TAG, "Saved %s as outgoing connection target", this->saved_.host);
#endif
}
void OutgoingConnectionManager::dump_config() const {
const char *host = this->target_host_();
if (host == nullptr) {
host = "none remembered yet";
}
// The boot delay differs from delay: on deep sleep builds, so print the
// value that actually applies
ESP_LOGCONFIG(TAG,
" Outgoing connection port: %u\n"
" Outgoing connection host: %s\n"
" Outgoing connection boot delay: %" PRIu32 "ms",
API_OUTGOING_CONNECTION_PORT, host, BOOT_WAIT_MS);
}
} // namespace esphome::api
#endif // USE_API && USE_API_OUTGOING_CONNECTION
@@ -0,0 +1,117 @@
#pragma once
#include "esphome/core/defines.h"
#if defined(USE_API) && defined(USE_API_OUTGOING_CONNECTION)
#ifdef USE_SOCKET_IMPL_LWIP_TCP
#error "api outgoing_connection needs a socket implementation that can make outgoing connections"
#endif
#ifndef USE_API_NOISE
#error "api outgoing_connection needs noise encryption so the peer is verified by key"
#endif
#include "esphome/components/socket/socket.h"
#include "esphome/core/preferences.h"
#include <memory>
namespace esphome::api {
class APIServer;
class APIConnection;
// Follows the build's address family (ifdef'd in socket/headers.h): toggling
// enable_ipv6 changes the blob size, load() rejects the old blob, and the
// target is simply relearned
static constexpr size_t SAVED_TARGET_HOST_LEN = socket::SOCKADDR_STR_LEN;
struct SavedOutgoingTarget {
// IP as text so the socket component's v4-mapped-IPv6 normalization is
// reused on both ends; empty = none remembered
char host[SAVED_TARGET_HOST_LEN];
} PACKED; // NOLINT
/// Dials out when no dial-back target client is connected. Only the TCP
/// direction flips: the device stays the Noise responder, so both sides
/// still verify by key. Targets the YAML host or the last remembered client.
class OutgoingConnectionManager {
public:
void setup();
void loop(APIServer *server);
/// A key-verified client declared itself a dial-back target; last one wins
void on_target_client(APIConnection *conn);
/// Clears the dialed-connection gate; dying unauthenticated escalates the backoff
void on_client_removed(APIConnection *conn, bool was_authenticated);
void on_shutdown() { this->dial_socket_.reset(); }
void dump_config() const;
protected:
enum class DialState : uint8_t {
DIAL_STATE_IDLE,
DIAL_STATE_WAITING,
DIAL_STATE_CONNECTING,
};
static constexpr uint32_t BACKOFF_MIN_MS = 5000;
static constexpr uint32_t BACKOFF_MAX_MS = 300000;
static constexpr uint32_t CONNECT_TIMEOUT_MS = 10000;
static constexpr uint32_t CONNECT_POLL_INTERVAL_MS = 250;
static constexpr uint32_t NETWORK_RETRY_MS = 500;
static constexpr uint32_t PRECONDITION_RETRY_MS = 5000;
// Boot waits for the client to connect in first; a deep sleep wake window
// is short, so connecting out immediately is the wake state
#ifdef USE_DEEP_SLEEP
static constexpr uint32_t BOOT_WAIT_MS = 0;
#else
static constexpr uint32_t BOOT_WAIT_MS = API_OUTGOING_CONNECTION_DELAY;
#endif
void try_dial_(APIServer *server, uint32_t now);
void poll_connect_(APIServer *server, uint32_t now);
// Hand the connected socket to the server and gate on the new connection
void handoff_(APIServer *server, uint32_t now);
// Close any half-open dial and wait a jittered backoff before retrying
void schedule_retry_(uint32_t now);
// Wait without escalating the backoff (used for unmet preconditions)
void schedule_wait_(uint32_t now, uint32_t wait);
#ifndef API_OUTGOING_CONNECTION_HOST
// Write saved_ to flash, tracking success in host_persisted_
bool persist_target_() {
this->host_persisted_ = this->target_pref_.save(&this->saved_) && global_preferences->sync();
return this->host_persisted_;
}
#endif
const char *target_host_() const {
#ifdef API_OUTGOING_CONNECTION_HOST
return API_OUTGOING_CONNECTION_HOST;
#else
return this->saved_.host[0] != '\0' ? this->saved_.host : nullptr;
#endif
}
// Pointers first (4 bytes each on 32-bit)
std::unique_ptr<socket::Socket> dial_socket_;
// Compared only, never dereferenced
APIConnection *dialed_conn_{nullptr};
#ifndef API_OUTGOING_CONNECTION_HOST
ESPPreferenceObject target_pref_;
#endif
// 4-byte types
uint32_t backoff_{BACKOFF_MIN_MS};
uint32_t wait_{BOOT_WAIT_MS};
uint32_t state_ts_{0};
uint32_t last_poll_{0};
// Byte-aligned types last
#ifndef API_OUTGOING_CONNECTION_HOST
SavedOutgoingTarget saved_{};
// False while saved_ holds a value the flash write failed for; retried on
// the next flagged hello
bool host_persisted_{false};
#endif
DialState state_{DialState::DIAL_STATE_WAITING};
};
} // namespace esphome::api
#endif // USE_API && USE_API_OUTGOING_CONNECTION
+11
View File
@@ -15,6 +15,11 @@ bool HelloRequest::decode_varint(uint32_t field_id, proto_varint_value_t value)
case 3:
this->api_version_minor = value;
break;
#ifdef USE_API_OUTGOING_CONNECTION
case 4:
this->outgoing_connection_target = value != 0;
break;
#endif
default:
return false;
}
@@ -175,6 +180,9 @@ uint8_t *DeviceInfoResponse::encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_
#endif
#ifdef USE_API_NOISE
ProtoEncode::encode_bool(pos PROTO_ENCODE_DEBUG_ARG, 26, this->api_encryption_provisionable);
#endif
#ifdef USE_API_OUTGOING_CONNECTION
ProtoEncode::encode_bool(pos PROTO_ENCODE_DEBUG_ARG, 27, this->api_outgoing_connection_supported);
#endif
return pos;
}
@@ -240,6 +248,9 @@ uint32_t DeviceInfoResponse::calculate_size() const {
#endif
#ifdef USE_API_NOISE
size += ProtoSize::calc_bool(2, this->api_encryption_provisionable);
#endif
#ifdef USE_API_OUTGOING_CONNECTION
size += ProtoSize::calc_bool(2, this->api_outgoing_connection_supported);
#endif
return size;
}
+8 -2
View File
@@ -412,13 +412,16 @@ class CommandProtoMessage : public ProtoDecodableMessage {
class HelloRequest final : public ProtoDecodableMessage {
public:
static constexpr uint16_t MESSAGE_TYPE = 1;
static constexpr uint8_t ESTIMATED_SIZE = 17;
static constexpr uint8_t ESTIMATED_SIZE = 19;
#ifdef HAS_PROTO_MESSAGE_DUMP
const LogString *message_name() const override { return LOG_STR("hello_request"); }
#endif
StringRef client_info{};
uint32_t api_version_major{0};
uint32_t api_version_minor{0};
#ifdef USE_API_OUTGOING_CONNECTION
bool outgoing_connection_target{false};
#endif
#ifdef HAS_PROTO_MESSAGE_DUMP
const char *dump_to(DumpBuffer &out) const override;
#endif
@@ -549,7 +552,7 @@ class SerialProxyInfo final : public ProtoMessage {
class DeviceInfoResponse final : public ProtoMessage {
public:
static constexpr uint16_t MESSAGE_TYPE = 10;
static constexpr uint16_t ESTIMATED_SIZE = 312;
static constexpr uint16_t ESTIMATED_SIZE = 315;
#ifdef HAS_PROTO_MESSAGE_DUMP
const LogString *message_name() const override { return LOG_STR("device_info_response"); }
#endif
@@ -607,6 +610,9 @@ class DeviceInfoResponse final : public ProtoMessage {
#endif
#ifdef USE_API_NOISE
bool api_encryption_provisionable{false};
#endif
#ifdef USE_API_OUTGOING_CONNECTION
bool api_outgoing_connection_supported{false};
#endif
uint8_t *encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const;
uint32_t calculate_size() const;
+6
View File
@@ -885,6 +885,9 @@ const char *HelloRequest::dump_to(DumpBuffer &out) const {
dump_field(out, ESPHOME_PSTR("client_info"), this->client_info);
dump_field(out, ESPHOME_PSTR("api_version_major"), this->api_version_major);
dump_field(out, ESPHOME_PSTR("api_version_minor"), this->api_version_minor);
#ifdef USE_API_OUTGOING_CONNECTION
dump_field(out, ESPHOME_PSTR("outgoing_connection_target"), this->outgoing_connection_target);
#endif
return out.c_str();
}
const char *HelloResponse::dump_to(DumpBuffer &out) const {
@@ -1008,6 +1011,9 @@ const char *DeviceInfoResponse::dump_to(DumpBuffer &out) const {
#endif
#ifdef USE_API_NOISE
dump_field(out, ESPHOME_PSTR("api_encryption_provisionable"), this->api_encryption_provisionable);
#endif
#ifdef USE_API_OUTGOING_CONNECTION
dump_field(out, ESPHOME_PSTR("api_outgoing_connection_supported"), this->api_outgoing_connection_supported);
#endif
return out.c_str();
}
+135 -47
View File
@@ -34,7 +34,46 @@ APIServer::APIServer() { global_api_server = this; }
void APIServer::socket_failed_(const LogString *msg) {
ESP_LOGW(TAG, "Socket %s: errno %d", LOG_STR_ARG(msg), errno);
this->destroy_socket_();
this->mark_failed();
}
bool APIServer::create_listen_socket_() {
this->socket_ = socket::socket_ip_loop_monitored(SOCK_STREAM, 0).release(); // monitored for incoming connections
if (this->socket_ == nullptr) {
this->socket_failed_(LOG_STR("creation"));
return false;
}
int enable = 1;
int err = this->socket_->setsockopt(SOL_SOCKET, SO_REUSEADDR, &enable, sizeof(int));
if (err != 0) {
ESP_LOGW(TAG, "Socket reuseaddr: errno %d", errno);
// we can still continue
}
err = this->socket_->setblocking(false);
if (err != 0) {
this->socket_failed_(LOG_STR("nonblocking"));
return false;
}
struct sockaddr_storage server;
socklen_t sl = socket::set_sockaddr_any((struct sockaddr *) &server, sizeof(server), this->port_);
if (sl == 0) {
this->socket_failed_(LOG_STR("set sockaddr"));
return false;
}
err = this->socket_->bind((struct sockaddr *) &server, sl);
if (err != 0) {
this->socket_failed_(LOG_STR("bind"));
return false;
}
err = this->socket_->listen(this->listen_backlog_);
if (err != 0) {
this->socket_failed_(LOG_STR("listen"));
return false;
}
return true;
}
void APIServer::setup() {
@@ -53,41 +92,14 @@ void APIServer::setup() {
#endif
#endif
this->socket_ = socket::socket_ip_loop_monitored(SOCK_STREAM, 0).release(); // monitored for incoming connections
if (this->socket_ == nullptr) {
this->socket_failed_(LOG_STR("creation"));
return;
}
int enable = 1;
int err = this->socket_->setsockopt(SOL_SOCKET, SO_REUSEADDR, &enable, sizeof(int));
if (err != 0) {
ESP_LOGW(TAG, "Socket reuseaddr: errno %d", errno);
// we can still continue
}
err = this->socket_->setblocking(false);
if (err != 0) {
this->socket_failed_(LOG_STR("nonblocking"));
return;
}
struct sockaddr_storage server;
socklen_t sl = socket::set_sockaddr_any((struct sockaddr *) &server, sizeof(server), this->port_);
if (sl == 0) {
this->socket_failed_(LOG_STR("set sockaddr"));
return;
}
err = this->socket_->bind((struct sockaddr *) &server, sl);
if (err != 0) {
this->socket_failed_(LOG_STR("bind"));
return;
}
err = this->socket_->listen(this->listen_backlog_);
if (err != 0) {
this->socket_failed_(LOG_STR("listen"));
if (!this->create_listen_socket_()) {
#ifdef USE_API_OUTGOING_CONNECTION
// Dial-out needs no listener; degrade instead of stopping the component
this->status_set_error(LOG_STR("listen socket failed"));
#else
this->mark_failed();
return;
#endif
}
#ifdef USE_LOGGER
@@ -135,6 +147,9 @@ void APIServer::setup() {
if (this->reboot_timeout_ != 0 && !this->provisioning_pending_()) {
this->status_set_warning(LOG_STR("waiting for client connection"));
}
#ifdef USE_API_OUTGOING_CONNECTION
this->outgoing_conn_.setup();
#endif
}
void APIServer::loop() {
@@ -143,6 +158,12 @@ void APIServer::loop() {
this->accept_new_connections_();
}
#ifdef USE_API_OUTGOING_CONNECTION
if (!this->shutting_down_) {
this->outgoing_conn_.loop(this);
}
#endif
if (this->api_connection_count_ == 0) {
// Check reboot timeout - done in loop to avoid scheduler heap churn
// (cancelled scheduler items sit in heap memory until their scheduled time).
@@ -151,7 +172,12 @@ void APIServer::loop() {
if (this->reboot_timeout_ != 0 && !this->provisioning_pending_()) {
const uint32_t now = App.get_loop_component_start_time();
if (now - this->last_connected_ > this->reboot_timeout_) {
ESP_LOGE(TAG, "No clients; rebooting");
// Distinguish a wrong-key peer from nothing connecting at all
if (this->saw_unauthenticated_client_) {
ESP_LOGE(TAG, "Clients connected but none authenticated; rebooting");
} else {
ESP_LOGE(TAG, "No clients; rebooting");
}
App.reboot();
}
}
@@ -203,6 +229,15 @@ void APIServer::remove_client_(uint8_t client_index) {
std::string client_peername(client->get_peername_to(peername_buf));
#endif
// Read before the swap-and-reset below destroys the connection
const bool was_authenticated = client->is_authenticated();
#ifdef USE_API_OUTGOING_CONNECTION
if (client->flags_.outgoing_connection_target) {
this->outgoing_target_count_--;
}
this->outgoing_conn_.on_client_removed(client.get(), was_authenticated);
#endif
// Close socket now (was deferred from on_fatal_error to allow getpeername)
client->helper_->close();
@@ -221,9 +256,18 @@ void APIServer::remove_client_(uint8_t client_index) {
// Last client disconnected - set warning and start tracking for reboot timeout
// (suppressed while provisioning is pending - see loop()).
// Refresh on every authenticated removal, not just the last one, so an
// unauthenticated straggler removed later (e.g. a port scan, or a dial to
// a host that accepts TCP but never speaks the API) cannot discard a
// healthy session's timestamp and trigger a spurious reboot
if (was_authenticated) {
this->last_connected_ = App.get_loop_component_start_time();
this->saw_unauthenticated_client_ = false;
} else {
this->saw_unauthenticated_client_ = true;
}
if (this->api_connection_count_ == 0 && this->reboot_timeout_ != 0 && !this->provisioning_pending_()) {
this->status_set_warning(LOG_STR("waiting for client connection"));
this->last_connected_ = App.get_loop_component_start_time();
}
#ifdef USE_API_CLIENT_DISCONNECTED_TRIGGER
@@ -245,7 +289,7 @@ void __attribute__((flatten)) APIServer::accept_new_connections_() {
sock->getpeername_to(peername);
// Check if we're at the connection limit
if (this->api_connection_count_ >= MAX_API_CONNECTIONS) {
if (this->at_client_limit_()) {
ESP_LOGW(TAG, "Max connections (%d), rejecting %s", MAX_API_CONNECTIONS, peername);
// Immediately close - socket destructor will handle cleanup
sock.reset();
@@ -254,18 +298,54 @@ void __attribute__((flatten)) APIServer::accept_new_connections_() {
ESP_LOGD(TAG, "Accept %s", peername);
auto *conn = new APIConnection(std::move(sock), this);
this->clients_[this->api_connection_count_++].reset(conn);
conn->start();
// First client connected - clear warning and update timestamp
if (this->api_connection_count_ == 1 && this->reboot_timeout_ != 0 && !this->provisioning_pending_()) {
this->status_clear_warning();
this->last_connected_ = App.get_loop_component_start_time();
}
this->add_client_(new APIConnection(std::move(sock), this));
}
}
bool APIServer::add_client_(APIConnection *conn) {
if (this->at_client_limit_()) {
// The accept path checks first to skip the allocation; the outgoing
// handoff relies on this check
ESP_LOGW(TAG, "Max connections (%d), dropping client", MAX_API_CONNECTIONS);
delete conn;
return false;
}
this->clients_[this->api_connection_count_++].reset(conn);
conn->start();
// First client connected - clear warning. The reboot watchdog timestamp is
// refreshed when an authenticated client is removed (see remove_client_),
// never on bare TCP connects.
if (this->api_connection_count_ == 1 && this->reboot_timeout_ != 0 && !this->provisioning_pending_()) {
this->status_clear_warning();
}
return true;
}
#ifdef USE_API_OUTGOING_CONNECTION
APIConnection *APIServer::add_outgoing_client_(std::unique_ptr<socket::Socket> sock) {
// Re-check at the handoff: the PSK may have been cleared since the dial
// started (mark_outgoing() needs the noise helper); add_client_ re-checks
// the slot limit
if (!this->noise_ctx_.has_psk()) {
ESP_LOGW(TAG, "Dropping outgoing connection (no key)");
return nullptr;
}
auto *conn = new APIConnection(std::move(sock), this);
if (!this->add_client_(conn)) {
return nullptr;
}
// After start(): sends our server hello first so the peer can pick the key
conn->mark_outgoing();
return conn;
}
void APIServer::on_outgoing_target_client(APIConnection *conn) {
this->outgoing_target_count_++;
this->outgoing_conn_.on_target_client(conn);
}
#endif
void APIServer::dump_config() {
char addr_buf[network::USE_ADDRESS_BUFFER_SIZE];
ESP_LOGCONFIG(TAG,
@@ -282,6 +362,9 @@ void APIServer::dump_config() {
#else
ESP_LOGCONFIG(TAG, " Noise encryption: NO");
#endif
#ifdef USE_API_OUTGOING_CONNECTION
this->outgoing_conn_.dump_config();
#endif
}
void APIServer::handle_disconnect(APIConnection *conn) {}
@@ -577,6 +660,8 @@ bool APIServer::update_noise_psk_(const SavedNoisePsk &new_psk, const LogString
if (!c->send_message(req)) {
API_LOG_MSG_DROPPED(TAG, "Disconnect request");
}
// Force it: a session from before the key was active must not survive
c->flags_.next_close = true;
}
});
}
@@ -678,6 +763,9 @@ void APIServer::on_shutdown() {
// Close the listening socket to prevent new connections
this->destroy_socket_();
#ifdef USE_API_OUTGOING_CONNECTION
this->outgoing_conn_.on_shutdown();
#endif
// Change batch delay to 5ms for quick flushing during shutdown
this->batch_delay_ = 5;
+28 -1
View File
@@ -11,6 +11,7 @@
#endif
#include "api_pb2.h"
#include "api_pb2_service.h"
#include "api_outgoing_connection.h"
#include "esphome/components/socket/socket.h"
#include "esphome/core/automation.h"
#include "esphome/core/component.h"
@@ -86,6 +87,10 @@ class APIServer final : public Component,
void set_noise_psk(const uint8_t *psk) { this->noise_ctx_.set_psk(psk); }
noise::NoiseContext &get_noise_ctx() { return this->noise_ctx_; }
#endif // USE_API_NOISE
#ifdef USE_API_OUTGOING_CONNECTION
// Called by APIConnection when a client declares itself a dial-back target in its hello
void on_outgoing_target_client(APIConnection *conn);
#endif
void handle_disconnect(APIConnection *conn);
#ifdef USE_BINARY_SENSOR
@@ -263,6 +268,16 @@ class APIServer final : public Component,
protected:
// Accept incoming socket connections. Only called when socket has pending connections.
void __attribute__((noinline)) accept_new_connections_();
// Insert a constructed connection into the client slots and start it.
// Takes ownership; deletes the connection and returns false at the limit
bool add_client_(APIConnection *conn);
bool at_client_limit_() const { return this->api_connection_count_ >= MAX_API_CONNECTIONS; }
#ifdef USE_API_OUTGOING_CONNECTION
// Returns the new connection, or nullptr (socket dropped) when at the limit
APIConnection *add_outgoing_client_(std::unique_ptr<socket::Socket> sock);
bool has_outgoing_target_client_() const { return this->outgoing_target_count_ != 0; }
friend class OutgoingConnectionManager;
#endif
// Remove a disconnected client by index. Swaps with the last populated slot and resets it.
void __attribute__((noinline)) remove_client_(uint8_t client_index);
@@ -304,6 +319,7 @@ class APIServer final : public Component,
this->socket_ = nullptr;
}
void socket_failed_(const LogString *msg);
bool create_listen_socket_();
// Pointers and pointer-like types first (4 bytes each)
socket::ListenSocket *socket_{nullptr};
#ifdef USE_API_CLIENT_CONNECTED_TRIGGER
@@ -356,8 +372,16 @@ class APIServer final : public Component,
// Connection limits - these defaults will be overridden by config values
// from cv.SplitDefault in __init__.py which sets platform-specific defaults.
uint8_t listen_backlog_{4};
bool shutting_down_ = false;
// Bit-packed so the two flags share one byte
bool shutting_down_ : 1 = false;
// For the reboot log: whether any removal since the last watchdog refresh
// was an unauthenticated session (e.g. a wrong-key peer)
bool saw_unauthenticated_client_ : 1 = false;
uint8_t api_connection_count_{0};
#ifdef USE_API_OUTGOING_CONNECTION
// Connected clients whose hello declared them a dial-back target
uint8_t outgoing_target_count_{0};
#endif
#if defined(USE_PROVISIONING) && defined(USE_API_NOISE)
// Index assigned by the provisioning manager for reporting this transport's state.
uint8_t provisioning_source_{0};
@@ -370,6 +394,9 @@ class APIServer final : public Component,
#endif
ESPPreferenceObject noise_pref_;
#endif // USE_API_NOISE
#ifdef USE_API_OUTGOING_CONNECTION
OutgoingConnectionManager outgoing_conn_;
#endif
};
extern APIServer *global_api_server; // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
@@ -42,7 +42,16 @@ bool AsyncClient::connect(const char *host, uint16_t port) {
return false;
}
socket_->setblocking(false);
if (socket_->setblocking(false) != 0) {
// Capture before the log and reset() below can clobber errno; a blocking
// connect()/read() would otherwise stall the whole loop
const int saved_errno = errno;
ESP_LOGE(TAG, "Failed to set nonblocking: errno %d", saved_errno);
socket_.reset();
if (error_cb_)
error_cb_(error_arg_, this, saved_errno);
return false;
}
int err = socket_->connect((struct sockaddr *) &addr, addrlen);
if (err == 0) {
@@ -97,45 +106,22 @@ void AsyncClient::loop() {
return;
if (connecting_) {
// For connecting, we need to check writability, not readability
// The Application's select() only monitors read FDs, so we do our own check here
// For ESP platforms lwip_select() might be faster, but this code isn't used
// on those platforms anyway. If it was, we'd fix the Application select()
// to report writability instead of doing it this way.
int fd = socket_->get_fd();
if (fd < 0) {
ESP_LOGW(TAG, "Invalid socket fd");
close();
return;
}
fd_set writefds;
FD_ZERO(&writefds);
FD_SET(fd, &writefds);
struct timeval tv = {0, 0};
int ret = select(fd + 1, nullptr, &writefds, nullptr, &tv);
if (ret > 0 && FD_ISSET(fd, &writefds)) {
int error = 0;
socklen_t len = sizeof(error);
if (socket_->getsockopt(SOL_SOCKET, SO_ERROR, &error, &len) == 0 && error == 0) {
int err = 0;
switch (socket::poll_connect(*socket_, err)) {
case socket::ConnectPollResult::CONNECT_POLL_PENDING:
break;
case socket::ConnectPollResult::CONNECT_POLL_CONNECTED:
connecting_ = false;
connected_ = true;
if (connect_cb_)
connect_cb_(connect_arg_, this);
} else {
ESP_LOGW(TAG, "Connection failed: %d", error);
break;
case socket::ConnectPollResult::CONNECT_POLL_ERROR:
ESP_LOGW(TAG, "Connection failed: %d", err);
close();
if (error_cb_)
error_cb_(error_arg_, this, error);
}
} else if (ret < 0) {
const int err = errno;
ESP_LOGE(TAG, "Select error: %d", err);
close();
if (error_cb_)
error_cb_(error_arg_, this, err);
error_cb_(error_arg_, this, err);
break;
}
} else if (connected_) {
// For connected sockets, use the Application's select() results
+1
View File
@@ -23,6 +23,7 @@ CONF_ENABLE_OTA_DOWNGRADE_PROTECTION = "enable_ota_downgrade_protection"
CONF_ENABLED = "enabled"
CONF_GYROSCOPE_ODR = "gyroscope_odr"
CONF_GYROSCOPE_RANGE = "gyroscope_range"
CONF_HOST = "host"
CONF_IAQ = "iaq"
CONF_IGNORE_NOT_FOUND = "ignore_not_found"
CONF_IS_WRGB = "is_wrgb"
@@ -444,7 +444,10 @@ void ESPHomeOTAComponent::handle_data_() {
tv.tv_usec = 0;
this->client_->setsockopt(SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
this->client_->setsockopt(SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
this->client_->setblocking(true);
if (this->client_->setblocking(true) != 0) {
this->log_socket_error_(LOG_STR("blocking"));
goto error; // NOLINT(cppcoreguidelines-avoid-goto)
}
// Acknowledge auth OK - 1 byte
this->data_write_byte_(ota::OTA_RESPONSE_AUTH_OK);
+2
View File
@@ -17,6 +17,8 @@ CONF_IMPLEMENTATION = "implementation"
IMPLEMENTATION_LWIP_TCP = "lwip_tcp"
IMPLEMENTATION_LWIP_SOCKETS = "lwip_sockets"
IMPLEMENTATION_BSD_SOCKETS = "bsd_sockets"
# Implementations whose sockets cannot make outgoing connections
IMPLEMENTATIONS_WITHOUT_CONNECT = frozenset({IMPLEMENTATION_LWIP_TCP})
# Socket tracking infrastructure
# Components register their socket needs and platforms read this to configure appropriately
@@ -59,13 +59,15 @@ int BSDSocketImpl::close() {
int BSDSocketImpl::setblocking(bool blocking) {
int fl = ::fcntl(this->fd_, F_GETFL, 0);
if (fl < 0) {
return fl;
}
if (blocking) {
fl &= ~O_NONBLOCK;
} else {
fl |= O_NONBLOCK;
}
::fcntl(this->fd_, F_SETFL, fl);
return 0;
return ::fcntl(this->fd_, F_SETFL, fl);
}
size_t BSDSocketImpl::getpeername_to(std::span<char, SOCKADDR_STR_LEN> buf) {
@@ -49,13 +49,15 @@ int LwIPSocketImpl::close() {
int LwIPSocketImpl::setblocking(bool blocking) {
int fl = lwip_fcntl(this->fd_, F_GETFL, 0);
if (fl < 0) {
return fl;
}
if (blocking) {
fl &= ~O_NONBLOCK;
} else {
fl |= O_NONBLOCK;
}
lwip_fcntl(this->fd_, F_SETFL, fl);
return 0;
return lwip_fcntl(this->fd_, F_SETFL, fl);
}
size_t LwIPSocketImpl::getpeername_to(std::span<char, SOCKADDR_STR_LEN> buf) {
+54 -2
View File
@@ -2,6 +2,9 @@
#if defined(USE_SOCKET_IMPL_LWIP_TCP) || defined(USE_SOCKET_IMPL_LWIP_SOCKETS) || defined(USE_SOCKET_IMPL_BSD_SOCKETS)
#include <cerrno>
#include <cstring>
#ifdef USE_SOCKET_IMPL_BSD_SOCKETS
#include <sys/select.h>
#endif
#include <string>
#include "esphome/core/log.h"
#include "esphome/core/application.h"
@@ -165,7 +168,10 @@ socklen_t set_sockaddr(struct sockaddr *addr, socklen_t addrlen, const char *ip_
#else
// Use LWIP-specific functions
ip6_addr_t ip6;
inet6_aton(ip_address, &ip6);
if (inet6_aton(ip_address, &ip6) == 0) {
errno = EINVAL;
return 0;
}
memcpy(server->sin6_addr.un.u32_addr, ip6.addr, sizeof(ip6.addr));
#endif
return sizeof(sockaddr_in6);
@@ -185,12 +191,58 @@ socklen_t set_sockaddr(struct sockaddr *addr, socklen_t addrlen, const char *ip_
return 0;
}
#else
server->sin_addr.s_addr = inet_addr(ip_address);
// Unlike inet_addr(), inet_aton() can signal failure while still
// accepting the broadcast address 255.255.255.255
if (inet_aton(ip_address, &server->sin_addr) == 0) {
errno = EINVAL;
return 0;
}
#endif
server->sin_port = htons(port);
return sizeof(sockaddr_in);
}
#if defined(USE_SOCKET_IMPL_BSD_SOCKETS) || defined(USE_SOCKET_IMPL_LWIP_SOCKETS)
ConnectPollResult poll_connect(Socket &sock, int &err_out) {
int fd = sock.get_fd();
if (fd < 0 || fd >= FD_SETSIZE) {
// FD_SET on either is undefined behavior
err_out = EBADF;
return ConnectPollResult::CONNECT_POLL_ERROR;
}
// Connect completion is a write event; the main loop only selects on reads
fd_set writefds;
FD_ZERO(&writefds);
FD_SET(fd, &writefds);
struct timeval tv = {0, 0};
#ifdef USE_SOCKET_IMPL_LWIP_SOCKETS
// LWIP_COMPAT_SOCKETS may be off (LibreTiny), so use the lwip symbol directly
int ret = lwip_select(fd + 1, nullptr, &writefds, nullptr, &tv);
#else
// Global-scope select: the entity namespace esphome::select shadows it here
int ret = ::select(fd + 1, nullptr, &writefds, nullptr, &tv);
#endif
if (ret < 0) {
err_out = errno;
return ConnectPollResult::CONNECT_POLL_ERROR;
}
if (ret == 0 || !FD_ISSET(fd, &writefds)) {
return ConnectPollResult::CONNECT_POLL_PENDING;
}
int error = 0;
socklen_t len = sizeof(error);
if (sock.getsockopt(SOL_SOCKET, SO_ERROR, &error, &len) != 0) {
err_out = errno;
return ConnectPollResult::CONNECT_POLL_ERROR;
}
if (error != 0) {
err_out = error;
return ConnectPollResult::CONNECT_POLL_ERROR;
}
return ConnectPollResult::CONNECT_POLL_CONNECTED;
}
#endif
socklen_t set_sockaddr_any(struct sockaddr *addr, socklen_t addrlen, uint16_t port) {
#if USE_NETWORK_IPV6
if (addrlen < sizeof(sockaddr_in6)) {
+13
View File
@@ -145,6 +145,19 @@ inline socklen_t set_sockaddr(struct sockaddr *addr, socklen_t addrlen, const st
/// Set a sockaddr to the any address and specified port for the IP version used by socket_ip().
socklen_t set_sockaddr_any(struct sockaddr *addr, socklen_t addrlen, uint16_t port);
#if defined(USE_SOCKET_IMPL_BSD_SOCKETS) || defined(USE_SOCKET_IMPL_LWIP_SOCKETS)
enum class ConnectPollResult : uint8_t {
CONNECT_POLL_PENDING,
CONNECT_POLL_CONNECTED,
CONNECT_POLL_ERROR,
};
/// Check a non-blocking connect() for completion without blocking. On
/// CONNECT_POLL_ERROR, err_out holds the socket's SO_ERROR, or errno when the
/// poll itself failed.
ConnectPollResult poll_connect(Socket &sock, int &err_out);
#endif
/// Format sockaddr into caller-provided buffer, returns length written (excluding null)
size_t format_sockaddr_to(const struct sockaddr *addr_ptr, socklen_t len, std::span<char, SOCKADDR_STR_LEN> buf);
+1 -1
View File
@@ -1,5 +1,6 @@
import esphome.codegen as cg
from esphome.components import binary_sensor, sensor
from esphome.components.const import CONF_HOST
import esphome.config_validation as cv
from esphome.const import (
CONF_BINARY_SENSORS,
@@ -14,7 +15,6 @@ AUTO_LOAD = ["socket"]
CODEOWNERS = ["@Links2004"]
DEPENDENCIES = ["network"]
CONF_HOST = "host"
CONF_PREFIX = "prefix"
statsd_component_ns = cg.esphome_ns.namespace("statsd")
+15 -2
View File
@@ -13,9 +13,16 @@ void UDPComponent::setup() {
#if defined(USE_SOCKET_IMPL_BSD_SOCKETS) || defined(USE_SOCKET_IMPL_LWIP_SOCKETS)
for (const auto &address : this->addresses_) {
struct sockaddr saddr {};
socket::set_sockaddr(&saddr, sizeof(saddr), address, this->broadcast_port_);
if (socket::set_sockaddr(&saddr, sizeof(saddr), address, this->broadcast_port_) == 0) {
ESP_LOGW(TAG, "Invalid address %s", address);
continue;
}
this->sockaddrs_.push_back(saddr);
}
if (this->sockaddrs_.size() != this->addresses_.size()) {
// A dropped address silently receives nothing; surface the misconfiguration
this->status_set_warning(LOG_STR("invalid address"));
}
// set up broadcast socket
if (this->should_broadcast_) {
this->broadcast_socket_ = socket::socket(AF_INET, SOCK_DGRAM, IPPROTO_IP);
@@ -94,9 +101,15 @@ void UDPComponent::setup() {
// 8266 and RP2040 `Duino
for (const auto &address : this->addresses_) {
auto ipaddr = IPAddress();
ipaddr.fromString(address);
if (!ipaddr.fromString(address)) {
ESP_LOGW(TAG, "Invalid address %s", address);
continue;
}
this->ipaddrs_.push_back(ipaddr);
}
if (this->ipaddrs_.size() != this->addresses_.size()) {
this->status_set_warning(LOG_STR("invalid address"));
}
if (this->should_listen_)
this->udp_client_.begin(this->listen_port_);
#endif
@@ -34,6 +34,10 @@ void WakeOnLanButton::press_action() {
struct sockaddr_storage saddr {};
auto addr_len =
socket::set_sockaddr(reinterpret_cast<sockaddr *>(&saddr), sizeof(saddr), "255.255.255.255", this->port_);
if (addr_len == 0) {
ESP_LOGW(TAG, "Invalid broadcast address");
return;
}
uint8_t buffer[6 + sizeof this->macaddr_ * 16];
memcpy(buffer, PREFIX, sizeof(PREFIX));
for (size_t i = 0; i != 16; i++) {
+5
View File
@@ -216,6 +216,11 @@
#define USE_API_HOMEASSISTANT_SERVICES
#define USE_API_HOMEASSISTANT_STATES
#define USE_API_NOISE
#if !defined(USE_ESP8266) && !defined(USE_RP2) // raw-lwip sockets cannot make outgoing connections
#define USE_API_OUTGOING_CONNECTION
#define API_OUTGOING_CONNECTION_PORT 6054
#define API_OUTGOING_CONNECTION_DELAY 60000
#endif
#define USE_API_VARINT64
#define USE_API_PLAINTEXT
#define USE_API_USER_DEFINED_ACTIONS
+17
View File
@@ -1104,6 +1104,10 @@ def get_components_per_integration_fixture() -> dict[str, set[str]]:
_TEST_FUNC_RE = re.compile(r"async def (test_\w+)")
# Any usage form (decorator, pytestmark assignment or list element); only
# test_*.py files are scanned, so the marker docs elsewhere cannot false-hit
_SHARED_YAML_USE_RE = re.compile(r"\bmark\.shared_yaml")
_SHARED_YAML_ARG_RE = re.compile(r"\(\s*[\"'](\w+)[\"']\s*\)")
@cache
@@ -1123,6 +1127,19 @@ def get_fixture_to_test_files() -> dict[str, frozenset[str]]:
for func in _TEST_FUNC_RE.findall(content):
base_name = func.replace("test_", "").partition("[")[0]
result.setdefault(base_name, set()).add(rel_path)
# Shared fixtures are named by marker, not by a test function; each
# decorator must carry a string literal or its fixture would silently
# map to no tests
for use in _SHARED_YAML_USE_RE.finditer(content):
arg = _SHARED_YAML_ARG_RE.match(content, use.end())
if arg is None:
line = content.count("\n", 0, use.start()) + 1
raise ValueError(
f"{rel_path}:{line}: shared_yaml marker must take a "
"single-line string literal so CI test selection can map "
"its fixture"
)
result.setdefault(arg.group(1), set()).add(rel_path)
return {k: frozenset(v) for k, v in result.items()}
@@ -0,0 +1,148 @@
"""Tests for the api outgoing_connection option."""
from collections.abc import Callable
from pathlib import Path
import pytest
from esphome.components import socket
from esphome.components.api import (
CONFIG_SCHEMA,
_validate_outgoing_host_ipv6,
_validate_outgoing_socket_implementation,
)
from esphome.components.esp32 import KEY_BOARD, KEY_VARIANT, VARIANT_ESP32
import esphome.config_validation as cv
from esphome.const import PlatformFramework
from esphome.core import CORE
import esphome.final_validate as fv
from esphome.types import ConfigType
from tests.component_tests.types import SetCoreConfigCallable
KEY = "bOFFzzvfpg5DB94DuBGLXD/hMnhpDKgP9UQyBulwWVU="
ESP32_PLATFORM_DATA = {KEY_BOARD: "esp32dev", KEY_VARIANT: VARIANT_ESP32}
def _api_config(outgoing: ConfigType, *, encryption: bool = True) -> ConfigType:
config: ConfigType = {"outgoing_connection": outgoing}
if encryption:
config["encryption"] = {"key": KEY}
return config
def test_outgoing_connection_generates_defines(
generate_main: Callable[[str | Path], str],
) -> None:
"""A valid config emits the compile-time defines with defaults applied."""
generate_main("tests/component_tests/api/test_outgoing_connection.yaml")
defines = {define.name: define.value for define in CORE.defines}
assert "USE_API_OUTGOING_CONNECTION" in defines
assert str(defines["API_OUTGOING_CONNECTION_HOST"]) == '"192.168.1.2"'
assert str(defines["API_OUTGOING_CONNECTION_PORT"]) == "6054"
assert str(defines["API_OUTGOING_CONNECTION_DELAY"]) == "60000"
def test_outgoing_connection_defaults(
set_core_config: SetCoreConfigCallable,
) -> None:
set_core_config(PlatformFramework.ESP32_IDF, platform_data=ESP32_PLATFORM_DATA)
config = CONFIG_SCHEMA(_api_config({"host": "192.168.1.2"}))
outgoing = config["outgoing_connection"]
assert outgoing["port"] == 6054
assert outgoing["delay"].total_milliseconds == 60000
def test_outgoing_connection_bare_block(
set_core_config: SetCoreConfigCallable,
) -> None:
"""A bare outgoing_connection: block is valid; the device dials the
remembered last dial-back client."""
set_core_config(PlatformFramework.ESP32_IDF, platform_data=ESP32_PLATFORM_DATA)
config = CONFIG_SCHEMA(_api_config(None))
outgoing = config["outgoing_connection"]
assert "host" not in outgoing
assert outgoing["port"] == 6054
def test_outgoing_connection_delay_bounded(
set_core_config: SetCoreConfigCallable,
) -> None:
"""A delay past half the uint32 millisecond range is rejected, not wrapped."""
set_core_config(PlatformFramework.ESP32_IDF, platform_data=ESP32_PLATFORM_DATA)
with pytest.raises(cv.Invalid, match="value must be at most"):
CONFIG_SCHEMA(_api_config({"delay": "60d"}))
def test_outgoing_connection_requires_encryption(
set_core_config: SetCoreConfigCallable,
) -> None:
set_core_config(PlatformFramework.ESP32_IDF, platform_data=ESP32_PLATFORM_DATA)
with pytest.raises(cv.Invalid, match="requires 'encryption'"):
CONFIG_SCHEMA(_api_config({"host": "192.168.1.2"}, encryption=False))
@pytest.mark.parametrize(
("platform_framework", "platform_data", "socket_conf"),
[
# The platform default on these two, resolved like AUTO_LOAD does
(PlatformFramework.ESP8266_ARDUINO, None, None),
(PlatformFramework.RP2040_ARDUINO, None, None),
# An explicit selection elsewhere
(
PlatformFramework.ESP32_IDF,
ESP32_PLATFORM_DATA,
{"implementation": "lwip_tcp"},
),
],
)
def test_outgoing_connection_rejects_lwip_tcp(
set_core_config: SetCoreConfigCallable,
platform_framework: PlatformFramework,
platform_data: ConfigType | None,
socket_conf: ConfigType | None,
) -> None:
"""The resolved lwip_tcp socket is rejected at final validate."""
set_core_config(platform_framework, platform_data=platform_data)
fv.full_config.set({"socket": socket_conf or socket.CONFIG_SCHEMA({})})
config = CONFIG_SCHEMA(_api_config({"host": "192.168.1.2"}))
with pytest.raises(cv.Invalid, match="lwip_tcp"):
_validate_outgoing_socket_implementation(config)
def test_outgoing_connection_rejects_hostnames(
set_core_config: SetCoreConfigCallable,
) -> None:
set_core_config(PlatformFramework.ESP32_IDF, platform_data=ESP32_PLATFORM_DATA)
with pytest.raises(cv.Invalid, match="not a valid IP address"):
CONFIG_SCHEMA(_api_config({"host": "homeassistant.local"}))
def test_outgoing_connection_ipv6_host_requires_ipv6(
set_core_config: SetCoreConfigCallable,
) -> None:
set_core_config(PlatformFramework.ESP32_IDF, platform_data=ESP32_PLATFORM_DATA)
config = CONFIG_SCHEMA(_api_config({"host": "fd00::1"}))
with pytest.raises(cv.Invalid, match="IPv6 is not"):
_validate_outgoing_host_ipv6(config)
def test_outgoing_connection_ipv6_host_passes_with_ipv6_enabled(
set_core_config: SetCoreConfigCallable,
) -> None:
set_core_config(
PlatformFramework.ESP32_IDF,
platform_data=ESP32_PLATFORM_DATA,
full_config={"network": {"enable_ipv6": True}},
)
config = CONFIG_SCHEMA(_api_config({"host": "fd00::1"}))
assert _validate_outgoing_host_ipv6(config) is config
def test_outgoing_connection_ipv6_host_with_ipv6(
generate_main: Callable[[str | Path], str],
) -> None:
generate_main("tests/component_tests/api/test_outgoing_connection_ipv6.yaml")
defines = {define.name: define.value for define in CORE.defines}
assert str(defines["API_OUTGOING_CONNECTION_HOST"]) == '"fd00::1"'
@@ -0,0 +1,17 @@
esphome:
name: test
esp32:
board: esp32dev
wifi:
ssid: SomeNetwork
password: SomePassword
logger:
api:
encryption:
key: bOFFzzvfpg5DB94DuBGLXD/hMnhpDKgP9UQyBulwWVU=
outgoing_connection:
host: 192.168.1.2
@@ -0,0 +1,20 @@
esphome:
name: test
esp32:
board: esp32dev
wifi:
ssid: SomeNetwork
password: SomePassword
network:
enable_ipv6: true
logger:
api:
encryption:
key: bOFFzzvfpg5DB94DuBGLXD/hMnhpDKgP9UQyBulwWVU=
outgoing_connection:
host: fd00::1
@@ -0,0 +1,13 @@
packages:
common: !include common-base.yaml
wifi:
ssid: MySSID
password: password1
# Outgoing connection on the lwip_sockets implementation used by LibreTiny
api:
encryption:
key: bOFFzzvfpg5DB94DuBGLXD/hMnhpDKgP9UQyBulwWVU=
outgoing_connection:
host: 192.168.1.2
@@ -0,0 +1,16 @@
packages:
common: !include common-base.yaml
wifi:
ssid: MySSID
password: password1
# Outgoing connection: the device dials out when no dial-back client is
# connected. Requires encryption so the peer is verified by key.
api:
encryption:
key: bOFFzzvfpg5DB94DuBGLXD/hMnhpDKgP9UQyBulwWVU=
outgoing_connection:
host: 192.168.1.2
port: 6054
delay: 60s
@@ -0,0 +1,11 @@
packages:
common: !include common-base.yaml
network:
# No host set: the device dials the last remembered Home Assistant address
api:
encryption:
key: bOFFzzvfpg5DB94DuBGLXD/hMnhpDKgP9UQyBulwWVU=
outgoing_connection:
delay: 30s
+7
View File
@@ -21,6 +21,13 @@ The `yaml_config` fixture automatically loads YAML configurations based on the t
- The fixture file must exist or the test will fail with a clear error message
- The fixture automatically injects a dynamic port number into the API configuration
Tests marked `@pytest.mark.shared_yaml("name")` load `fixtures/name.yaml` instead
of the test-named file and compile it in a shared, hash-keyed build directory, so
the whole group pays one full compile and each test only a relink. The marker
argument must be a single-line string literal (CI test selection maps fixtures to
test files by scanning for it), and marked tests must hand the `yaml_config`
content to `run_compiled` unmodified.
### Key Fixtures
- `run_compiled` - Combines write, compile, and run operations into a single context manager
+335 -81
View File
@@ -4,17 +4,22 @@ from __future__ import annotations
import asyncio
from collections.abc import AsyncGenerator, Callable, Generator
from contextlib import AbstractAsyncContextManager, asynccontextmanager
from contextlib import AbstractAsyncContextManager, asynccontextmanager, suppress
import fcntl
from functools import cache
import hashlib
import logging
import os
from pathlib import Path
import platform
import re
import shutil
import signal
import socket
import subprocess
import sys
import tempfile
import time
from typing import TextIO
from aioesphomeapi import APIClient, APIConnectionError, LogParser, ReconnectLogic
@@ -23,7 +28,13 @@ import pytest_asyncio
import esphome.config
from esphome.core import CORE
from esphome.helpers import get_usable_cpu_count
from esphome.helpers import (
get_usable_cpu_count,
read_file,
rmtree,
write_file,
write_file_if_changed,
)
from esphome.platformio.toolchain import get_idedata
from .const import (
@@ -56,6 +67,21 @@ import pty # not available on Windows
pytest.register_assert_rewrite("tests.integration.entity_utils")
def pytest_configure(config: pytest.Config) -> None:
config.addinivalue_line(
"markers",
"shared_yaml(name): load fixtures/<name>.yaml and compile it in a shared, "
"hash-keyed incremental build directory",
)
FIXTURES_DIR = Path(__file__).parent / "fixtures"
REPO_ROOT = Path(__file__).resolve().parent.parent.parent
# CI caches parts of this path; keep in sync with ci.yml integration-tests.
INTEGRATION_TESTS_ROOT = Path.home() / ".esphome-integration-tests"
def _get_platformio_env(cache_dir: Path) -> dict[str, str]:
"""Get environment variables for PlatformIO with shared cache."""
env = os.environ.copy()
@@ -78,7 +104,7 @@ def _get_platformio_env(cache_dir: Path) -> dict[str, str]:
)
# Compile with THIS tree's esphome sources, not wherever the venv's editable
# install points (which may be a different git worktree or checkout).
repo_root = str(Path(__file__).resolve().parent.parent.parent)
repo_root = str(REPO_ROOT)
existing = env.get("PYTHONPATH")
env["PYTHONPATH"] = f"{repo_root}{os.pathsep}{existing}" if existing else repo_root
return env
@@ -88,8 +114,7 @@ def _get_platformio_env(cache_dir: Path) -> dict[str, str]:
def shared_platformio_cache() -> Generator[Path]:
"""Initialize a shared PlatformIO cache for all integration tests."""
# Use a dedicated directory for integration tests to avoid conflicts.
# CI caches parts of this path; keep in sync with ci.yml integration-tests.
test_cache_dir = Path.home() / ".esphome-integration-tests"
test_cache_dir = INTEGRATION_TESTS_ROOT
cache_dir = test_cache_dir / "platformio"
# Use a lock file in the home directory to ensure only one process initializes the cache
@@ -112,7 +137,9 @@ def shared_platformio_cache() -> Generator[Path]:
init_dir = Path(tmpdir)
fixture_path = Path(__file__).parent / "fixtures" / "cache_init.yaml"
config_path = init_dir / "cache_init.yaml"
config_path.write_text(fixture_path.read_text())
config_path.write_text(
fixture_path.read_text(encoding="utf-8"), encoding="utf-8"
)
# Run compilation to populate the cache
# We must succeed here to avoid race conditions where multiple
@@ -162,13 +189,6 @@ def integration_test_dir() -> Generator[Path]:
yield Path(tmpdir)
@pytest.fixture
def isolated_preferences(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""Host preferences persist per device name; give the test its own so a
provisioned key never leaks into another run."""
monkeypatch.setenv("ESPHOME_PREFDIR", str(tmp_path / "prefs"))
@pytest.fixture
def reserved_tcp_port() -> Generator[tuple[int, socket.socket]]:
"""Reserve an unused TCP port by holding the socket open."""
@@ -188,21 +208,29 @@ def unused_tcp_port(reserved_tcp_port: tuple[int, socket.socket]) -> int:
return reserved_tcp_port[0]
@pytest.fixture(autouse=True)
def isolated_preferences(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Path:
"""Give every test its own host prefs dir; prefs are keyed only by device
name, which tests sharing a fixture also share."""
prefdir = tmp_path / "prefs"
monkeypatch.setenv("ESPHOME_PREFDIR", str(prefdir))
return prefdir
@pytest_asyncio.fixture
async def yaml_config(request: pytest.FixtureRequest, unused_tcp_port: int) -> str:
"""Load YAML configuration based on test name."""
# Get the test function name
test_name: str = request.node.name
# Extract the base test name (remove test_ prefix and any parametrization)
base_name = test_name.replace("test_", "").partition("[")[0]
shared_name = _shared_yaml_name(request)
# Base test name: test_ prefix and any parametrization stripped
base_name = shared_name or request.node.name.replace("test_", "").partition("[")[0]
# Load the fixture file
fixture_path = Path(__file__).parent / "fixtures" / f"{base_name}.yaml"
fixture_path = FIXTURES_DIR / f"{base_name}.yaml"
if not fixture_path.exists():
raise FileNotFoundError(f"Fixture file not found: {fixture_path}")
loop = asyncio.get_running_loop()
content = await loop.run_in_executor(None, fixture_path.read_text)
content = await loop.run_in_executor(None, read_file, fixture_path)
# Replace the port in the config if it contains api section
if "api:" in content:
@@ -226,11 +254,13 @@ async def yaml_config(request: pytest.FixtureRequest, unused_tcp_port: int) -> s
# Replace external component path placeholder if present
if "EXTERNAL_COMPONENT_PATH" in content:
external_components_path = str(
Path(__file__).parent / "fixtures" / "external_components"
)
external_components_path = str(FIXTURES_DIR / "external_components")
content = content.replace("EXTERNAL_COMPONENT_PATH", external_components_path)
if shared_name is not None:
# _compile verifies the marked test compiles this content unmodified
request.node._shared_yaml_content = content
return content
@@ -240,24 +270,218 @@ async def write_yaml_config(
) -> AsyncGenerator[ConfigWriter]:
"""Write YAML configuration to a file."""
# Get the test name for default filename
test_name = request.node.name
base_name = test_name.replace("test_", "").split("[")[0]
base_name = request.node.name.replace("test_", "").partition("[")[0]
async def _write_config(content: str, filename: str | None = None) -> Path:
if filename is None:
filename = f"{base_name}.yaml"
config_path = integration_test_dir / filename
loop = asyncio.get_running_loop()
await loop.run_in_executor(None, config_path.write_text, content)
await loop.run_in_executor(None, write_file, config_path, content)
return config_path
yield _write_config
# Deliberately not CI-cached (ci.yml caches only platformio/ subpaths); stale
# dirs for a fixture are pruned when its content hash changes.
SHARED_BUILDS_ROOT = INTEGRATION_TESTS_ROOT / "builds"
# In the dir name (not just the hash) so pruning stays inside this checkout
_REPO_KEY = hashlib.sha256(str(REPO_ROOT).encode()).hexdigest()[:8]
# Give a contended shared build lock time for a full cold compile ahead of us
_SHARED_LOCK_TIMEOUT_S = 900
_SHARED_LOCK_POLL_S = 0.1
_SHARED_LOCK_REPORT_S = 30
# Reclaims dirs orphaned by fixture renames or deleted checkouts
_STALE_BUILD_MAX_AGE_S = 30 * 24 * 3600
# ELF path per shared build dir; constant once compiled, so resolve it only once
_shared_elf_paths: dict[Path, Path] = {}
# Dirs this process already swept; pruning is session-scoped work
_pruned_dirs: set[Path] = set()
def _shared_yaml_name(request: pytest.FixtureRequest) -> str | None:
"""Name passed to the shared_yaml marker, or None when unmarked."""
marker = request.node.get_closest_marker("shared_yaml")
if marker is None:
return None
# Exactly one \w+ positional arg: the name doubles as a build dir
# component, and CI test selection (script/helpers.py) parses the same shape
if (
len(marker.args) != 1
or marker.kwargs
or not re.fullmatch(r"\w+", str(marker.args[0]))
):
raise ValueError(
"shared_yaml marker requires exactly one \\w+ fixture name literal"
)
return marker.args[0]
def _shared_build_prefix(name: str) -> str:
return f"{name}-{_REPO_KEY}-"
@cache
def _shared_build_dir(name: str) -> Path:
"""Dir keyed by checkout and fixture source, before per-test injections."""
key = hashlib.sha256((FIXTURES_DIR / f"{name}.yaml").read_bytes()).hexdigest()[:16]
return SHARED_BUILDS_ROOT / (_shared_build_prefix(name) + key)
def _read_stamp(stamp: Path, shared_dir: Path) -> Path | None:
"""ELF path recorded by the last completed compile, or None."""
try:
text = stamp.read_text(encoding="utf-8").strip()
except FileNotFoundError:
return None
except OSError as err:
print(f"Cannot read {stamp}: {err}")
return None
if not text:
print(f"Ignoring empty stamp {stamp}")
return None
built = Path(text)
# Never trust a stamp pointing outside its own build dir as an unlink target
if shared_dir.resolve() in built.resolve().parents:
return built
print(f"Ignoring stamp {stamp} pointing outside {shared_dir}")
return None
def _unused_since(stale: Path, cutoff: float) -> bool:
"""Whether a build dir looks untouched since cutoff; unknown counts as used."""
# Newest of the .built stamp (rewritten by every completed compile) and the
# dir itself (freshened by a worker claiming the dir before locking)
newest: float | None = None
for probe in (stale / ".built", stale):
try:
mtime = probe.stat().st_mtime
except FileNotFoundError:
continue
except NotADirectoryError:
return True # a stray file where a dir should be; reclaimable
except OSError as err:
print(f"Cannot age-probe {stale}: {err}")
return False # unknown never authorizes deletion
newest = mtime if newest is None else max(newest, mtime)
return newest is not None and newest < cutoff
def _prune_stale_builds(name: str, keep: Path) -> None:
"""Remove outdated build dirs (blocking, run in executor): this checkout's
other dirs for the fixture, plus anything untouched for 30 days. Tolerates
other workers pruning the same dirs concurrently."""
cutoff = time.time() - _STALE_BUILD_MAX_AGE_S
prefix = _shared_build_prefix(name)
for stale in SHARED_BUILDS_ROOT.iterdir():
if stale == keep:
continue
same_fixture = stale.name.startswith(prefix)
if not same_fixture and not _unused_since(stale, cutoff):
continue
# Creating .lock bumps the dir mtime, so remember whether the re-probe
# under the lock can trust it
lock_preexisting = (stale / ".lock").exists()
try:
lock_file = (stale / ".lock").open("w")
except FileNotFoundError:
continue # pruned by another worker meanwhile
except NotADirectoryError:
print(f"Removing stray file {stale}")
stale.unlink(missing_ok=True)
continue
except OSError as err:
print(f"Cannot prune {stale}: {err}")
continue
with lock_file:
try:
fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError:
continue # still in use by another run
# Re-probe under the lock: a worker freshens its dir before
# locking, so a just-claimed dir no longer looks unused. A dir
# whose .lock we just created cannot be held by anyone, and our
# own open bumped its mtime, so its pre-open probe stands
if (
lock_preexisting
and not same_fixture
and not _unused_since(stale, cutoff)
):
continue
# rmtree tolerates races; a leftover partial tree only costs a
# rebuild, since the ELF is deleted before every compile
try:
rmtree(stale)
except OSError as err:
print(f"Failed to prune {stale}: {err}")
async def _run_esphome_compile(
config_path: Path, cwd: Path, env: dict[str, str]
) -> None:
"""Run `esphome compile`, retrying up to 3 times on a segfault."""
max_retries = 3
for attempt in range(max_retries):
# Compile using subprocess, inheriting stdout/stderr to show progress
proc = await asyncio.create_subprocess_exec(
sys.executable,
"-m",
"esphome",
"compile",
str(config_path),
cwd=cwd,
stdout=None, # Inherit stdout
stderr=None, # Inherit stderr
stdin=asyncio.subprocess.DEVNULL,
# Start in a new process group to isolate signal handling
start_new_session=True,
env=env,
close_fds=False,
)
await proc.wait()
if proc.returncode == 0:
break
if proc.returncode == -11 and attempt < max_retries - 1:
# Segfault (-11 = SIGSEGV), retry
print(
f"Compilation segfaulted (attempt {attempt + 1}/{max_retries}), retrying..."
)
await asyncio.sleep(1) # Brief pause before retry
continue
raise RuntimeError(
f"Failed to compile {config_path}, return code: {proc.returncode}. "
f"Run with 'pytest -s' to see compilation output."
)
def _resolve_compiled_binary(config_path: Path) -> Path:
"""Load the config to learn the compiled ELF path (blocking, run in executor)."""
CORE.reset() # Reset CORE state between test runs
CORE.config_path = config_path
config = esphome.config.read_config(
{"command": "compile", "config": str(config_path)}
)
if config is None:
raise RuntimeError(f"Failed to read config from {config_path}")
idedata = get_idedata(config)
binary_path = Path(idedata.firmware_elf_path)
if not binary_path.exists():
raise RuntimeError(f"Compiled binary not found at {binary_path}")
return binary_path
@pytest_asyncio.fixture
async def compile_esphome(
integration_test_dir: Path,
shared_platformio_cache: Path,
request: pytest.FixtureRequest,
) -> AsyncGenerator[CompileFunction]:
"""Compile an ESPHome configuration and return the binary path."""
@@ -265,66 +489,96 @@ async def compile_esphome(
# Use the shared PlatformIO cache for faster compilation
# This avoids re-downloading dependencies for each test
env = _get_platformio_env(shared_platformio_cache)
# Retry compilation up to 3 times if we get a segfault
max_retries = 3
for attempt in range(max_retries):
# Compile using subprocess, inheriting stdout/stderr to show progress
proc = await asyncio.create_subprocess_exec(
sys.executable,
"-m",
"esphome",
"compile",
str(config_path),
cwd=integration_test_dir,
stdout=None, # Inherit stdout
stderr=None, # Inherit stderr
stdin=asyncio.subprocess.DEVNULL,
# Start in a new process group to isolate signal handling
start_new_session=True,
env=env,
close_fds=False,
)
await proc.wait()
if proc.returncode == 0:
# Success!
break
if proc.returncode == -11 and attempt < max_retries - 1:
# Segfault (-11 = SIGSEGV), retry
print(
f"Compilation segfaulted (attempt {attempt + 1}/{max_retries}), retrying..."
)
await asyncio.sleep(1) # Brief pause before retry
continue
# Other error or final retry
raise RuntimeError(
f"Failed to compile {config_path}, return code: {proc.returncode}. "
f"Run with 'pytest -s' to see compilation output."
)
# Load the config to get idedata (blocking call, must use executor)
loop = asyncio.get_running_loop()
def _read_config_and_get_binary():
CORE.reset() # Reset CORE state between test runs
CORE.config_path = config_path
config = esphome.config.read_config(
{"command": "compile", "config": str(config_path)}
name = _shared_yaml_name(request)
if name is None:
await _run_esphome_compile(config_path, integration_test_dir, env)
return await loop.run_in_executor(
None, _resolve_compiled_binary, config_path
)
if config is None:
raise RuntimeError(f"Failed to read config from {config_path}")
# Get the compiled binary path
idedata = get_idedata(config)
return Path(idedata.firmware_elf_path)
binary_path = await loop.run_in_executor(None, _read_config_and_get_binary)
if not binary_path.exists():
raise RuntimeError(f"Compiled binary not found at {binary_path}")
return binary_path
# Shared fixture: build in a hash-keyed dir so tests sharing a config
# pay one full compile and later only a main.cpp (port) rebuild + relink
shared_dir = _shared_build_dir(name)
shared_dir.mkdir(parents=True, exist_ok=True)
# Freshen the dir before locking so a concurrent age sweep, which
# re-probes under the lock, never reaps a dir a worker just claimed;
# if a peer reaped it already, the guarded lock open recreates it
with suppress(FileNotFoundError):
os.utime(shared_dir)
if shared_dir not in _pruned_dirs:
_pruned_dirs.add(shared_dir)
await loop.run_in_executor(None, _prune_stale_builds, name, shared_dir)
shared_config = shared_dir / f"{name}.yaml"
private_binary = integration_test_dir / f"{name}.elf"
content = await loop.run_in_executor(None, read_file, config_path)
if content != getattr(request.node, "_shared_yaml_content", None):
# The dir is keyed by the fixture source; a mutated config would be
# cached under a hash that does not describe it
raise RuntimeError(
"shared_yaml tests must compile the yaml_config content unmodified"
)
# flock serializes concurrent xdist workers; closing the fd releases it.
# Hand-rolled rather than filelock.FileLock: non-blocking retries keep
# the wait cancellable, while a blocking acquire in an executor thread
# would survive test cancellation holding the fd
try:
lock_file = (shared_dir / ".lock").open("w")
except FileNotFoundError:
# A peer run pruning divergent hashes reaped the dir between our
# mkdir and this open; recreate it and pay a full rebuild
shared_dir.mkdir(parents=True, exist_ok=True)
lock_file = (shared_dir / ".lock").open("w")
with lock_file:
start = time.monotonic()
last_report = start
while True:
try:
fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
break
except BlockingIOError:
now = time.monotonic()
if now - start > _SHARED_LOCK_TIMEOUT_S:
raise RuntimeError(
f"Timed out waiting for the {shared_dir} lock"
) from None
if now - last_report >= _SHARED_LOCK_REPORT_S:
last_report = now
print(
f"Waited {now - start:.0f}s for another worker's "
f"build of {shared_dir.name}"
)
await asyncio.sleep(_SHARED_LOCK_POLL_S)
# .built carries the ELF path of the last completed compile, so
# later workers skip the config re-read in _resolve_compiled_binary
stamp = shared_dir / ".built"
if (built := _shared_elf_paths.get(shared_dir)) is None:
built = await loop.run_in_executor(None, _read_stamp, stamp, shared_dir)
# Delete the ELF before compiling: whatever exists afterwards is
# this compile's output, so no staleness check is ever needed.
# With no usable stamp, sweep any leftover at the known layout
if built is not None:
built.unlink(missing_ok=True)
else:
# Layout-agnostic: ESPHOME_BUILD_PATH can move the build tree
for leftover in shared_dir.rglob("program"):
if leftover.is_file():
leftover.unlink()
await loop.run_in_executor(
None, write_file_if_changed, shared_config, content
)
await _run_esphome_compile(shared_config, shared_dir, env)
if built is None or not built.exists():
built = await loop.run_in_executor(
None, _resolve_compiled_binary, shared_config
)
_shared_elf_paths[shared_dir] = built
await loop.run_in_executor(None, write_file, stamp, str(built))
# Copy out before unlocking: another worker may relink firmware.elf
# while this test is still running its private copy
await loop.run_in_executor(None, shutil.copy2, built, private_binary)
return private_binary
yield _compile
@@ -0,0 +1,14 @@
esphome:
name: outgoing-conn-test
host:
logger:
api:
encryption:
key: bOFFzzvfpg5DB94DuBGLXD/hMnhpDKgP9UQyBulwWVU=
outgoing_connection:
host: 127.0.0.1
port: OUTGOING_PORT
delay: 1s
@@ -0,0 +1,13 @@
esphome:
name: outgoing-conn-test
host:
logger:
api:
encryption:
key: bOFFzzvfpg5DB94DuBGLXD/hMnhpDKgP9UQyBulwWVU=
outgoing_connection:
port: OUTGOING_PORT
delay: 1s
@@ -0,0 +1,7 @@
esphome:
name: api-reboot-test
host:
api:
reboot_timeout: 2s # Headroom to connect and authenticate a client first
logger:
level: DEBUG
@@ -1,58 +0,0 @@
esphome:
name: test-batch-window-filters
host:
api:
batch_delay: 0ms # Disable batching to receive all state updates
logger:
level: DEBUG
# Template sensor that we'll use to publish values
sensor:
- platform: template
name: "Source Sensor"
id: source_sensor
accuracy_decimals: 2
# Batch window filters (window_size == send_every) - use streaming filters
- platform: copy
source_id: source_sensor
name: "Min Sensor"
id: min_sensor
filters:
- min:
window_size: 5
send_every: 5
send_first_at: 1
- platform: copy
source_id: source_sensor
name: "Max Sensor"
id: max_sensor
filters:
- max:
window_size: 5
send_every: 5
send_first_at: 1
- platform: copy
source_id: source_sensor
name: "Moving Avg Sensor"
id: moving_avg_sensor
filters:
- sliding_window_moving_average:
window_size: 5
send_every: 5
send_first_at: 1
# Button to trigger publishing test values
button:
- platform: template
name: "Publish Values Button"
id: publish_button
on_press:
- lambda: |-
// Publish 10 values: 1.0, 2.0, ..., 10.0
for (int i = 1; i <= 10; i++) {
id(source_sensor).publish_state(float(i));
}
@@ -1,111 +0,0 @@
esphome:
name: uart-mock-modbus-cli-rw
host:
api:
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
# The actual UART bus used is the uart_mock component below
uart:
baud_rate: 115200
port: /dev/null
# Two virtual buses looped back to each other: the client's transmissions reach the server and the
# server's replies reach the client. auto_start so forwarding is active before the button fires.
uart_mock:
- id: virtual_uart_server
baud_rate: 9600
auto_start: true
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_client
data: !lambda return data;
- id: virtual_uart_client
baud_rate: 9600
auto_start: true
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_server
data: !lambda return data;
globals:
- id: stored_1
type: uint16_t
initial_value: "0"
modbus:
- uart_id: virtual_uart_server
id: virtual_modbus_server
role: server
- uart_id: virtual_uart_client
id: virtual_modbus_client
role: client
turnaround_time: 10ms
modbus_server:
- address: 1
modbus_id: virtual_modbus_server
registers:
# Writable + readable register: the read publishes what it returns, so the test can confirm the
# write half of the 0x17 ran before the read half (Modbus 6.17).
- address: 0x01
value_type: U_WORD
read_lambda: |-
id(srv_read_1).publish_state(id(stored_1));
return id(stored_1);
write_lambda: |-
id(stored_1) = x;
id(srv_write_1).publish_state(x);
return true;
# Read-only register, returned together with 0x01 by the 2-register read half.
- address: 0x02
value_type: U_WORD
read_lambda: return 0x00AA;
sensor:
# Server-side observations.
- platform: template
name: "srv_write_1"
id: srv_write_1
- platform: template
name: "srv_read_1"
id: srv_read_1
# Client-side read-back: the values the client's on_response received.
- platform: template
name: "client_read_0"
id: client_read_0
- platform: template
name: "client_read_1"
id: client_read_1
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
on_press:
# FC 0x17: write reg 0x0001 = 0x1234, then read regs 0x0001..0x0002 back in the same transaction.
- modbus_client.read_write_multiple_registers:
address: 0x01
read_address: 0x0001
read_count: 2
write_address: 0x0001
values: [0x1234]
on_response:
then:
- lambda: |-
// values is the read-back block: reg 0x0001 (must be the just-written 0x1234) and reg 0x0002.
if (values.size() >= 2) {
id(client_read_0).publish_state(values[0]);
id(client_read_1).publish_state(values[1]);
}
@@ -1,88 +0,0 @@
esphome:
name: uart-mock-modbus-custom-pdu
host:
api:
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
# The actual UART bus used is the uart_mock component below
uart:
baud_rate: 115200
port: /dev/null
uart_mock:
- id: virtual_uart_server
baud_rate: 9600
auto_start: true
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_controller
data: !lambda return data;
- id: virtual_uart_controller
baud_rate: 9600
auto_start: true
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_server
data: !lambda return data;
modbus:
- uart_id: virtual_uart_server
id: virtual_modbus_server
role: server
- uart_id: virtual_uart_controller
id: virtual_modbus_controller
role: client
turnaround_time: 10ms
modbus_controller:
- address: 1
modbus_id: virtual_modbus_controller
id: modbus_controller_1
update_interval: 1s
modbus_server:
- address: 1
modbus_id: virtual_modbus_server
id: modbus_server_1
registers:
- address: 0x01
value_type: U_WORD
read_lambda: return 259;
sensor:
# Plain read to confirm the controller <-> server link is up.
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "plain_read"
address: 0x01
register_type: holding
value_type: U_WORD
# Custom PDU: read holding register 0x0001, count 1. The PDU is
# {function code, address hi, address lo, count hi, count lo}; the device
# address and CRC are added by the hub. The lambda parses the response payload
# (the register value, big-endian).
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "custom_read"
custom_pdu: [0x03, 0x00, 0x01, 0x00, 0x01]
lambda: |-
if (data.size() < 2) return {};
return (float) ((data[0] << 8) | data[1]);
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
# This test does not have anything to start (mock is autostart)
@@ -1,106 +0,0 @@
esphome:
name: uart-mock-modbus-dep-buffer
host:
api:
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
# The actual UART bus used is the uart_mock component below
uart:
baud_rate: 115200
port: /dev/null
uart_mock:
- id: virtual_uart_server
baud_rate: 9600
auto_start: true
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_controller
data: !lambda return data;
- id: virtual_uart_controller
baud_rate: 9600
auto_start: true
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_server
data: !lambda return data;
globals:
- id: reg10
type: uint16_t
initial_value: "0"
modbus:
- uart_id: virtual_uart_server
id: virtual_modbus_server
role: server
- uart_id: virtual_uart_controller
id: virtual_modbus_controller
role: client
turnaround_time: 10ms
modbus_controller:
- address: 1
modbus_id: virtual_modbus_controller
id: modbus_controller_1
update_interval: 1s
modbus_server:
- address: 1
modbus_id: virtual_modbus_server
id: modbus_server_1
registers:
- address: 0x10
value_type: U_WORD
read_lambda: return id(reg10);
write_lambda: |-
id(reg10) = x;
return true;
# A number whose write_lambda uses the DEPRECATED buffer parameter (fills `payload` with a legacy raw
# frame as words: device address + function code + data) instead of the new item->write_* API. The write
# must still land with its legacy semantics, and the one-time deprecation warning must fire only once per
# entity no matter how many writes happen.
number:
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "buf_number"
id: buf_number
address: 0x10
register_type: holding
value_type: U_WORD
min_value: 0
max_value: 1000
step: 1
write_lambda: |-
// Legacy raw frame as words: [addr 0x01 | fc 0x06], register 0x0010, value.
payload.push_back(0x0106);
payload.push_back(0x0010);
payload.push_back((uint16_t) x);
return {};
# Reports the server-side register so the test can observe that the deprecated buffer write landed.
sensor:
- platform: template
name: "written_value"
id: written_value
update_interval: 0.5s
lambda: "return id(reg10);"
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
# The test drives the writes via number_command; the mock is autostart.
@@ -1,95 +0,0 @@
esphome:
name: uart-mock-modbus-lambda-invert
host:
api:
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
# The actual UART bus used is the uart_mock component below
uart:
baud_rate: 115200
port: /dev/null
uart_mock:
- id: virtual_uart_server
baud_rate: 9600
auto_start: true
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_controller
data: !lambda return data;
- id: virtual_uart_controller
baud_rate: 9600
auto_start: true
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_server
data: !lambda return data;
globals:
- id: reg40
type: uint16_t
initial_value: "5"
modbus:
- uart_id: virtual_uart_server
id: virtual_modbus_server
role: server
- uart_id: virtual_uart_controller
id: virtual_modbus_controller
role: client
turnaround_time: 10ms
modbus_controller:
- address: 1
modbus_id: virtual_modbus_controller
id: modbus_controller_1
update_interval: 1s
modbus_server:
- address: 1
modbus_id: virtual_modbus_server
id: modbus_server_1
registers:
- address: 0x40
value_type: U_WORD
read_lambda: return id(reg40);
write_lambda: id(reg40) = x; return true;
# An active-low holding switch: the write_lambda inverts the wire value, but the entity must still
# report the REQUESTED state. assumed_state keeps the register unpolled, so the published state comes
# only from write_state() - turning ON writes 0x0000 yet the switch shows ON.
switch:
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "invert_switch"
register_type: holding
address: 0x40
assumed_state: true
write_lambda: |-
return !x;
sensor:
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_40"
address: 0x40
register_type: holding
value_type: U_WORD
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
# This test does not have anything to start (mock is autostart)
@@ -1,97 +0,0 @@
esphome:
name: uart-mock-modbus-lambda-write
host:
api:
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
# The actual UART bus used is the uart_mock component below
uart:
baud_rate: 115200
port: /dev/null
uart_mock:
- id: virtual_uart_server
baud_rate: 9600
auto_start: true
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_controller
data: !lambda return data;
- id: virtual_uart_controller
baud_rate: 9600
auto_start: true
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_server
data: !lambda return data;
globals:
- id: reg30
type: uint16_t
initial_value: "0"
modbus:
- uart_id: virtual_uart_server
id: virtual_modbus_server
role: server
- uart_id: virtual_uart_controller
id: virtual_modbus_controller
role: client
turnaround_time: 10ms
modbus_controller:
- address: 1
modbus_id: virtual_modbus_controller
id: modbus_controller_1
update_interval: 1s
modbus_server:
- address: 1
modbus_id: virtual_modbus_server
id: modbus_server_1
registers:
- address: 0x30
value_type: U_WORD
read_lambda: return id(reg30);
write_lambda: id(reg30) = x; return true;
# A COIL-type switch (assumed_state, write-only) whose write_lambda ignores its own coil type and instead
# drives a HOLDING-REGISTER write on the mock server through the entity itself: `item` IS the command, so
# item->write_single_register() sends a register write from a coil entity (cross-type). Returning nothing
# (an empty optional) tells the write path the lambda already dispatched the frame - no default coil write.
switch:
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "cross_switch"
register_type: coil
address: 0x00
assumed_state: true
write_lambda: |-
item->write_single_register(0x30, x ? 1234 : 0);
return {};
sensor:
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_30"
address: 0x30
register_type: holding
value_type: U_WORD
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
# This test does not have anything to start (mock is autostart)
@@ -0,0 +1,233 @@
esphome:
name: uart-mock-modbus-loopback
host:
api:
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
# The actual UART bus used is the uart_mock component below
uart:
baud_rate: 115200
port: /dev/null
# Shared loopback fixture (see the shared_yaml markers in the test file);
# register spaces are disjoint so each test only observes its own entities.
uart_mock:
- id: virtual_uart_server
baud_rate: 9600
auto_start: true
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_controller
data: !lambda return data;
- id: virtual_uart_controller
baud_rate: 9600
auto_start: true
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_server
data: !lambda return data;
globals:
- id: reg10
type: uint16_t
initial_value: "100"
- id: reg11
type: uint16_t
initial_value: "200"
- id: reg12
type: uint16_t
initial_value: "300"
- id: reg13
type: uint16_t
initial_value: "0xABCD"
- id: reg30
type: uint16_t
initial_value: "0"
- id: reg40
type: uint16_t
initial_value: "5"
- id: reg50
type: uint16_t
initial_value: "0"
modbus:
- uart_id: virtual_uart_server
id: virtual_modbus_server
role: server
- uart_id: virtual_uart_controller
id: virtual_modbus_controller
role: client
turnaround_time: 10ms
modbus_controller:
- address: 1
modbus_id: virtual_modbus_controller
id: modbus_controller_1
update_interval: 1s
modbus_server:
- address: 1
modbus_id: virtual_modbus_server
registers:
- address: 0x01
value_type: U_WORD
read_lambda: return 259;
- address: 0x10
value_type: U_WORD
read_lambda: return id(reg10);
write_lambda: id(reg10) = x; return true;
- address: 0x11
value_type: U_WORD
read_lambda: return id(reg11);
write_lambda: id(reg11) = x; return true;
- address: 0x12
value_type: U_WORD
read_lambda: return id(reg12);
write_lambda: id(reg12) = x; return true;
- address: 0x13
value_type: U_WORD
read_lambda: return id(reg13);
- address: 0x30
value_type: U_WORD
read_lambda: return id(reg30);
write_lambda: id(reg30) = x; return true;
- address: 0x40
value_type: U_WORD
read_lambda: return id(reg40);
write_lambda: id(reg40) = x; return true;
- address: 0x50
value_type: U_WORD
read_lambda: return id(reg50);
write_lambda: id(reg50) = x; return true;
# Byte-based offset: 2 bytes -> register 0x11 (the old code folded it in as a
# register count, hitting 0x12). assumed_state keeps the switch write-only.
switch:
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "offset_switch"
register_type: holding
address: 0x10
offset: 2
assumed_state: true
# Reading switch, byte offset 6 -> register 0x13; the pre-fix resolution (0x16)
# would draw ILLEGAL_DATA_ADDRESS and never publish.
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "read_offset_switch"
register_type: holding
address: 0x10
offset: 6
bitmask: 0x1
# Coil switch whose write_lambda dispatches a holding-register write via `item`;
# returning an empty optional suppresses the default coil write.
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "cross_switch"
register_type: coil
address: 0x00
assumed_state: true
write_lambda: |-
item->write_single_register(0x30, x ? 1234 : 0);
return {};
# Active-low: the write_lambda inverts the wire value but the entity must still
# report the requested state (assumed_state keeps the register unpolled).
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "invert_switch"
register_type: holding
address: 0x40
assumed_state: true
write_lambda: |-
return !x;
# Uses the deprecated buffer parameter (legacy raw frame as words); the write
# must land and the deprecation warning must fire only once per entity.
number:
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "buf_number"
id: buf_number
address: 0x50
register_type: holding
value_type: U_WORD
min_value: 0
max_value: 1000
step: 1
write_lambda: |-
// Legacy raw frame as words: [addr 0x01 | fc 0x06], register 0x0050, value.
payload.push_back(0x0106);
payload.push_back(0x0050);
payload.push_back((uint16_t) x);
return {};
sensor:
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "plain_read"
address: 0x01
register_type: holding
value_type: U_WORD
# Custom PDU: read holding register 0x0001; device address and CRC are added
# by the hub. The lambda parses the big-endian register value.
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "custom_read"
custom_pdu: [0x03, 0x00, 0x01, 0x00, 0x01]
lambda: |-
if (data.size() < 2) return {};
return (float) ((data[0] << 8) | data[1]);
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_10"
address: 0x10
register_type: holding
value_type: U_WORD
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_11"
address: 0x11
register_type: holding
value_type: U_WORD
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_12"
address: 0x12
register_type: holding
value_type: U_WORD
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_30"
address: 0x30
register_type: holding
value_type: U_WORD
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_40"
address: 0x40
register_type: holding
value_type: U_WORD
# Reports the server-side register so the test can observe that the deprecated buffer write landed.
- platform: template
name: "written_value"
id: written_value
update_interval: 0.5s
lambda: "return id(reg50);"
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
# Nothing to start (mock is autostart); tests drive entities directly
@@ -1,5 +1,5 @@
esphome:
name: uart-mock-modbus-server-contro
name: uart-mock-modbus-mesh
host:
api:
@@ -17,13 +17,14 @@ uart:
baud_rate: 115200
port: /dev/null
# Shared 3-bus mesh (see the shared_yaml markers): addr 1 = typed read-only
# registers, addr 5 = the read/write 0x17 target, addr 2/3 on the second
# server hub. auto_start everywhere: the controller polls at boot, so the
# forwarding must already be live or early requests generate warnings.
# Every test presses Start Scenario, so all merged actions fire in every test.
uart_mock:
- id: virtual_uart_server
baud_rate: 9600
# auto_start must be true for loopback fixtures: the modbus controller
# polls on its update_interval immediately at boot, so the uart_mock
# forwarding must already be active or early requests are lost and
# generate modbus warnings.
auto_start: true
debug:
on_tx:
@@ -31,35 +32,68 @@ uart_mock:
- uart_mock.inject_rx:
id: virtual_uart_controller
data: !lambda return data;
- id: virtual_uart_controller
- uart_mock.inject_rx:
id: virtual_uart_server_2
data: !lambda return data;
- id: virtual_uart_server_2
baud_rate: 9600
auto_start: true # See comment on virtual_uart_server above
auto_start: true
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_server
data: !lambda return data;
- uart_mock.inject_rx:
id: virtual_uart_controller
data: !lambda return data;
- id: virtual_uart_controller
baud_rate: 9600
auto_start: true
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_server
data: !lambda return data;
- uart_mock.inject_rx:
id: virtual_uart_server_2
data: !lambda return data;
globals:
- id: stored_1
type: uint16_t
initial_value: "0"
modbus:
- uart_id: virtual_uart_server
id: virtual_modbus_server
role: server
- uart_id: virtual_uart_server_2
id: virtual_modbus_server_2
role: server
- uart_id: virtual_uart_controller
id: virtual_modbus_controller
id: virtual_modbus_client
role: client
turnaround_time: 10ms
modbus_controller:
- address: 1
modbus_id: virtual_modbus_controller
modbus_id: virtual_modbus_client
id: modbus_controller_1
update_interval: 1s
- address: 2
modbus_id: virtual_modbus_client
id: modbus_controller_2
update_interval: 1s
- address: 3
modbus_id: virtual_modbus_client
id: modbus_controller_3
update_interval: 1s
modbus_server:
- address: 1
modbus_id: virtual_modbus_server
id: modbus_server_1
registers:
- address: 0x01
value_type: U_WORD
@@ -103,6 +137,34 @@ modbus_server:
- address: 0x28
value_type: FP32_R
read_lambda: return 3.14;
- address: 5
modbus_id: virtual_modbus_server
registers:
# Writable + readable register: srv_write_1 plus the client's read-back
# confirm the write half of the 0x17 ran before the read half (Modbus 6.17).
- address: 0x01
value_type: U_WORD
read_lambda: return id(stored_1);
write_lambda: |-
id(stored_1) = x;
id(srv_write_1).publish_state(x);
return true;
# Read-only register, returned together with 0x01 by the 2-register read half.
- address: 0x02
value_type: U_WORD
read_lambda: return 0x00AA;
- address: 2
modbus_id: virtual_modbus_server_2
registers:
- address: 0x01
value_type: U_WORD
read_lambda: return 919;
- address: 3
modbus_id: virtual_modbus_server_2
registers:
- address: 0x01
value_type: U_WORD
read_lambda: return 929;
sensor:
- platform: modbus_controller
@@ -195,9 +257,46 @@ sensor:
address: 0x28
register_type: holding
value_type: FP32_R
- platform: modbus_controller
modbus_controller_id: modbus_controller_2
name: "multi_reg_a"
address: 0x01
register_type: holding
value_type: U_WORD
- platform: modbus_controller
modbus_controller_id: modbus_controller_3
name: "multi_reg_b"
address: 0x01
register_type: holding
value_type: U_WORD
# client_read_write observations, server- and client-side.
- platform: template
name: "srv_write_1"
id: srv_write_1
- platform: template
name: "client_read_0"
id: client_read_0
- platform: template
name: "client_read_1"
id: client_read_1
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
# This test does not have anything to start (mock is autostart)
on_press:
# FC 0x17: write reg 0x0001 = 0x1234, then read regs 0x0001..0x0002 back in the same transaction.
- modbus_client.read_write_multiple_registers:
address: 5
read_address: 0x0001
read_count: 2
write_address: 0x0001
values: [0x1234]
on_response:
then:
- lambda: |-
// values is the read-back block: reg 0x0001 (must be the just-written 0x1234) and reg 0x0002.
if (values.size() >= 2) {
id(client_read_0).publish_state(values[0]);
id(client_read_1).publish_state(values[1]);
}
@@ -1,138 +0,0 @@
esphome:
name: uart-mock-modbus-reg-offset
host:
api:
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
# The actual UART bus used is the uart_mock component below
uart:
baud_rate: 115200
port: /dev/null
uart_mock:
- id: virtual_uart_server
baud_rate: 9600
auto_start: true
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_controller
data: !lambda return data;
- id: virtual_uart_controller
baud_rate: 9600
auto_start: true
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_server
data: !lambda return data;
globals:
- id: reg10
type: uint16_t
initial_value: "100"
- id: reg11
type: uint16_t
initial_value: "200"
- id: reg12
type: uint16_t
initial_value: "300"
- id: reg13
type: uint16_t
initial_value: "0xABCD"
modbus:
- uart_id: virtual_uart_server
id: virtual_modbus_server
role: server
- uart_id: virtual_uart_controller
id: virtual_modbus_controller
role: client
turnaround_time: 10ms
modbus_controller:
- address: 1
modbus_id: virtual_modbus_controller
id: modbus_controller_1
update_interval: 1s
modbus_server:
- address: 1
modbus_id: virtual_modbus_server
id: modbus_server_1
registers:
- address: 0x10
value_type: U_WORD
read_lambda: return id(reg10);
write_lambda: id(reg10) = x; return true;
- address: 0x11
value_type: U_WORD
read_lambda: return id(reg11);
write_lambda: id(reg11) = x; return true;
- address: 0x12
value_type: U_WORD
read_lambda: return id(reg12);
write_lambda: id(reg12) = x; return true;
- address: 0x13
value_type: U_WORD
read_lambda: return id(reg13);
write_lambda: id(reg13) = x; return true;
# A holding-register switch at 0x10 with a 2-BYTE offset. offset is byte-based, so the write must target
# register 0x10 + 2/2 = 0x11. The old (pre-fix) behavior folded offset into the address as a register
# count, hitting 0x12 instead. assumed_state keeps the switch write-only so it does not read any register.
switch:
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "offset_switch"
register_type: holding
address: 0x10
offset: 2
assumed_state: true
# A holding-register switch that READS its state. Byte offset 6 -> register 0x10 + 6/2 = 0x13. Post-fix
# the switch itself resolves to 0x13 (the even byte offset folds into the address as whole registers) and
# joins the 0x10..0x13 range, so no separate 0x13 sensor is needed. Pre-fix the whole byte offset folds
# into the address (0x16), where the server answers ILLEGAL_DATA_ADDRESS and the switch never publishes.
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "read_offset_switch"
register_type: holding
address: 0x10
offset: 6
bitmask: 0x1
sensor:
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_10"
address: 0x10
register_type: holding
value_type: U_WORD
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_11"
address: 0x11
register_type: holding
value_type: U_WORD
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_12"
address: 0x12
register_type: holding
value_type: U_WORD
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
# This test does not have anything to start (mock is autostart)
@@ -1,124 +0,0 @@
esphome:
name: uart-mock-modbus-server-test
host:
api:
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
# The actual UART bus used is the uart_mock component below
uart:
baud_rate: 115200
port: /dev/null
uart_mock:
- id: virtual_uart_dev
baud_rate: 9600
rx_full_threshold: 120
rx_timeout: 2
auto_start: false
debug:
injections:
- delay: 100ms
inject_rx: [0x01, 0x03, 0x00, 0x03, 0x00, 0x01, 0x74, 0x0A] # Read holding register 3 on device 1 (basic_read)
- delay: 100ms
# Read holding register 7 on device 2
# Reply from device 2
# Read holding register 5 on device 1 (read_after_peer_response)
inject_rx:
[
0x02,
0x03,
0x00,
0x07,
0x00,
0x01,
0x35,
0xF8,
0x02,
0x03,
0x02,
0x00,
0xF0,
0xFC,
0x00,
0x01,
0x03,
0x00,
0x05,
0x00,
0x01,
0x94,
0x0B,
]
- delay: 100ms
inject_rx: [0x02, 0x03, 0x00, 0x07, 0x00, 0x01, 0x35, 0xF8] # Read holding register 7 on device 2, with no response
- delay: 100ms
# Read holding register 7 on device 2, with no response
# Read holding register A on device 1 (read_after_peer_timeout)
inject_rx:
[
0x02,
0x03,
0x00,
0x07,
0x00,
0x01,
0x35,
0xF8,
0x01,
0x03,
0x00,
0x0A,
0x00,
0x01,
0xA4,
0x08,
]
modbus:
uart_id: virtual_uart_dev
role: server
modbus_server:
- address: 1
registers:
- address: 0x03
value_type: U_WORD
read_lambda: |-
id(basic_read).publish_state(1);
return 1;
- address: 0x05
value_type: U_WORD
read_lambda: |-
id(read_after_peer_response).publish_state(1);
return 1;
- address: 0x0A
value_type: U_WORD
read_lambda: |-
id(read_after_peer_timeout).publish_state(1);
return 1;
sensor:
- platform: template
name: "basic_read"
id: basic_read
- platform: template
name: "read_after_peer_response"
id: read_after_peer_response
- platform: template
name: "read_after_peer_timeout"
id: read_after_peer_timeout
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
on_press:
- lambda: "id(virtual_uart_dev).start_scenario();"
@@ -1,116 +0,0 @@
esphome:
name: uart-mock-modbus-server-mult
host:
api:
logger:
level: VERBOSE
external_components:
- source:
type: local
path: EXTERNAL_COMPONENT_PATH
# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
# The actual UART bus used is the uart_mock component below
uart:
baud_rate: 115200
port: /dev/null
uart_mock:
- id: virtual_uart_server
baud_rate: 9600
# auto_start must be true for loopback fixtures: the modbus controller
# polls on its update_interval immediately at boot, so the uart_mock
# forwarding must already be active or early requests are lost and
# generate modbus warnings.
auto_start: true
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_controller
data: !lambda return data;
- uart_mock.inject_rx:
id: virtual_uart_server_2
data: !lambda return data;
- id: virtual_uart_server_2
baud_rate: 9600
auto_start: true # See comment on virtual_uart_server above
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_server
data: !lambda return data;
- uart_mock.inject_rx:
id: virtual_uart_controller
data: !lambda return data;
- id: virtual_uart_controller
baud_rate: 9600
auto_start: true # See comment on virtual_uart_server above
debug:
on_tx:
- then:
- uart_mock.inject_rx:
id: virtual_uart_server
data: !lambda return data;
- uart_mock.inject_rx:
id: virtual_uart_server_2
data: !lambda return data;
modbus:
- uart_id: virtual_uart_server
id: virtual_modbus_server
role: server
- uart_id: virtual_uart_server_2
id: virtual_modbus_server_2
role: server
- uart_id: virtual_uart_controller
id: virtual_modbus_client
role: client
turnaround_time: 10ms
modbus_controller:
- address: 1
modbus_id: virtual_modbus_client
update_interval: 1s
id: modbus_controller_1
- address: 2
modbus_id: virtual_modbus_client
update_interval: 1s
id: modbus_controller_2
modbus_server:
- address: 1
modbus_id: virtual_modbus_server
registers:
- address: 0x01
value_type: U_WORD
read_lambda: return 919;
- address: 2
modbus_id: virtual_modbus_server_2
registers:
- address: 0x01
value_type: U_WORD
read_lambda: return 929;
sensor:
- platform: modbus_controller
modbus_controller_id: modbus_controller_1
name: "reg_u_word"
address: 0x01
register_type: holding
value_type: U_WORD
- platform: modbus_controller
modbus_controller_id: modbus_controller_2
name: "reg_u_word_2"
address: 0x01
register_type: holding
value_type: U_WORD
button:
- platform: template
name: "Start Scenario"
id: start_scenario_btn
# This test does not have anything to start (mock is autostart)
@@ -1,5 +1,5 @@
esphome:
name: uart-mock-modbus-srv-rw
name: uart-mock-modbus-srv-injected
host:
api:
@@ -17,6 +17,8 @@ uart:
baud_rate: 115200
port: /dev/null
# Shared server-role fixture (see the shared_yaml markers in the test file);
# the injections concatenate and each test waits only on its own sensors.
uart_mock:
- id: virtual_uart_dev
baud_rate: 9600
@@ -25,18 +27,31 @@ uart_mock:
auto_start: false
debug:
injections:
# FC 0x17 Read/Write Multiple Registers on device 1:
# write reg 0x0001 = 0x1234 (qty 1), then read regs 0x0001..0x0002 (qty 2).
# Per Modbus 6.17 the write is performed before the read, so reg 0x0001 must
# read back the just-written 0x1234 in the same request.
- delay: 100ms
inject_rx: [0x01, 0x03, 0x00, 0x03, 0x00, 0x01, 0x74, 0x0A] # Read holding register 3 on device 1 (basic_read)
- delay: 100ms
# Read holding register 7 on device 2, its reply, then read holding
# register 5 on device 1 (read_after_peer_response)
inject_rx: [0x02, 0x03, 0x00, 0x07, 0x00, 0x01, 0x35, 0xF8,
0x02, 0x03, 0x02, 0x00, 0xF0, 0xFC,
0x00, 0x01, 0x03, 0x00, 0x05, 0x00, 0x01, 0x94, 0x0B]
- delay: 100ms
inject_rx: [0x02, 0x03, 0x00, 0x07, 0x00, 0x01, 0x35, 0xF8] # Read holding register 7 on device 2, with no response
- delay: 100ms
# Read holding register 7 on device 2 with no response, then read
# holding register A on device 1 (read_after_peer_timeout)
inject_rx: [0x02, 0x03, 0x00, 0x07, 0x00, 0x01, 0x35, 0xF8,
0x01, 0x03, 0x00, 0x0A, 0x00, 0x01, 0xA4, 0x08]
# FC 0x17 on device 1: write reg 0x0001 = 0x1234 then read 0x0001..0x0002;
# per Modbus 6.17 the write runs first, so 0x0001 must read back 0x1234.
- delay: 100ms
inject_rx:
[0x01, 0x17, 0x00, 0x01, 0x00, 0x02, 0x00, 0x01, 0x00, 0x01, 0x02, 0x12, 0x34, 0x49, 0xD8]
# FC 0x17: write reg 0x0003 = 0x5678 (qty 1), then read reg 0x0003 (qty 1) -
# FC 0x17: write reg 0x0006 = 0x5678 (qty 1), then read reg 0x0006 (qty 1) -
# a write and read targeting a different register block.
- delay: 100ms
inject_rx:
[0x01, 0x17, 0x00, 0x03, 0x00, 0x01, 0x00, 0x03, 0x00, 0x01, 0x02, 0x56, 0x78, 0x9B, 0x10]
[0x01, 0x17, 0x00, 0x06, 0x00, 0x01, 0x00, 0x06, 0x00, 0x01, 0x02, 0x56, 0x78, 0x8B, 0x55]
globals:
- id: stored_1
@@ -70,8 +85,18 @@ modbus_server:
read_lambda: |-
id(rw_read_2).publish_state(0x00AA);
return 0x00AA;
# Second writable + readable register, targeted by the second request.
- address: 0x03
value_type: U_WORD
read_lambda: |-
id(basic_read).publish_state(1);
return 1;
- address: 0x05
value_type: U_WORD
read_lambda: |-
id(read_after_peer_response).publish_state(1);
return 1;
# Second writable + readable register, targeted by the second FC 0x17 request.
- address: 0x06
value_type: U_WORD
read_lambda: |-
id(rw_read_3).publish_state(id(stored_3));
@@ -80,8 +105,22 @@ modbus_server:
id(stored_3) = x;
id(rw_write_3).publish_state(x);
return true;
- address: 0x0A
value_type: U_WORD
read_lambda: |-
id(read_after_peer_timeout).publish_state(1);
return 1;
sensor:
- platform: template
name: "basic_read"
id: basic_read
- platform: template
name: "read_after_peer_response"
id: read_after_peer_response
- platform: template
name: "read_after_peer_timeout"
id: read_after_peer_timeout
- platform: template
name: "rw_write_1"
id: rw_write_1
+11 -3
View File
@@ -1,7 +1,7 @@
"""Helpers for manipulating the host platform's preferences file.
ESPHome's host platform stores preferences in
``~/.esphome/prefs/<app_name>.prefs`` using a simple binary layout that
``$ESPHOME_PREFDIR/<app_name>.prefs`` using a simple binary layout that
mirrors ``HostPreferences::sync()``:
``[uint32_t key][uint8_t len][uint8_t data[len]]`` per entry.
@@ -11,13 +11,21 @@ boot (e.g. forcing safe mode) or to clear stale state between runs.
from __future__ import annotations
import os
from pathlib import Path
import struct
def host_prefs_path(device_name: str) -> Path:
"""Return the on-disk prefs file path for a host-platform device."""
return Path.home() / ".esphome" / "prefs" / f"{device_name}.prefs"
"""Return the on-disk prefs file path for a host-platform device.
Requires ESPHOME_PREFDIR, which the autouse isolated_preferences fixture
sets; refusing the ~/.esphome/prefs fallback keeps tests off real user
data if the fixture is ever bypassed."""
prefdir = os.environ.get("ESPHOME_PREFDIR")
if not prefdir:
raise RuntimeError("ESPHOME_PREFDIR is not set; refusing the real prefs dir")
return Path(prefdir) / f"{device_name}.prefs"
def clear_host_prefs(device_name: str) -> None:
@@ -0,0 +1,175 @@
"""Integration tests for the api outgoing_connection option.
The device dials out to the test's listener when no dial-back target client is
connected. The listener plays the Home Assistant side over the accepted socket
using aioesphomeapi's sans-IO Noise handshake: the device sends its server
hello first so the listener could pick the right key, and the NNpsk0 handshake
then verifies both sides. Protocol roles stay unchanged, so the client speaks
exactly the same frames as over a normal connection. A client becomes the
remembered dial-back target by setting the outgoing_connection_target flag in
its hello.
"""
from __future__ import annotations
import asyncio
import socket
from typing import Any
from aioesphomeapi import api_pb2
import pytest
from .raw_api_client import MESSAGE_TYPE_OF
from .types import RunCompiledFunction
KEY = "bOFFzzvfpg5DB94DuBGLXD/hMnhpDKgP9UQyBulwWVU="
DEVICE_NAME = "outgoing-conn-test"
HA_CLIENT_INFO = "Home Assistant 2026.8.0"
# HelloRequest field 4 (outgoing_connection_target) as raw protobuf bytes; the
# installed aioesphomeapi's api_pb2 predates the field, so append it manually.
HELLO_TARGET_FLAG = b"\x20\x01"
def _frame(payload: bytes) -> bytes:
return bytes((0x01, len(payload) >> 8, len(payload) & 0xFF)) + payload
async def _read_frame(reader: asyncio.StreamReader, timeout: float = 10.0) -> bytes:
header = await asyncio.wait_for(reader.readexactly(3), timeout)
assert header[0] == 0x01, f"Bad frame indicator: {header[0]}"
return await asyncio.wait_for(
reader.readexactly((header[1] << 8) | header[2]), timeout
)
def _check_server_hello(server_hello: bytes) -> None:
assert server_hello[0] == 0x01, "Bad chosen proto in server hello"
name, mac, _rest = server_hello[1:].split(b"\x00", 2)
assert name.decode() == DEVICE_NAME
assert len(mac) == 12, f"Expected bare MAC, got {mac!r}"
async def _run_ha_session(
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
*,
device_dialed_out: bool,
) -> None:
"""Handshake and exchange the usual first messages as Home Assistant would."""
# Lazy import per the module's own contract (pulls in the noise stack)
from aioesphomeapi.noise import NoiseHandshake
if device_dialed_out:
# On an outgoing connection the device announces itself first so the
# peer can pick the matching key before its PSK-mixed first message.
_check_server_hello(await _read_frame(reader))
handshake = NoiseHandshake(KEY, b"NoiseAPIInit\x00\x00")
writer.write(b"\x01\x00\x00" + _frame(b"\x00" + handshake.write_message()))
await writer.drain()
if not device_dialed_out:
_check_server_hello(await _read_frame(reader))
reply = await _read_frame(reader)
assert reply[0] == 0, f"Handshake rejected: {reply[1:].decode(errors='replace')}"
handshake.read_message(reply[1:])
encrypt_cipher, decrypt_cipher = handshake.get_ciphers()
async def transact(msg: Any, response_cls: Any, extra_payload: bytes = b"") -> Any:
msg_type = MESSAGE_TYPE_OF[type(msg)]
payload = msg.SerializeToString() + extra_payload
plaintext = (
bytes(
(msg_type >> 8, msg_type & 0xFF, len(payload) >> 8, len(payload) & 0xFF)
)
+ payload
)
writer.write(_frame(encrypt_cipher.encrypt(plaintext)))
await writer.drain()
want = MESSAGE_TYPE_OF[response_cls]
while True:
plain = decrypt_cipher.decrypt(await _read_frame(reader))
if ((plain[0] << 8) | plain[1]) == want:
response = response_cls()
response.ParseFromString(bytes(plain[4:]))
return response
# Declare this client a dial-back target in the hello
await transact(
api_pb2.HelloRequest(client_info=HA_CLIENT_INFO),
api_pb2.HelloResponse,
extra_payload=HELLO_TARGET_FLAG,
)
device_info = await transact(
api_pb2.DeviceInfoRequest(), api_pb2.DeviceInfoResponse
)
assert device_info.name == DEVICE_NAME
async def _serve_home_assistant(listener: socket.socket) -> None:
"""Accept one dial-in from the device and run the client side over it."""
loop = asyncio.get_running_loop()
conn, _ = await asyncio.wait_for(loop.sock_accept(listener), timeout=30)
reader, writer = await asyncio.open_connection(sock=conn)
try:
await _run_ha_session(reader, writer, device_dialed_out=True)
finally:
writer.close()
@pytest.mark.asyncio
async def test_api_outgoing_connection(
yaml_config: str,
run_compiled: RunCompiledFunction,
) -> None:
"""With a configured host the device dials out and speaks the normal API."""
listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
listener.bind(("127.0.0.1", 0))
listener.listen(2)
listener.setblocking(False)
port = listener.getsockname()[1]
try:
yaml = yaml_config.replace("OUTGOING_PORT", str(port))
async with run_compiled(yaml):
await _serve_home_assistant(listener)
finally:
listener.close()
@pytest.mark.asyncio
async def test_api_outgoing_connection_remembered(
yaml_config: str,
run_compiled: RunCompiledFunction,
unused_tcp_port: int,
) -> None:
"""No host configured: the device remembers the client whose hello carried
the dial-back flag and dials that address after a restart."""
listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# Bound but not yet listening so first-phase dials cannot queue stale
# connections; whether the device attempts any dial before the restart
# is timing dependent and not asserted here.
listener.bind(("127.0.0.1", 0))
port = listener.getsockname()[1]
try:
yaml = yaml_config.replace("OUTGOING_PORT", str(port))
async with run_compiled(yaml):
# Connect inbound with the dial-back flag; the device persists the
# peer address during the hello.
reader, writer = await asyncio.open_connection("127.0.0.1", unused_tcp_port)
try:
await _run_ha_session(reader, writer, device_dialed_out=False)
finally:
writer.close()
# Restart with the same preferences: the device now dials the
# remembered address on its own.
listener.listen(2)
listener.setblocking(False)
async with run_compiled(yaml):
await _serve_home_assistant(listener)
finally:
listener.close()
+31 -2
View File
@@ -5,7 +5,7 @@ import re
import pytest
from .types import RunCompiledFunction
from .types import APIClientConnectedFactory, RunCompiledFunction
@pytest.mark.asyncio
@@ -16,7 +16,9 @@ async def test_api_reboot_timeout(
"""Test that the device reboots when no API clients connect within the timeout."""
loop = asyncio.get_running_loop()
reboot_future = loop.create_future()
reboot_pattern = re.compile(r"No clients; rebooting")
# The harness port probe always connects without authenticating, so the
# reboot deterministically reports the unauthenticated form
reboot_pattern = re.compile(r"none authenticated; rebooting")
def check_output(line: str) -> None:
"""Check output for reboot message."""
@@ -33,3 +35,30 @@ async def test_api_reboot_timeout(
pytest.fail("Device did not reboot within expected timeout")
# Test passes if we get here - reboot was detected
@pytest.mark.asyncio
async def test_api_reboot_timeout_after_authenticated_disconnect(
yaml_config: str,
run_compiled: RunCompiledFunction,
api_client_connected: APIClientConnectedFactory,
) -> None:
"""An authenticated disconnect resets the flag; the clean branch reboots."""
loop = asyncio.get_running_loop()
reboot_future = loop.create_future()
reboot_pattern = re.compile(r"No clients; rebooting")
def check_output(line: str) -> None:
"""Check output for reboot message."""
if not reboot_future.done() and reboot_pattern.search(line):
reboot_future.set_result(True)
async with run_compiled(yaml_config, line_callback=check_output):
# An authenticated session refreshes the watchdog and clears the
# unauthenticated flag the harness probe set
async with api_client_connected() as client:
await client.device_info()
try:
await asyncio.wait_for(reboot_future, timeout=5.0)
except TimeoutError:
pytest.fail("Device did not reboot within expected timeout")
@@ -24,7 +24,6 @@ from .types import (
RunCompiledFunction,
)
pytestmark = pytest.mark.usefixtures("isolated_preferences")
NEW_KEY = PROVISIONING_PSK
@@ -41,15 +41,6 @@ async def _poll_until_exists(path: Path) -> None:
await asyncio.sleep(0.05)
@pytest.fixture(autouse=True)
def isolated_preferences(monkeypatch: pytest.MonkeyPatch, tmp_path) -> Path:
"""Keep host preferences per-test so this test never touches the real
~/.esphome/prefs and never races other tests over ESPHOME_PREFDIR."""
prefdir = tmp_path / "prefs"
monkeypatch.setenv("ESPHOME_PREFDIR", str(prefdir))
return prefdir / f"{DEVICE_NAME}.prefs"
@pytest.mark.asyncio
async def test_host_preferences_suspend_resume(
yaml_config: str,
@@ -58,7 +49,7 @@ async def test_host_preferences_suspend_resume(
isolated_preferences: Path,
) -> None:
"""Test that a running syncer flushes, a suspended one doesn't, and resume restores flushing."""
pref_file = isolated_preferences
pref_file = isolated_preferences / f"{DEVICE_NAME}.prefs"
loop = asyncio.get_running_loop()
saved_in_memory = loop.create_future()
@@ -11,14 +11,6 @@ from .state_utils import InitialStateHelper, require_entity
from .types import APIClientConnectedFactory, RunCompiledFunction
@pytest.fixture(autouse=True)
def isolated_preferences(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None:
"""Keep host preferences per-test so RESTORE_AND_ON never loads a stale value left
behind by a previous run (host preferences otherwise persist to ~/.esphome/prefs,
keyed only by device name)."""
monkeypatch.setenv("ESPHOME_PREFDIR", str(tmp_path / "prefs"))
@pytest.mark.asyncio
async def test_light_initial_state(
yaml_config: str,
+16 -7
View File
@@ -173,6 +173,7 @@ async def test_uart_mock_modbus_no_threshold(
_assert_no_modbus_errors(error_log_lines, warning_log_lines)
@pytest.mark.shared_yaml("uart_mock_modbus_server_injected")
@pytest.mark.asyncio
async def test_uart_mock_modbus_server(
yaml_config: str,
@@ -203,6 +204,7 @@ async def test_uart_mock_modbus_server(
_assert_no_modbus_errors(error_log_lines, warning_log_lines)
@pytest.mark.shared_yaml("uart_mock_modbus_server_injected")
@pytest.mark.asyncio
async def test_uart_mock_modbus_server_read_write(
yaml_config: str,
@@ -231,8 +233,8 @@ async def test_uart_mock_modbus_server_read_write(
"rw_write_1": 4660, # 0x1234 written to reg 0x0001
"rw_read_1": 4660, # reg 0x0001 reads back the just-written value
"rw_read_2": 170, # 0x00AA read from reg 0x0002 in the same request
"rw_write_3": 22136, # 0x5678 written to reg 0x0003
"rw_read_3": 22136, # reg 0x0003 reads back the just-written value
"rw_write_3": 22136, # 0x5678 written to reg 0x0006
"rw_read_3": 22136, # reg 0x0006 reads back the just-written value
}
)
@@ -241,7 +243,8 @@ async def test_uart_mock_modbus_server_read_write(
api_client_connected() as client,
):
await tracker.setup_and_start_scenario(client)
await tracker.await_all(futures)
# The FC 0x17 injections fire last, behind four earlier 100ms delays
await tracker.await_all(futures, timeout=4.0)
_assert_no_modbus_errors(error_log_lines, warning_log_lines)
@@ -296,6 +299,7 @@ async def test_uart_mock_modbus_server_read_write_invalid(
)
@pytest.mark.shared_yaml("uart_mock_modbus_mesh")
@pytest.mark.asyncio
async def test_uart_mock_modbus_server_controller(
yaml_config: str,
@@ -485,6 +489,7 @@ async def test_uart_mock_modbus_server_controller_bits(
_assert_no_modbus_errors(error_log_lines, warning_log_lines)
@pytest.mark.shared_yaml("uart_mock_modbus_mesh")
@pytest.mark.asyncio
async def test_uart_mock_modbus_server_controller_multiple(
yaml_config: str,
@@ -495,7 +500,7 @@ async def test_uart_mock_modbus_server_controller_multiple(
line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
expected_values = {"reg_u_word": 919, "reg_u_word_2": 929}
expected_values = {"multi_reg_a": 919, "multi_reg_b": 929}
tracker = SensorTracker(list(expected_values.keys()))
futures = tracker.expect_all(expected_values)
@@ -706,6 +711,7 @@ async def test_uart_mock_modbus_shared_address(
_assert_no_modbus_errors(error_log_lines, warning_log_lines)
@pytest.mark.shared_yaml("uart_mock_modbus_loopback")
@pytest.mark.asyncio
async def test_uart_mock_modbus_custom_pdu(
yaml_config: str,
@@ -932,6 +938,7 @@ async def test_uart_mock_modbus_broadcast_write(
_assert_no_modbus_errors(error_log_lines, warning_log_lines)
@pytest.mark.shared_yaml("uart_mock_modbus_mesh")
@pytest.mark.asyncio
async def test_uart_mock_modbus_client_read_write(
yaml_config: str,
@@ -947,9 +954,7 @@ async def test_uart_mock_modbus_client_read_write(
"""
line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
tracker = SensorTracker(
["srv_write_1", "srv_read_1", "client_read_0", "client_read_1"]
)
tracker = SensorTracker(["srv_write_1", "client_read_0", "client_read_1"])
futures = tracker.expect_all(
{
"srv_write_1": 4660, # server wrote 0x1234 to reg 0x0001
@@ -967,6 +972,7 @@ async def test_uart_mock_modbus_client_read_write(
_assert_no_modbus_errors(error_log_lines, warning_log_lines)
@pytest.mark.shared_yaml("uart_mock_modbus_loopback")
@pytest.mark.asyncio
async def test_uart_mock_modbus_register_offset(
yaml_config: str,
@@ -1022,6 +1028,7 @@ async def test_uart_mock_modbus_register_offset(
)
@pytest.mark.shared_yaml("uart_mock_modbus_loopback")
@pytest.mark.asyncio
async def test_uart_mock_modbus_lambda_write(
yaml_config: str,
@@ -1058,6 +1065,7 @@ async def test_uart_mock_modbus_lambda_write(
await tracker.await_change(wrote_30, "reg_30", timeout=4.0)
@pytest.mark.shared_yaml("uart_mock_modbus_loopback")
@pytest.mark.asyncio
async def test_uart_mock_modbus_lambda_invert(
yaml_config: str,
@@ -1113,6 +1121,7 @@ async def test_uart_mock_modbus_lambda_invert(
)
@pytest.mark.shared_yaml("uart_mock_modbus_loopback")
@pytest.mark.asyncio
async def test_uart_mock_modbus_deprecated_write_buffer(
yaml_config: str,
+28
View File
@@ -2122,6 +2122,34 @@ def test_get_cpp_changed_components_independent_of_cwd(
) == ["time"]
def test_fixture_map_includes_shared_yaml_markers() -> None:
"""Fixtures named only by shared_yaml markers must map to their test file."""
helpers.get_fixture_to_test_files.cache_clear()
mapping = helpers.get_fixture_to_test_files()
for fixture in (
"uart_mock_modbus_loopback",
"uart_mock_modbus_mesh",
"uart_mock_modbus_server_injected",
):
assert mapping[fixture] == frozenset(
{"tests/integration/test_uart_mock_modbus.py"}
)
def test_no_orphan_integration_fixtures() -> None:
"""Every fixture must reach CI test selection; an orphan selects nothing."""
helpers.get_fixture_to_test_files.cache_clear()
mapping = helpers.get_fixture_to_test_files()
fixtures_dir = (Path(__file__).parent.parent / "integration" / "fixtures").resolve()
fixtures = list(fixtures_dir.glob("*.yaml"))
assert fixtures, f"no fixtures found under {fixtures_dir}"
# cache_init is covered via INTEGRATION_TESTS_TRIGGER_FILES instead
orphans = [
f.stem for f in fixtures if f.stem != "cache_init" and f.stem not in mapping
]
assert not orphans, f"fixtures invisible to CI test selection: {orphans}"
def test_lpt_partition_balances_skewed_weights() -> None:
"""Heavy items spread across groups instead of clustering."""
items = [f"i{n}" for n in range(6)]