mirror of
https://github.com/esphome/esphome.git
synced 2026-08-27 08:28:30 +00:00
[improv_serial] Add uart bus support and host integration test (#18794)
This commit is contained in:
@@ -50,6 +50,7 @@ async def to_code(config: ConfigType) -> None:
|
||||
cg.add_define("USE_ESPHOME_HOST_MAC_ADDRESS", config[CONF_MAC_ADDRESS].parts)
|
||||
cg.add_build_flag("-std=gnu++20")
|
||||
cg.add_define("ESPHOME_BOARD", "host")
|
||||
cg.add_define("ESPHOME_VARIANT", "HOST")
|
||||
cg.add_define(ThreadModel.MULTI_ATOMICS)
|
||||
cg.add_platformio_option("platform", "platformio/native")
|
||||
cg.add_platformio_option("lib_ldf_mode", "off")
|
||||
|
||||
@@ -1,9 +1,15 @@
|
||||
import esphome.codegen as cg
|
||||
from esphome.components import improv_base
|
||||
from esphome.components import improv_base, uart
|
||||
from esphome.components.esp32 import VARIANT_ESP32S3, get_esp32_variant
|
||||
from esphome.components.logger import USB_CDC
|
||||
import esphome.config_validation as cv
|
||||
from esphome.const import CONF_BAUD_RATE, CONF_HARDWARE_UART, CONF_ID, CONF_LOGGER
|
||||
from esphome.const import (
|
||||
CONF_BAUD_RATE,
|
||||
CONF_HARDWARE_UART,
|
||||
CONF_ID,
|
||||
CONF_LOGGER,
|
||||
CONF_UART_ID,
|
||||
)
|
||||
from esphome.core import CORE
|
||||
import esphome.final_validate as fv
|
||||
from esphome.types import ConfigType
|
||||
@@ -17,13 +23,35 @@ improv_serial_ns = cg.esphome_ns.namespace("improv_serial")
|
||||
ImprovSerialComponent = improv_serial_ns.class_("ImprovSerialComponent", cg.Component)
|
||||
|
||||
CONFIG_SCHEMA = (
|
||||
cv.Schema({cv.GenerateID(): cv.declare_id(ImprovSerialComponent)})
|
||||
cv.Schema(
|
||||
{
|
||||
cv.GenerateID(): cv.declare_id(ImprovSerialComponent),
|
||||
# YAML only: rewiring Improv onto another UART is not a knob for a
|
||||
# visual editor and the device builder must not expose it
|
||||
cv.Optional(CONF_UART_ID, visibility=cv.Visibility.YAML_ONLY): cv.use_id(
|
||||
uart.UARTComponent
|
||||
),
|
||||
}
|
||||
)
|
||||
.extend(improv_base.IMPROV_SCHEMA)
|
||||
.extend(cv.COMPONENT_SCHEMA)
|
||||
)
|
||||
|
||||
|
||||
def validate_logger(config: ConfigType) -> None:
|
||||
_UART_FINAL_VALIDATE = uart.final_validate_device_schema(
|
||||
"improv_serial", require_tx=True, require_rx=True
|
||||
)
|
||||
|
||||
|
||||
def validate_transport(config: ConfigType) -> None:
|
||||
if CONF_UART_ID in config:
|
||||
# A dedicated UART bus is used; the logger's serial settings are irrelevant,
|
||||
# but the bus itself must be bidirectional and not claimed by another device
|
||||
_UART_FINAL_VALIDATE(config)
|
||||
return
|
||||
# The host logger has no serial port for Improv to share
|
||||
if CORE.is_host:
|
||||
raise cv.Invalid("improv_serial on the host platform requires uart_id")
|
||||
logger_conf = fv.full_config.get()[CONF_LOGGER]
|
||||
if logger_conf[CONF_BAUD_RATE] == 0:
|
||||
raise cv.Invalid("improv_serial requires the logger baud_rate to be not 0")
|
||||
@@ -36,7 +64,7 @@ def validate_logger(config: ConfigType) -> None:
|
||||
)
|
||||
|
||||
|
||||
FINAL_VALIDATE_SCHEMA = validate_logger
|
||||
FINAL_VALIDATE_SCHEMA = validate_transport
|
||||
|
||||
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
@@ -44,3 +72,6 @@ async def to_code(config: ConfigType) -> None:
|
||||
await cg.register_component(var, config)
|
||||
await improv_base.setup_improv_core(var, config, "improv_serial")
|
||||
cg.add_define("USE_IMPROV_SERIAL")
|
||||
if (uart_id := config.get(CONF_UART_ID)) is not None:
|
||||
cg.add(var.set_uart(await cg.get_variable(uart_id)))
|
||||
cg.add_define("USE_IMPROV_SERIAL_UART")
|
||||
|
||||
@@ -15,7 +15,9 @@ static const char *const TAG = "improv_serial";
|
||||
|
||||
void ImprovSerialComponent::setup() {
|
||||
global_improv_serial_component = this;
|
||||
#ifdef USE_ESP32
|
||||
#ifdef USE_IMPROV_SERIAL_UART
|
||||
// Transport is a dedicated UART bus set via set_uart() in generated code
|
||||
#elif defined(USE_ESP32)
|
||||
this->uart_num_ = logger::global_logger->get_uart_num();
|
||||
this->uart_selection_ = logger::global_logger->get_uart();
|
||||
#elif defined(USE_ARDUINO)
|
||||
@@ -89,7 +91,13 @@ void ImprovSerialComponent::write_data_(const uint8_t *data, const size_t size)
|
||||
}
|
||||
this->tx_header_[TX_CHECKSUM_IDX] = checksum;
|
||||
|
||||
#ifdef USE_ESP32
|
||||
#ifdef USE_IMPROV_SERIAL_UART
|
||||
this->uart_->write_array(this->tx_header_, header_tx_len);
|
||||
if (there_is_data) {
|
||||
this->uart_->write_array(data, size);
|
||||
this->uart_->write_array(&this->tx_header_[TX_CHECKSUM_IDX], 2); // Footer: checksum and newline
|
||||
}
|
||||
#elif defined(USE_ESP32)
|
||||
switch (this->uart_selection_) {
|
||||
case logger::UART_SELECTION_UART0:
|
||||
case logger::UART_SELECTION_UART1:
|
||||
|
||||
@@ -10,7 +10,9 @@
|
||||
#include <improv.h>
|
||||
#include <vector>
|
||||
|
||||
#ifdef USE_ESP32
|
||||
#ifdef USE_IMPROV_SERIAL_UART
|
||||
#include "esphome/components/uart/uart_component.h"
|
||||
#elif defined(USE_ESP32)
|
||||
#include <driver/uart.h>
|
||||
#ifdef USE_LOGGER_USB_SERIAL_JTAG
|
||||
#include <driver/usb_serial_jtag.h>
|
||||
@@ -53,6 +55,10 @@ class ImprovSerialComponent final : public Component, public improv_base::Improv
|
||||
|
||||
float get_setup_priority() const override { return setup_priority::AFTER_WIFI; }
|
||||
|
||||
#ifdef USE_IMPROV_SERIAL_UART
|
||||
void set_uart(uart::UARTComponent *uart) { this->uart_ = uart; }
|
||||
#endif
|
||||
|
||||
protected:
|
||||
bool parse_improv_serial_byte_(uint8_t byte);
|
||||
bool parse_improv_payload_(improv::ImprovCommand &command);
|
||||
@@ -69,7 +75,11 @@ class ImprovSerialComponent final : public Component, public improv_base::Improv
|
||||
ESPHOME_ALWAYS_INLINE optional<uint8_t> read_byte_() {
|
||||
optional<uint8_t> byte;
|
||||
uint8_t data = 0;
|
||||
#ifdef USE_ESP32
|
||||
#ifdef USE_IMPROV_SERIAL_UART
|
||||
if (this->uart_->available() && this->uart_->read_byte(&data)) {
|
||||
byte = data;
|
||||
}
|
||||
#elif defined(USE_ESP32)
|
||||
switch (this->uart_selection_) {
|
||||
case logger::UART_SELECTION_UART0:
|
||||
case logger::UART_SELECTION_UART1:
|
||||
@@ -129,7 +139,9 @@ class ImprovSerialComponent final : public Component, public improv_base::Improv
|
||||
'\n',
|
||||
};
|
||||
|
||||
#ifdef USE_ESP32
|
||||
#ifdef USE_IMPROV_SERIAL_UART
|
||||
uart::UARTComponent *uart_{nullptr};
|
||||
#elif defined(USE_ESP32)
|
||||
uart_port_t uart_num_;
|
||||
logger::UARTSelection uart_selection_{logger::UART_SELECTION_UART0};
|
||||
#elif defined(USE_ARDUINO)
|
||||
|
||||
@@ -537,6 +537,8 @@
|
||||
|
||||
#ifdef USE_HOST
|
||||
#define USE_HTTP_REQUEST_RESPONSE
|
||||
// Host only: the uart arm would shadow the native logger UART arms in other envs
|
||||
#define USE_IMPROV_SERIAL_UART
|
||||
#define USE_SOCKET_IMPL_BSD_SOCKETS
|
||||
#define USE_ESPHOME_TASK_LOG_BUFFER
|
||||
#define ESPHOME_TASK_LOG_BUFFER_SIZE 64
|
||||
|
||||
@@ -247,6 +247,8 @@ def lint_ext_check(fname):
|
||||
"CLAUDE.md",
|
||||
"GEMINI.md",
|
||||
".github/copilot-instructions.md",
|
||||
# Symlink to the real wifi scan_list.h so the test stub cannot drift
|
||||
"tests/integration/fixtures/external_components/wifi/scan_list.h",
|
||||
]
|
||||
)
|
||||
def lint_executable_bit(fname: Path) -> str | None:
|
||||
|
||||
@@ -0,0 +1,11 @@
|
||||
wifi:
|
||||
ssid: MySSID
|
||||
password: password1
|
||||
|
||||
# Serial logging off; on a dedicated UART bus improv_serial must not
|
||||
# require the logger's serial settings
|
||||
logger:
|
||||
baud_rate: 0
|
||||
|
||||
improv_serial:
|
||||
uart_id: uart_bus
|
||||
@@ -0,0 +1,3 @@
|
||||
packages:
|
||||
uart: !include ../../test_build_components/common/uart/esp32-idf.yaml
|
||||
improv_serial: !include common-uart-bus.yaml
|
||||
@@ -0,0 +1,3 @@
|
||||
packages:
|
||||
uart: !include ../../test_build_components/common/uart/esp8266-ard.yaml
|
||||
improv_serial: !include common-uart-bus.yaml
|
||||
@@ -0,0 +1,39 @@
|
||||
"""Host-only stub of the wifi component for integration tests.
|
||||
|
||||
HOST-ONLY TEST COMPONENT: this shadows the real wifi component for EVERY
|
||||
fixture that uses the shared external_components directory. Any host fixture
|
||||
with a wifi block gets this stub, not the real component: fixed scan results,
|
||||
is_connected() hardwired true, and save_wifi_sta that only logs. See
|
||||
wifi_component.h for the full behavior.
|
||||
"""
|
||||
|
||||
import esphome.codegen as cg
|
||||
import esphome.config_validation as cv
|
||||
from esphome.const import CONF_ID, CONF_PASSWORD, CONF_SSID, CONF_USE_ADDRESS
|
||||
from esphome.types import ConfigType
|
||||
|
||||
CODEOWNERS = ["@esphome/tests"]
|
||||
|
||||
wifi_ns = cg.esphome_ns.namespace("wifi")
|
||||
WiFiComponent = wifi_ns.class_("WiFiComponent", cg.Component)
|
||||
|
||||
CONFIG_SCHEMA = cv.Schema(
|
||||
{
|
||||
cv.GenerateID(): cv.declare_id(WiFiComponent),
|
||||
# Accepted for fixture realism; the stub ignores them
|
||||
cv.Optional(CONF_SSID): cv.string,
|
||||
cv.Optional(CONF_PASSWORD): cv.string,
|
||||
# Read by StorageJSON via CORE.address whenever a wifi block exists
|
||||
cv.Optional(CONF_USE_ADDRESS, default="localhost"): cv.string,
|
||||
}
|
||||
).extend(cv.COMPONENT_SCHEMA)
|
||||
|
||||
|
||||
def check_placeholder_credentials(config: ConfigType) -> None:
|
||||
"""Compile-time hook the esphome CLI imports from the wifi module; no-op here."""
|
||||
|
||||
|
||||
async def to_code(config: ConfigType) -> None:
|
||||
var = cg.new_Pvariable(config[CONF_ID])
|
||||
await cg.register_component(var, config)
|
||||
cg.add_define("USE_WIFI")
|
||||
@@ -0,0 +1 @@
|
||||
../../../../../esphome/components/wifi/scan_list.h
|
||||
@@ -0,0 +1,40 @@
|
||||
#include "wifi_component.h"
|
||||
|
||||
#include "esphome/core/log.h"
|
||||
|
||||
namespace esphome::wifi {
|
||||
|
||||
static const char *const TAG = "wifi_stub";
|
||||
|
||||
WiFiComponent *global_wifi_component = nullptr; // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
|
||||
|
||||
WiFiComponent::WiFiComponent() { global_wifi_component = this; }
|
||||
|
||||
void WiFiComponent::setup() { ESP_LOGI(TAG, "Stub wifi ready"); }
|
||||
|
||||
void WiFiComponent::dump_config() { ESP_LOGCONFIG(TAG, "Stub wifi"); }
|
||||
|
||||
void WiFiComponent::start_scanning() {
|
||||
// Duplicate TestNet entry (weaker) and a hidden entry exercise the
|
||||
// should_show_scan_entry dedup and filtering logic
|
||||
this->scan_result_.clear();
|
||||
this->scan_result_.emplace_back("TestNet", -50, true, false);
|
||||
this->scan_result_.emplace_back("TestNet", -60, true, false);
|
||||
this->scan_result_.emplace_back("OpenNet", -70, false, false);
|
||||
this->scan_result_.emplace_back("", -40, false, true);
|
||||
ESP_LOGI(TAG, "Scan complete with %zu results", this->scan_result_.size());
|
||||
}
|
||||
|
||||
void WiFiComponent::set_sta(const WiFiAP &ap) { ESP_LOGI(TAG, "set_sta ssid=%s", ap.get_ssid().c_str()); }
|
||||
|
||||
void WiFiComponent::start_connecting(const WiFiAP &ap) {
|
||||
ESP_LOGI(TAG, "start_connecting ssid=%s", ap.get_ssid().c_str());
|
||||
}
|
||||
|
||||
void WiFiComponent::clear_sta() { ESP_LOGI(TAG, "clear_sta"); }
|
||||
|
||||
void WiFiComponent::save_wifi_sta(StringRef ssid, StringRef password) {
|
||||
ESP_LOGI(TAG, "save_wifi_sta ssid=%s password_len=%zu", ssid.c_str(), password.size());
|
||||
}
|
||||
|
||||
} // namespace esphome::wifi
|
||||
@@ -0,0 +1,77 @@
|
||||
#pragma once
|
||||
|
||||
// ============================================================================
|
||||
// HOST-ONLY TEST COMPONENT — DO NOT COPY TO PRODUCTION CODE
|
||||
//
|
||||
// Stub of the real wifi component with just enough API surface for
|
||||
// improv_serial to build and run on the host platform. Scan results are
|
||||
// fixed, "connecting" succeeds immediately, and save_wifi_sta only logs so
|
||||
// tests can assert on the log output.
|
||||
// ============================================================================
|
||||
|
||||
#include "esphome/components/network/ip_address.h"
|
||||
#include "esphome/core/component.h"
|
||||
#include "esphome/core/string_ref.h"
|
||||
|
||||
#include <string>
|
||||
#include <vector>
|
||||
|
||||
namespace esphome::wifi {
|
||||
|
||||
class WiFiAP {
|
||||
public:
|
||||
void set_ssid(const char *ssid) { this->ssid_ = ssid; }
|
||||
void set_password(const char *password) { this->password_ = password; }
|
||||
StringRef get_ssid() const { return StringRef(this->ssid_); }
|
||||
StringRef get_password() const { return StringRef(this->password_); }
|
||||
|
||||
protected:
|
||||
std::string ssid_;
|
||||
std::string password_;
|
||||
};
|
||||
|
||||
class WiFiScanResult {
|
||||
public:
|
||||
WiFiScanResult(const char *ssid, int8_t rssi, bool with_auth, bool hidden)
|
||||
: ssid_(ssid), rssi_(rssi), with_auth_(with_auth), hidden_(hidden) {}
|
||||
StringRef get_ssid() const { return StringRef(this->ssid_); }
|
||||
int8_t get_rssi() const { return this->rssi_; }
|
||||
bool get_with_auth() const { return this->with_auth_; }
|
||||
bool get_is_hidden() const { return this->hidden_; }
|
||||
bool ssid_equals(const WiFiScanResult &other) const { return this->ssid_ == other.ssid_; }
|
||||
|
||||
protected:
|
||||
std::string ssid_;
|
||||
int8_t rssi_;
|
||||
bool with_auth_;
|
||||
bool hidden_;
|
||||
};
|
||||
|
||||
class WiFiComponent : public Component {
|
||||
public:
|
||||
WiFiComponent();
|
||||
void setup() override;
|
||||
void dump_config() override;
|
||||
float get_setup_priority() const override { return setup_priority::WIFI; }
|
||||
|
||||
bool has_sta() const { return false; }
|
||||
bool is_disabled() const { return false; }
|
||||
// Always connected so network::is_connected() keeps the API server accepting clients
|
||||
bool is_connected() const { return true; }
|
||||
void start_scanning();
|
||||
const std::vector<WiFiScanResult> &get_scan_result() const { return this->scan_result_; }
|
||||
void set_sta(const WiFiAP &ap);
|
||||
void start_connecting(const WiFiAP &ap);
|
||||
void clear_sta();
|
||||
void save_wifi_sta(StringRef ssid, StringRef password);
|
||||
// Called by network::util on any USE_WIFI build
|
||||
const char *get_use_address() const { return "localhost"; }
|
||||
network::IPAddresses get_ip_addresses() { return {}; }
|
||||
|
||||
protected:
|
||||
std::vector<WiFiScanResult> scan_result_;
|
||||
};
|
||||
|
||||
extern WiFiComponent *global_wifi_component; // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
|
||||
|
||||
} // namespace esphome::wifi
|
||||
@@ -0,0 +1,40 @@
|
||||
esphome:
|
||||
# Short name keeps the device info payload under uart_mock's 64 byte log cap
|
||||
name: improv-uart
|
||||
|
||||
host:
|
||||
api:
|
||||
actions:
|
||||
- action: uart_inject
|
||||
variables:
|
||||
payload: int[]
|
||||
then:
|
||||
- uart_mock.inject_rx:
|
||||
id: mock_uart
|
||||
data: !lambda return std::vector<uint8_t>(payload.begin(), payload.end());
|
||||
|
||||
logger:
|
||||
level: DEBUG
|
||||
|
||||
external_components:
|
||||
- source:
|
||||
type: local
|
||||
path: EXTERNAL_COMPONENT_PATH
|
||||
|
||||
# Host-only stub shadowing the real wifi component (see external_components/wifi)
|
||||
wifi:
|
||||
ssid: TestNet
|
||||
password: password1
|
||||
|
||||
# Dummy uart entry so the uart component sources are part of the build; the
|
||||
# actual bus used by improv_serial is the uart_mock component below
|
||||
uart:
|
||||
baud_rate: 115200
|
||||
port: /dev/null
|
||||
|
||||
uart_mock:
|
||||
id: mock_uart
|
||||
baud_rate: 115200
|
||||
|
||||
improv_serial:
|
||||
uart_id: mock_uart
|
||||
@@ -0,0 +1,43 @@
|
||||
"""Helpers for asserting on log output in integration tests."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
|
||||
|
||||
class LineWaiter:
|
||||
"""Collects log lines and lets a test await one containing all needles.
|
||||
|
||||
Pass ``callback`` as ``run_compiled``'s ``line_callback``; the callback runs
|
||||
on the test's own event loop, so futures are resolved directly. Only one
|
||||
``wait_for`` may be outstanding at a time (tests await sequentially).
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.lines: list[str] = []
|
||||
self._needles: tuple[str, ...] = ()
|
||||
self._future: asyncio.Future | None = None
|
||||
|
||||
def callback(self, line: str) -> None:
|
||||
self.lines.append(line)
|
||||
if (
|
||||
self._future is not None
|
||||
and not self._future.done()
|
||||
and all(n in line for n in self._needles)
|
||||
):
|
||||
self._future.set_result(line)
|
||||
self._future = None
|
||||
|
||||
async def wait_for(self, *needles: str, timeout: float = 10.0) -> str:
|
||||
"""Return the first line, past or future, containing every needle."""
|
||||
for line in self.lines:
|
||||
if all(n in line for n in needles):
|
||||
return line
|
||||
assert self._future is None or self._future.done(), "concurrent wait_for"
|
||||
self._needles = needles
|
||||
self._future = asyncio.get_running_loop().create_future()
|
||||
try:
|
||||
return await asyncio.wait_for(self._future, timeout)
|
||||
finally:
|
||||
self._future = None
|
||||
self._needles = ()
|
||||
@@ -0,0 +1,140 @@
|
||||
"""Integration test for improv_serial over a mocked UART bus.
|
||||
|
||||
Drives the improv serial protocol end to end on the host platform:
|
||||
the fixture wires improv_serial to a uart_mock bus and shadows the wifi
|
||||
component with a host stub. The test injects improv frames through an API
|
||||
action and asserts on the framed responses that uart_mock logs as TX lines.
|
||||
|
||||
Covered:
|
||||
1. Get Current State reports AUTHORIZED
|
||||
2. Get Device Info returns the firmware/device info RPC response
|
||||
3. Get Wi-Fi Networks returns deduplicated scan results and a terminator
|
||||
4. Wi-Fi Settings provisions: saves credentials and reports PROVISIONED
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from .log_utils import LineWaiter
|
||||
from .types import APIClientConnectedFactory, RunCompiledFunction
|
||||
|
||||
# Improv serial framing (improv_serial_component.h)
|
||||
IMPROV_HEADER = b"IMPROV"
|
||||
IMPROV_VERSION = 1
|
||||
TYPE_CURRENT_STATE = 0x01
|
||||
TYPE_RPC = 0x03
|
||||
TYPE_RPC_RESPONSE = 0x04
|
||||
|
||||
# improv::Command values
|
||||
CMD_GET_CURRENT_STATE = 0x02
|
||||
CMD_GET_DEVICE_INFO = 0x03
|
||||
CMD_GET_WIFI_NETWORKS = 0x04
|
||||
CMD_WIFI_SETTINGS = 0x01
|
||||
|
||||
|
||||
def build_rpc_frame(command: int, data: bytes = b"") -> list[int]:
|
||||
"""Build a full improv serial frame carrying one RPC command."""
|
||||
payload = bytes([command, len(data)]) + data
|
||||
frame = IMPROV_HEADER + bytes([IMPROV_VERSION, TYPE_RPC, len(payload)]) + payload
|
||||
checksum = sum(frame) & 0xFF
|
||||
return list(frame + bytes([checksum]) + b"\n")
|
||||
|
||||
|
||||
def state_frame_hex(state: int) -> str:
|
||||
"""Full 12 byte current-state frame as hex, checksum and newline included."""
|
||||
frame = IMPROV_HEADER + bytes([IMPROV_VERSION, TYPE_CURRENT_STATE, 1, state])
|
||||
checksum = sum(frame) & 0xFF
|
||||
return ":".join(f"{b:02X}" for b in frame + bytes([checksum]) + b"\n")
|
||||
|
||||
|
||||
def rpc_footer_hex(payload: bytes) -> str:
|
||||
"""Checksum and newline footer written after an RPC response payload."""
|
||||
header = IMPROV_HEADER + bytes([IMPROV_VERSION, TYPE_RPC_RESPONSE, len(payload)])
|
||||
checksum = (sum(header) + sum(payload)) & 0xFF
|
||||
return f"{checksum:02X}:0A"
|
||||
|
||||
|
||||
def wifi_settings_data(ssid: str, password: str) -> bytes:
|
||||
ssid_b = ssid.encode()
|
||||
pass_b = password.encode()
|
||||
return bytes([len(ssid_b)]) + ssid_b + bytes([len(pass_b)]) + pass_b
|
||||
|
||||
|
||||
def hex_of(text: str) -> str:
|
||||
"""Colon separated uppercase hex as logged by format_hex_pretty."""
|
||||
return ":".join(f"{b:02X}" for b in text.encode())
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_improv_serial_uart(
|
||||
yaml_config: str,
|
||||
run_compiled: RunCompiledFunction,
|
||||
api_client_connected: APIClientConnectedFactory,
|
||||
) -> None:
|
||||
waiter = LineWaiter()
|
||||
|
||||
async with (
|
||||
run_compiled(yaml_config, line_callback=waiter.callback),
|
||||
api_client_connected() as client,
|
||||
):
|
||||
_entities, services = await client.list_entities_services()
|
||||
inject = next(s for s in services if s.name == "uart_inject")
|
||||
|
||||
# 1. Get Current State: expect the complete current-state frame reporting
|
||||
# AUTHORIZED (0x02), checksum and newline included
|
||||
await client.execute_service(
|
||||
inject, {"payload": build_rpc_frame(CMD_GET_CURRENT_STATE)}
|
||||
)
|
||||
await waiter.wait_for("uart_mock", f"TX 12 bytes: {state_frame_hex(0x02)}")
|
||||
|
||||
# 2. Get Device Info: the always logged 9 byte response header, then the
|
||||
# payload with the firmware name (must stay under uart_mock's 64 byte
|
||||
# hex dump cap or the payload line reads "too large to log")
|
||||
await client.execute_service(
|
||||
inject, {"payload": build_rpc_frame(CMD_GET_DEVICE_INFO)}
|
||||
)
|
||||
await waiter.wait_for("uart_mock", "TX 9 bytes: 49:4D:50:52:4F:56:01:04")
|
||||
await waiter.wait_for("uart_mock", "TX ", hex_of("ESPHome"))
|
||||
|
||||
# 3. Get Wi-Fi Networks: stub scan has TestNet twice (dedup keeps the
|
||||
# stronger), OpenNet, and a hidden entry (filtered). Expect one response
|
||||
# per visible network plus the empty terminator.
|
||||
await client.execute_service(
|
||||
inject, {"payload": build_rpc_frame(CMD_GET_WIFI_NETWORKS)}
|
||||
)
|
||||
await waiter.wait_for("uart_mock", hex_of("TestNet"))
|
||||
await waiter.wait_for("uart_mock", hex_of("OpenNet"))
|
||||
# Terminator: all three writes of the response frame; 9 byte header,
|
||||
# payload [0x04, 0x00, 0x00], then the checksum and newline footer
|
||||
await waiter.wait_for("uart_mock", "TX 9 bytes: 49:4D:50:52:4F:56:01:04:03")
|
||||
await waiter.wait_for("uart_mock", "TX 3 bytes: 04:00:00")
|
||||
await waiter.wait_for(
|
||||
"uart_mock", f"TX 2 bytes: {rpc_footer_hex(bytes([0x04, 0x00, 0x00]))}"
|
||||
)
|
||||
testnet_count = sum(
|
||||
1
|
||||
for line in waiter.lines
|
||||
if "uart_mock" in line and "TX " in line and hex_of("TestNet") in line
|
||||
)
|
||||
assert testnet_count == 1, (
|
||||
f"Duplicate scan entry not deduplicated: {testnet_count} TestNet responses"
|
||||
)
|
||||
|
||||
# 4. Wi-Fi Settings: stub connects immediately; expect the credentials
|
||||
# saved, the PROVISIONED state frame (0x04), and the settings response
|
||||
await client.execute_service(
|
||||
inject,
|
||||
{
|
||||
"payload": build_rpc_frame(
|
||||
CMD_WIFI_SETTINGS, wifi_settings_data("NewNet", "secret123")
|
||||
)
|
||||
},
|
||||
)
|
||||
await waiter.wait_for("save_wifi_sta ssid=NewNet")
|
||||
await waiter.wait_for("uart_mock", f"TX 12 bytes: {state_frame_hex(0x04)}")
|
||||
# Settings RPC response with no URLs: payload [0x01, 0x00, 0x00] and footer
|
||||
await waiter.wait_for("uart_mock", "TX 3 bytes: 01:00:00")
|
||||
await waiter.wait_for(
|
||||
"uart_mock", f"TX 2 bytes: {rpc_footer_hex(bytes([0x01, 0x00, 0x00]))}"
|
||||
)
|
||||
Reference in New Issue
Block a user