Merge branch 'api-encode-pos-by-value' into api-static-encode-entry

This commit is contained in:
J. Nick Koston
2026-09-10 19:23:12 -05:00
140 changed files with 4427 additions and 1845 deletions
@@ -40,6 +40,9 @@ class SerialProxy {
return SerialProxyResult::SERIAL_PROXY_RESULT_OK;
}
void write_from_client(api::APIConnection *api_connection, const uint8_t *data, size_t len) {}
SerialProxyResult set_mode_from_client(api::APIConnection *api_connection, api::enums::SerialProxyMode mode) {
return SerialProxyResult::SERIAL_PROXY_RESULT_OK;
}
SerialProxyResult set_modem_pins(api::APIConnection *api_connection, uint32_t line_states) {
return SerialProxyResult::SERIAL_PROXY_RESULT_OK;
}
@@ -0,0 +1,32 @@
esphome:
name: test-keyboard-no-label
esp32:
board: esp32dev
framework:
type: esp-idf
spi:
- id: spi_bus
clk_pin: GPIO18
mosi_pin: GPIO23
display:
- platform: mipi_spi
spi_id: spi_bus
model: st7789v
id: tft_display
dimensions:
width: 240
height: 320
cs_pin: GPIO22
dc_pin: GPIO21
auto_clear_enabled: false
invert_colors: false
update_interval: never
lvgl:
displays: tft_display
widgets:
- keyboard:
id: keyboard_widget
@@ -0,0 +1,34 @@
esphome:
name: test-qrcode-no-label
esp32:
board: esp32dev
framework:
type: esp-idf
spi:
- id: spi_bus
clk_pin: GPIO18
mosi_pin: GPIO23
display:
- platform: mipi_spi
spi_id: spi_bus
model: st7789v
id: tft_display
dimensions:
width: 240
height: 320
cs_pin: GPIO22
dc_pin: GPIO21
auto_clear_enabled: false
invert_colors: false
update_interval: never
lvgl:
displays: tft_display
widgets:
- qrcode:
id: qr_widget
size: 100
text: "esphome.io"
@@ -0,0 +1,35 @@
esphome:
name: test-tabview-no-label
esp32:
board: esp32dev
framework:
type: esp-idf
spi:
- id: spi_bus
clk_pin: GPIO18
mosi_pin: GPIO23
display:
- platform: mipi_spi
spi_id: spi_bus
model: st7789v
id: tft_display
dimensions:
width: 240
height: 320
cs_pin: GPIO22
dc_pin: GPIO21
auto_clear_enabled: false
invert_colors: false
update_interval: never
lvgl:
displays: tft_display
widgets:
- tabview:
id: tabview_widget
tabs:
- name: "Tab 1"
id: tab_1
@@ -0,0 +1,32 @@
"""Widgets whose LVGL C implementation creates or references labels
internally (tab titles, key legends, the QR canvas fallback) must declare
the label dependency in ``get_uses()``. Otherwise a config that contains
no ``label`` widget of its own compiles LVGL without ``LV_USE_LABEL`` and
fails at C compile time with undefined ``lv_label_*`` symbols.
"""
from __future__ import annotations
from collections.abc import Callable
from pathlib import Path
import pytest
from esphome.components.lvgl import defines as df
@pytest.mark.parametrize(
"yaml_file",
[
"qrcode_no_label.yaml",
"keyboard_no_label.yaml",
"tabview_no_label.yaml",
],
)
def test_label_less_config_enables_lv_use_label(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
yaml_file: str,
) -> None:
generate_main(component_config_path(yaml_file))
assert "LV_USE_LABEL" in df.get_defines()
@@ -0,0 +1,12 @@
esphome:
name: test
esp32:
board: esp32dev
framework:
type: esp-idf
wifi:
ap:
sendspin:
@@ -0,0 +1,18 @@
esphome:
name: test
project:
name: project_manufacturer.project_model
version: 9.9.9
esp32:
board: esp32dev
framework:
type: esp-idf
wifi:
ap:
sendspin:
manufacturer: Explicit Manufacturer
model: Explicit Model
firmware_version: 1.2.3
@@ -0,0 +1,15 @@
esphome:
name: test
project:
name: project_manufacturer.project_model
version: 9.9.9
esp32:
board: esp32dev
framework:
type: esp-idf
wifi:
ap:
sendspin:
@@ -0,0 +1,83 @@
"""Tests for the device information the sendspin hub reports to the server."""
from __future__ import annotations
from collections.abc import Callable
from pathlib import Path
import pytest
from esphome import config_validation as cv
from esphome.components.sendspin import (
CONF_FIRMWARE_VERSION,
CONF_MANUFACTURER,
CONFIG_SCHEMA,
)
from esphome.const import CONF_MODEL, PlatformFramework
from tests.component_tests.types import SetCoreConfigCallable
def test_explicit_device_info_wins_over_project(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
"""Configured values take precedence over the project information."""
main_cpp = generate_main(component_config_path("device_info_explicit.yaml"))
assert 'set_manufacturer("Explicit Manufacturer")' in main_cpp
assert 'set_model("Explicit Model")' in main_cpp
assert 'set_firmware_version("1.2.3")' in main_cpp
def test_project_supplies_device_info(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
"""Without configured values, the project name splits into manufacturer and model."""
main_cpp = generate_main(component_config_path("device_info_project.yaml"))
assert 'set_manufacturer("project_manufacturer")' in main_cpp
assert 'set_model("project_model")' in main_cpp
assert 'set_firmware_version("9.9.9")' in main_cpp
def test_no_device_info_leaves_hub_defaults(
generate_main: Callable[[str | Path], str],
component_config_path: Callable[[str], Path],
) -> None:
"""With neither source, nothing is emitted and the hub keeps its own defaults."""
main_cpp = generate_main(component_config_path("device_info_default.yaml"))
assert "set_manufacturer(" not in main_cpp
assert "set_model(" not in main_cpp
assert "set_firmware_version(" not in main_cpp
@pytest.mark.parametrize(
"conf_key", [CONF_MANUFACTURER, CONF_MODEL, CONF_FIRMWARE_VERSION]
)
def test_empty_device_info_rejected(
set_core_config: SetCoreConfigCallable, conf_key: str
) -> None:
"""An empty string would be sent to the server as an empty value, so it is not accepted."""
set_core_config(PlatformFramework.ESP32_IDF)
with pytest.raises(cv.Invalid):
CONFIG_SCHEMA({conf_key: ""})
@pytest.mark.parametrize(
"conf_key", [CONF_MANUFACTURER, CONF_MODEL, CONF_FIRMWARE_VERSION]
)
def test_device_info_capped_at_127_bytes(
set_core_config: SetCoreConfigCallable, conf_key: str
) -> None:
"""The cap is in bytes so the protobuf length prefix stays a single byte."""
set_core_config(PlatformFramework.ESP32_IDF)
CONFIG_SCHEMA({conf_key: "a" * 127})
with pytest.raises(cv.Invalid):
CONFIG_SCHEMA({conf_key: "a" * 128})
# 64 two-byte characters is 128 bytes.
with pytest.raises(cv.Invalid):
CONFIG_SCHEMA({conf_key: "é" * 64})
@@ -0,0 +1,90 @@
"""Validation tests for the sendspin media_source platform.
These cover the codec preference list, whose rejection branches a compile test
cannot reach: a `test*.yaml` can only assert that a configuration is accepted.
"""
from typing import Any
import pytest
from esphome import config_validation as cv
from esphome.components.sendspin import CONF_CODECS, _get_data
from esphome.components.sendspin.media_source import CONFIG_SCHEMA
from esphome.const import PlatformFramework
from esphome.types import ConfigType
from tests.component_tests.types import SetCoreConfigCallable
def _media_source_config(**overrides: Any) -> ConfigType:
"""Build a minimal valid media source config, allowing field overrides."""
config: ConfigType = {
"id": "sendspin_media_source",
"sendspin_id": "sendspin_hub",
}
config.update(overrides)
return config
def test_default_codecs_at_48_khz(set_core_config: SetCoreConfigCallable) -> None:
"""Every codec is advertised when the sample rate suits all of them."""
set_core_config(PlatformFramework.ESP32_IDF)
config = CONFIG_SCHEMA(_media_source_config())
assert config[CONF_CODECS] == ["flac", "opus", "pcm"]
def test_default_codecs_drop_opus_at_other_rates(
set_core_config: SetCoreConfigCallable,
) -> None:
"""Opus only supports 48 kHz, so it leaves the default list at other rates."""
set_core_config(PlatformFramework.ESP32_IDF)
config = CONFIG_SCHEMA(_media_source_config(sample_rate=44100))
assert config[CONF_CODECS] == ["flac", "pcm"]
def test_configured_order_is_preserved(set_core_config: SetCoreConfigCallable) -> None:
"""The list is a preference order, so it reaches the player role as written."""
set_core_config(PlatformFramework.ESP32_IDF)
CONFIG_SCHEMA(_media_source_config(codecs=["pcm", "flac"]))
assert _get_data().player_config[CONF_CODECS] == ["pcm", "flac"]
def test_empty_codec_list_rejected(set_core_config: SetCoreConfigCallable) -> None:
"""A player with no codecs at all could never be given a stream."""
set_core_config(PlatformFramework.ESP32_IDF)
with pytest.raises(cv.Invalid, match="length of value must be at least 1"):
CONFIG_SCHEMA(_media_source_config(codecs=[]))
def test_duplicate_codec_rejected(set_core_config: SetCoreConfigCallable) -> None:
"""A repeated codec has no meaning in a preference order."""
set_core_config(PlatformFramework.ESP32_IDF)
with pytest.raises(cv.Invalid, match="may only be listed once"):
CONFIG_SCHEMA(_media_source_config(codecs=["flac", "flac"]))
def test_unknown_codec_rejected(set_core_config: SetCoreConfigCallable) -> None:
"""Only codecs the player role can decode are accepted."""
set_core_config(PlatformFramework.ESP32_IDF)
with pytest.raises(cv.Invalid, match="Unknown value"):
CONFIG_SCHEMA(_media_source_config(codecs=["mp3"]))
def test_opus_at_wrong_sample_rate_rejected(
set_core_config: SetCoreConfigCallable,
) -> None:
"""Asking for Opus at a rate it cannot handle fails rather than silently
dropping the stated preference."""
set_core_config(PlatformFramework.ESP32_IDF)
with pytest.raises(cv.Invalid, match="requires a sample_rate of 48000"):
CONFIG_SCHEMA(_media_source_config(codecs=["opus"], sample_rate=44100))
@@ -0,0 +1,76 @@
"""The template platforms surface value-describing metadata on the main form.
Hardware platforms get sensible defaults for unit/device_class/etc., so those
fields fall through to the editor's advanced disclosure. A ``template`` entity
has no such defaults -- the user is expected to define them -- so the template
platforms pass ``visibility=cv.Visibility.UI`` to promote them onto the form.
"""
from __future__ import annotations
import importlib
import pytest
import esphome.config_validation as cv
def _markers(schema: cv.Schema) -> dict[str, object]:
s = schema
if hasattr(s, "validators"):
# cv.All -> the schema is the first validator.
s = s.validators[0]
return {str(k): k for k in s.schema}
@pytest.mark.parametrize(
("platform", "fields"),
[
(
"sensor",
[
"unit_of_measurement",
"accuracy_decimals",
"device_class",
"state_class",
"force_update",
],
),
("binary_sensor", ["device_class"]),
("switch", ["device_class"]),
("cover", ["device_class"]),
("button", ["device_class"]),
("valve", ["device_class"]),
("event", ["device_class"]),
("text_sensor", ["device_class"]),
("number", ["device_class", "unit_of_measurement"]),
],
)
def test_template_metadata_is_ui(platform: str, fields: list[str]) -> None:
mod = importlib.import_module(f"esphome.components.template.{platform}")
markers = _markers(mod.CONFIG_SCHEMA)
for field in fields:
assert markers[field].visibility is cv.Visibility.UI, f"{platform}.{field}"
def test_template_sensor_promotion_preserves_defaults() -> None:
"""Promoting to UI must not drop the fields' defaults."""
from esphome.components.template.sensor import CONFIG_SCHEMA
markers = _markers(CONFIG_SCHEMA)
assert markers["accuracy_decimals"].default() == 1
assert markers["force_update"].default() is False
def test_hardware_platform_metadata_not_promoted() -> None:
"""Without ``visibility=`` the builders leave metadata unset.
Unset markers fall through to the consumer's ``Optional`` default of
advanced, so hardware platforms are unaffected by the template promotion.
"""
from esphome.components import binary_sensor, sensor
hw_sensor = _markers(sensor.sensor_schema(device_class="temperature"))
assert hw_sensor["device_class"].visibility is None
hw_bs = _markers(binary_sensor.binary_sensor_schema(device_class="motion"))
assert hw_bs["device_class"].visibility is None
+5
View File
@@ -0,0 +1,5 @@
from tests.testing_helpers import ComponentManifestOverride
def override_manifest(manifest: ComponentManifestOverride) -> None:
manifest.dependencies = manifest.dependencies + ["sensor", "spi"]
@@ -0,0 +1,62 @@
#include <gtest/gtest.h>
#include "esphome/components/atm90e32/atm90e32.h"
namespace esphome::atm90e32::testing {
TEST(ATM90E32OffsetRegisterVerification, AcceptsExactSignedReadback) {
EXPECT_TRUE(offset_register_value_matches(0x007B, 123));
EXPECT_TRUE(offset_register_value_matches(0xFF85, -123));
}
TEST(ATM90E32OffsetRegisterVerification, RejectsMismatchedReadback) {
EXPECT_FALSE(offset_register_value_matches(0x007C, 123));
EXPECT_FALSE(offset_register_value_matches(0xFF84, -123));
}
TEST(ATM90E32OffsetRestoreState, ReportsVerifiedStoredValuesAsRestored) {
const auto state = resolve_offset_restore_state(true, true, false);
EXPECT_TRUE(state.restored);
EXPECT_TRUE(state.values_verified);
}
TEST(ATM90E32OffsetRestoreState, ReportsVerifiedConfigFallbackAsNotRestored) {
const auto state = resolve_offset_restore_state(true, false, true);
EXPECT_FALSE(state.restored);
EXPECT_TRUE(state.values_verified);
}
TEST(ATM90E32OffsetRestoreState, ReportsFailedConfigFallbackAsUnverified) {
const auto state = resolve_offset_restore_state(true, false, false);
EXPECT_FALSE(state.restored);
EXPECT_FALSE(state.values_verified);
}
TEST(ATM90E32OffsetRestoreState, ReportsConfigWithoutStoredValuesAsNotRestored) {
const auto state = resolve_offset_restore_state(false, true, false);
EXPECT_FALSE(state.restored);
EXPECT_TRUE(state.values_verified);
}
TEST(ATM90E32OffsetPersistence, RollsBackStoredValuesOrZeroSentinel) {
const OffsetCalibration previous[3]{{1, -1}, {2, -2}, {3, -3}};
OffsetCalibration rollback[3]{};
prepare_offset_rollback(previous, true, rollback);
for (uint8_t phase = 0; phase < 3; phase++) {
EXPECT_EQ(rollback[phase].first_offset, previous[phase].first_offset);
EXPECT_EQ(rollback[phase].second_offset, previous[phase].second_offset);
}
prepare_offset_rollback(previous, false, rollback);
for (const auto &phase : rollback) {
EXPECT_EQ(phase.first_offset, 0);
EXPECT_EQ(phase.second_offset, 0);
}
}
} // namespace esphome::atm90e32::testing
@@ -4,3 +4,6 @@ psram:
sendspin:
id: sendspin_hub_id
task_stack_in_psram: true
manufacturer: Test Manufacturer
model: Test Model
firmware_version: 1.2.3
@@ -0,0 +1,14 @@
substitutions:
tx_pin: GPIO4
rx_pin: GPIO5
# Compile the tap code paths; no tap is attached, so this exercises the
# null-tap branches that a normal build never defines.
esphome:
platformio_options:
build_flags:
- "-DUSE_SERIAL_PROXY_TAP"
packages:
uart: !include ../../test_build_components/common/uart/esp32-idf.yaml
serial_proxy: !include common.yaml
@@ -0,0 +1,29 @@
# Tuya without any network component (no wifi/ethernet/api), as used on
# serial-only or BLE-only Tuya MCU boards. Regression test for
# https://github.com/esphome/esphome/issues/18942
substitutions:
status_pin: P6
packages:
uart: !include ../../test_build_components/common/uart/bk72xx-ard.yaml
tuya:
status_pin: ${status_pin}
binary_sensor:
- platform: tuya
id: tuya_presence
sensor_datapoint: 101
sensor:
- platform: tuya
id: tuya_light_intensity
sensor_datapoint: 103
number:
- platform: tuya
id: tuya_far_detection
number_datapoint: 109
min_value: 0
max_value: 600
step: 1
+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
@@ -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:
+152 -141
View File
@@ -1,143 +1,154 @@
{
"tests/integration/test_action_concurrent_reentry.py": 57.91,
"tests/integration/test_addressable_light_transition.py": 21.25,
"tests/integration/test_alarm_control_panel_state_transitions.py": 70.71,
"tests/integration/test_api_action_metadata.py": 66.6,
"tests/integration/test_api_action_responses.py": 36.1,
"tests/integration/test_api_action_timeout.py": 68.86,
"tests/integration/test_api_conditional_memory.py": 15.48,
"tests/integration/test_api_custom_services.py": 18.77,
"tests/integration/test_api_get_time_response_timezone.py": 21.08,
"tests/integration/test_api_homeassistant.py": 65.59,
"tests/integration/test_api_homeassistant_action_no_subscriber.py": 18.44,
"tests/integration/test_api_homeassistant_binary_sensor_initial_state.py": 15.05,
"tests/integration/test_api_list_entities_backpressure.py": 13.88,
"tests/integration/test_api_message_size_batching.py": 29.98,
"tests/integration/test_api_reboot_timeout.py": 16.05,
"tests/integration/test_api_string_lambda.py": 15.31,
"tests/integration/test_api_vv_logging.py": 19.28,
"tests/integration/test_api_zero_psk_provisioning.py": 31.5,
"tests/integration/test_areas_and_devices.py": 24.95,
"tests/integration/test_automation_wait_actions.py": 20.92,
"tests/integration/test_automations.py": 35.19,
"tests/integration/test_batch_delay_zero_rapid_transitions.py": 17.99,
"tests/integration/test_binary_sensor_autorepeat_filter.py": 20.39,
"tests/integration/test_binary_sensor_invalidate_state.py": 18.41,
"tests/integration/test_blocking_warning_log_time_not_charged_to_next_operation.py": 24.69,
"tests/integration/test_build_info.py": 18.7,
"tests/integration/test_camera_mock.py": 16.23,
"tests/integration/test_climate_control_action.py": 21.14,
"tests/integration/test_climate_custom_modes.py": 20.74,
"tests/integration/test_continuation_actions.py": 16.81,
"tests/integration/test_cover_control_action.py": 20.34,
"tests/integration/test_crc8_helper.py": 9.36,
"tests/integration/test_device_id_in_state.py": 44.67,
"tests/integration/test_duplicate_entities.py": 23.58,
"tests/integration/test_entity_icon.py": 34.35,
"tests/integration/test_fan_turn_on_action.py": 24.23,
"tests/integration/test_fnv1_hash_object_id.py": 16.21,
"tests/integration/test_fnv1a_hash.py": 13.38,
"tests/integration/test_gpio_expander_cache.py": 13.06,
"tests/integration/test_host_logger_thread_safety.py": 23.66,
"tests/integration/test_host_mode_basic.py": 8.01,
"tests/integration/test_host_mode_batch_delay.py": 21.0,
"tests/integration/test_host_mode_climate_basic_state.py": 22.14,
"tests/integration/test_host_mode_climate_control.py": 19.39,
"tests/integration/test_host_mode_empty_string_options.py": 21.76,
"tests/integration/test_host_mode_entity_fields.py": 29.61,
"tests/integration/test_host_mode_fan_preset.py": 20.01,
"tests/integration/test_host_mode_many_entities.py": 39.08,
"tests/integration/test_host_mode_many_entities_multiple_connections.py": 23.92,
"tests/integration/test_host_mode_noise_encryption.py": 42.42,
"tests/integration/test_host_mode_reconnect.py": 3.41,
"tests/integration/test_host_mode_sensor.py": 22.96,
"tests/integration/test_host_ota.py": 29.5,
"tests/integration/test_host_preferences.py": 16.06,
"tests/integration/test_host_preferences_suspend_resume.py": 18.71,
"tests/integration/test_improv_serial_uart.py": 20.22,
"tests/integration/test_large_message_batching.py": 26.56,
"tests/integration/test_legacy_area.py": 22.72,
"tests/integration/test_legacy_climate_compat.py": 14.13,
"tests/integration/test_legacy_fan_compat.py": 14.33,
"tests/integration/test_light_automations.py": 18.81,
"tests/integration/test_light_binary_effect_off_phase.py": 8.38,
"tests/integration/test_light_calls.py": 21.88,
"tests/integration/test_light_constant_brightness.py": 59.45,
"tests/integration/test_light_control_action.py": 31.91,
"tests/integration/test_light_dim_relative_action.py": 14.43,
"tests/integration/test_light_effect_zero_brightness.py": 25.05,
"tests/integration/test_light_initial_state.py": 18.97,
"tests/integration/test_light_toggle_action.py": 17.44,
"tests/integration/test_lock_automations.py": 18.9,
"tests/integration/test_logger_buffered_recursion_guard.py": 18.2,
"tests/integration/test_loop_disable_enable.py": 63.35,
"tests/integration/test_loop_interval_decoupling.py": 17.7,
"tests/integration/test_loop_interval_default_not_pulled_forward.py": 21.56,
"tests/integration/test_micros_to_millis.py": 15.89,
"tests/integration/test_multi_click_trigger.py": 17.23,
"tests/integration/test_multi_device_preferences.py": 19.4,
"tests/integration/test_noise_encryption_key_protection.py": 72.59,
"tests/integration/test_object_id_api_verification.py": 19.22,
"tests/integration/test_object_id_friendly_name_no_mac_suffix.py": 16.77,
"tests/integration/test_object_id_no_friendly_name.py": 45.8,
"tests/integration/test_online_image_auto_detects_image_bmp_mime.py": 86.73,
"tests/integration/test_online_image_auto_detects_redirected_image_bmp_mime.py": 40.4,
"tests/integration/test_online_image_bmp.py": 37.24,
"tests/integration/test_oversized_payloads.py": 55.75,
"tests/integration/test_preference_key_stability.py": 25.49,
"tests/integration/test_runtime_stats.py": 29.81,
"tests/integration/test_safe_mode_loop_runs.py": 6.26,
"tests/integration/test_scheduler_blocking_warning.py": 37.98,
"tests/integration/test_scheduler_bulk_cleanup.py": 18.67,
"tests/integration/test_scheduler_defer_cancel.py": 18.46,
"tests/integration/test_scheduler_defer_cancel_regular.py": 16.34,
"tests/integration/test_scheduler_defer_fifo_simple.py": 18.26,
"tests/integration/test_scheduler_defer_stress.py": 17.74,
"tests/integration/test_scheduler_heap_stress.py": 3.89,
"tests/integration/test_scheduler_internal_id_no_collision.py": 20.01,
"tests/integration/test_scheduler_interval_reschedule.py": 16.29,
"tests/integration/test_scheduler_interval_zero_coerced.py": 16.09,
"tests/integration/test_scheduler_null_name.py": 14.69,
"tests/integration/test_scheduler_numeric_id_test.py": 17.08,
"tests/integration/test_scheduler_pool.py": 19.88,
"tests/integration/test_scheduler_rapid_cancellation.py": 4.42,
"tests/integration/test_scheduler_recursive_timeout.py": 4.3,
"tests/integration/test_scheduler_removed_item_race.py": 15.49,
"tests/integration/test_scheduler_self_keyed.py": 25.77,
"tests/integration/test_scheduler_simultaneous_callbacks.py": 14.84,
"tests/integration/test_scheduler_string_test.py": 15.42,
"tests/integration/test_script_array_params.py": 12.73,
"tests/integration/test_script_delay_params.py": 12.69,
"tests/integration/test_script_queued.py": 20.38,
"tests/integration/test_script_queued_idle_loop.py": 25.06,
"tests/integration/test_script_wait_on_boot.py": 15.67,
"tests/integration/test_select_stringref_trigger.py": 19.48,
"tests/integration/test_sensor_filters_delta.py": 27.62,
"tests/integration/test_sensor_filters_ring_buffer.py": 20.27,
"tests/integration/test_sensor_filters_sliding_window.py": 56.28,
"tests/integration/test_sensor_filters_value_list.py": 20.6,
"tests/integration/test_sensor_timeout_filter.py": 22.21,
"tests/integration/test_socket_wake_gate_tcp.py": 16.37,
"tests/integration/test_status_flags.py": 29.68,
"tests/integration/test_strftime_to.py": 17.42,
"tests/integration/test_syslog.py": 18.39,
"tests/integration/test_template_alarm_control_panel_many_sensors.py": 25.61,
"tests/integration/test_template_text_save.py": 19.16,
"tests/integration/test_text_command.py": 16.43,
"tests/integration/test_text_sensor_raw_state.py": 17.19,
"tests/integration/test_uart_mock_ld2410.py": 37.0,
"tests/integration/test_uart_mock_ld2412.py": 40.82,
"tests/integration/test_uart_mock_ld2420.py": 32.7,
"tests/integration/test_uart_mock_ld2450.py": 32.84,
"tests/integration/test_uart_mock_modbus.py": 548.87,
"tests/integration/test_udp.py": 16.67,
"tests/integration/test_use_address_runtime.py": 27.26,
"tests/integration/test_valve_control_action.py": 24.58,
"tests/integration/test_varint_five_byte_device_id.py": 22.5,
"tests/integration/test_wait_until_mid_loop_timing.py": 22.05,
"tests/integration/test_wait_until_on_boot.py": 10.37,
"tests/integration/test_wait_until_ordering.py": 18.23,
"tests/integration/test_wait_until_reentrant_restart.py": 19.35,
"tests/integration/test_wake_loop_forces_phase_b.py": 17.83,
"tests/integration/test_water_heater_template.py": 25.7
"tests/integration/test_action_concurrent_reentry.py": 30.48,
"tests/integration/test_addressable_light_transition.py": 42.1,
"tests/integration/test_alarm_control_panel_state_transitions.py": 35.76,
"tests/integration/test_api_action_metadata.py": 22.35,
"tests/integration/test_api_action_responses.py": 30.31,
"tests/integration/test_api_action_timeout.py": 34.73,
"tests/integration/test_api_conditional_memory.py": 18.35,
"tests/integration/test_api_custom_services.py": 15.99,
"tests/integration/test_api_get_time_response_timezone.py": 24.21,
"tests/integration/test_api_homeassistant.py": 33.77,
"tests/integration/test_api_homeassistant_action_no_subscriber.py": 20.8,
"tests/integration/test_api_homeassistant_binary_sensor_initial_state.py": 23.55,
"tests/integration/test_api_list_entities_backpressure.py": 23.04,
"tests/integration/test_api_message_size_batching.py": 27.31,
"tests/integration/test_api_reboot_timeout.py": 29.32,
"tests/integration/test_api_string_lambda.py": 14.9,
"tests/integration/test_api_vv_logging.py": 26.25,
"tests/integration/test_api_zero_psk_provisioning.py": 38.19,
"tests/integration/test_areas_and_devices.py": 27.52,
"tests/integration/test_automation_wait_actions.py": 24.25,
"tests/integration/test_automations.py": 36.02,
"tests/integration/test_batch_delay_zero_rapid_transitions.py": 18.46,
"tests/integration/test_binary_sensor_autorepeat_filter.py": 17.47,
"tests/integration/test_binary_sensor_invalidate_state.py": 14.79,
"tests/integration/test_blocking_warning_log_time_not_charged_to_next_operation.py": 21.52,
"tests/integration/test_build_info.py": 21.42,
"tests/integration/test_camera_mock.py": 17.02,
"tests/integration/test_climate_control_action.py": 26.56,
"tests/integration/test_climate_custom_modes.py": 18.82,
"tests/integration/test_continuation_actions.py": 20.39,
"tests/integration/test_cover_control_action.py": 19.91,
"tests/integration/test_crc8_helper.py": 16.73,
"tests/integration/test_device_id_in_state.py": 58.41,
"tests/integration/test_duplicate_entities.py": 30.76,
"tests/integration/test_entity_icon.py": 25.34,
"tests/integration/test_fan_turn_on_action.py": 23.64,
"tests/integration/test_fnv1_hash_object_id.py": 25.44,
"tests/integration/test_fnv1a_hash.py": 20.85,
"tests/integration/test_gpio_expander_cache.py": 14.42,
"tests/integration/test_host_logger_thread_safety.py": 21.31,
"tests/integration/test_host_mode_basic.py": 2.65,
"tests/integration/test_host_mode_batch_delay.py": 22.21,
"tests/integration/test_host_mode_climate_basic_state.py": 27.12,
"tests/integration/test_host_mode_climate_control.py": 21.57,
"tests/integration/test_host_mode_empty_string_options.py": 27.17,
"tests/integration/test_host_mode_entity_fields.py": 30.1,
"tests/integration/test_host_mode_fan_preset.py": 17.55,
"tests/integration/test_host_mode_many_entities.py": 38.98,
"tests/integration/test_host_mode_many_entities_multiple_connections.py": 23.82,
"tests/integration/test_host_mode_noise_encryption.py": 39.84,
"tests/integration/test_host_mode_reconnect.py": 13.1,
"tests/integration/test_host_mode_sensor.py": 22.17,
"tests/integration/test_host_ota.py": 92.05,
"tests/integration/test_host_preferences.py": 20.29,
"tests/integration/test_host_preferences_suspend_resume.py": 15.02,
"tests/integration/test_improv_serial_uart.py": 30.15,
"tests/integration/test_large_message_batching.py": 25.84,
"tests/integration/test_legacy_area.py": 21.24,
"tests/integration/test_legacy_climate_compat.py": 17.34,
"tests/integration/test_legacy_fan_compat.py": 22.6,
"tests/integration/test_light_automations.py": 29.13,
"tests/integration/test_light_binary_effect_off_phase.py": 33.99,
"tests/integration/test_light_calls.py": 26.81,
"tests/integration/test_light_constant_brightness.py": 25.0,
"tests/integration/test_light_control_action.py": 25.57,
"tests/integration/test_light_dim_relative_action.py": 21.4,
"tests/integration/test_light_effect_zero_brightness.py": 19.65,
"tests/integration/test_light_initial_state.py": 17.58,
"tests/integration/test_light_toggle_action.py": 28.28,
"tests/integration/test_lock_automations.py": 23.3,
"tests/integration/test_logger_buffered_recursion_guard.py": 22.96,
"tests/integration/test_loop_disable_enable.py": 16.18,
"tests/integration/test_loop_interval_decoupling.py": 25.19,
"tests/integration/test_loop_interval_default_not_pulled_forward.py": 20.59,
"tests/integration/test_lvgl_headless_render.py": 87.78,
"tests/integration/test_micros_to_millis.py": 18.73,
"tests/integration/test_multi_click_trigger.py": 24.2,
"tests/integration/test_multi_device_preferences.py": 20.52,
"tests/integration/test_noise_encryption_key_protection.py": 19.1,
"tests/integration/test_object_id_api_verification.py": 26.24,
"tests/integration/test_object_id_friendly_name_no_mac_suffix.py": 14.88,
"tests/integration/test_object_id_no_friendly_name.py": 61.27,
"tests/integration/test_online_image_auto_detects_image_bmp_mime.py": 82.32,
"tests/integration/test_online_image_auto_detects_redirected_image_bmp_mime.py": 46.03,
"tests/integration/test_online_image_bmp.py": 34.21,
"tests/integration/test_oversized_payloads.py": 62.75,
"tests/integration/test_preference_key_stability.py": 26.8,
"tests/integration/test_runtime_stats.py": 28.26,
"tests/integration/test_safe_mode_loop_runs.py": 18.14,
"tests/integration/test_scheduler_blocking_warning.py": 28.7,
"tests/integration/test_scheduler_bulk_cleanup.py": 20.73,
"tests/integration/test_scheduler_defer_cancel.py": 22.99,
"tests/integration/test_scheduler_defer_cancel_regular.py": 21.97,
"tests/integration/test_scheduler_defer_fifo_simple.py": 24.15,
"tests/integration/test_scheduler_defer_stress.py": 23.11,
"tests/integration/test_scheduler_heap_stress.py": 20.2,
"tests/integration/test_scheduler_internal_id_no_collision.py": 23.75,
"tests/integration/test_scheduler_interval_reschedule.py": 15.32,
"tests/integration/test_scheduler_interval_zero_coerced.py": 20.1,
"tests/integration/test_scheduler_null_name.py": 17.43,
"tests/integration/test_scheduler_numeric_id_test.py": 25.51,
"tests/integration/test_scheduler_pool.py": 24.22,
"tests/integration/test_scheduler_rapid_cancellation.py": 24.01,
"tests/integration/test_scheduler_recursive_timeout.py": 22.94,
"tests/integration/test_scheduler_removed_item_race.py": 23.07,
"tests/integration/test_scheduler_self_keyed.py": 18.43,
"tests/integration/test_scheduler_simultaneous_callbacks.py": 21.99,
"tests/integration/test_scheduler_string_test.py": 17.27,
"tests/integration/test_script_array_params.py": 4.59,
"tests/integration/test_script_delay_params.py": 22.46,
"tests/integration/test_script_queued.py": 25.24,
"tests/integration/test_script_queued_idle_loop.py": 5.04,
"tests/integration/test_script_wait_on_boot.py": 21.77,
"tests/integration/test_sdl_headless_screenshot.py": 19.23,
"tests/integration/test_select_stringref_trigger.py": 19.31,
"tests/integration/test_sensor_filters_delta.py": 25.92,
"tests/integration/test_sensor_filters_ring_buffer.py": 22.39,
"tests/integration/test_sensor_filters_sliding_window.py": 57.93,
"tests/integration/test_sensor_filters_value_list.py": 20.32,
"tests/integration/test_sensor_timeout_filter.py": 25.35,
"tests/integration/test_snapshot_display.py": 19.7,
"tests/integration/test_socket_wake_gate_tcp.py": 14.5,
"tests/integration/test_status_flags.py": 33.83,
"tests/integration/test_strftime_to.py": 17.64,
"tests/integration/test_syslog.py": 24.49,
"tests/integration/test_template_alarm_control_panel_many_sensors.py": 24.81,
"tests/integration/test_template_climate_basic.py": 15.28,
"tests/integration/test_template_climate_custom_modes.py": 25.07,
"tests/integration/test_template_climate_nonoptimistic.py": 24.25,
"tests/integration/test_template_climate_on_control_ordering.py": 24.09,
"tests/integration/test_template_climate_publish_all_fields.py": 17.78,
"tests/integration/test_template_climate_sensor_push.py": 17.42,
"tests/integration/test_template_climate_set_actions.py": 23.63,
"tests/integration/test_template_climate_two_point_temperature.py": 25.19,
"tests/integration/test_template_text_save.py": 17.88,
"tests/integration/test_text_command.py": 22.71,
"tests/integration/test_text_sensor_raw_state.py": 25.17,
"tests/integration/test_uart_mock_ld2410.py": 58.15,
"tests/integration/test_uart_mock_ld2412.py": 61.14,
"tests/integration/test_uart_mock_ld2420.py": 33.87,
"tests/integration/test_uart_mock_ld2450.py": 26.06,
"tests/integration/test_uart_mock_modbus.py": 391.79,
"tests/integration/test_udp.py": 7.38,
"tests/integration/test_use_address_runtime.py": 24.09,
"tests/integration/test_valve_control_action.py": 23.22,
"tests/integration/test_varint_five_byte_device_id.py": 17.93,
"tests/integration/test_wait_until_mid_loop_timing.py": 22.26,
"tests/integration/test_wait_until_on_boot.py": 17.46,
"tests/integration/test_wait_until_ordering.py": 11.89,
"tests/integration/test_wait_until_reentrant_restart.py": 22.88,
"tests/integration/test_wake_loop_forces_phase_b.py": 16.6,
"tests/integration/test_water_heater_template.py": 19.66
}
@@ -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)]
+17 -17
View File
@@ -35,8 +35,8 @@ def _load_script():
def test_spec_key_collapses_destinations() -> None:
"""Two specs delivering one package share a directory and one key."""
mod = _load_script()
assert mod.spec_key("esphome/noise-c @ 0.1.24") == "noise-c"
assert mod.spec_key("esphome/noise-c@0.1.24") == "noise-c"
assert mod.spec_key("esphome/noise-c @ 0.1.26") == "noise-c"
assert mod.spec_key("esphome/noise-c@0.1.26") == "noise-c"
assert mod.spec_key("ESP32Async/AsyncTCP @ ^3.4.10") == mod.spec_key(
"esp32async/asynctcp @ 3.5.0"
)
@@ -54,23 +54,23 @@ def test_parse_specs_and_cli_args(tmp_path: Path) -> None:
"[env:a]\n"
"platform = fake/platform@1\n"
"lib_deps =\n"
" esphome/noise-c @ 0.1.24\n"
" esphome/noise-c @ 0.1.26\n"
" ${common.lib_deps}\n"
" internal_lib\n"
"[env:b]\n"
"lib_deps =\n"
" esphome/noise-c @ 0.1.24\n"
" esphome/noise-c @ 0.1.26\n"
)
mod = _load_script()
args = Namespace(libraries=True, platforms=True, tools=False)
libs, platforms, tools = mod.parse_specs(str(ini), args)
# exact-string duplicates collapse; distinct version pins survive
assert libs == ["esphome/noise-c @ 0.1.24"]
assert libs == ["esphome/noise-c @ 0.1.26"]
assert platforms == ["fake/platform@1"]
assert tools == []
assert mod.build_cli_args(libs, platforms, tools) == [
"-l",
"esphome/noise-c @ 0.1.24",
"esphome/noise-c @ 0.1.26",
"-p",
"fake/platform@1",
]
@@ -162,13 +162,13 @@ def test_parallel_install_behavior(tmp_path: Path) -> None:
mod.parallel_install(
cls,
[
"esphome/noise-c @ 0.1.24",
"esphome/noise-c @ 0.1.24",
"esphome/noise-c @ 0.1.26",
"esphome/noise-c @ 0.1.26",
"esphome/already @ 1.0",
"https://x/framework.tar.xz",
],
)
assert cls.calls == ["esphome/noise-c @ 0.1.24"]
assert cls.calls == ["esphome/noise-c @ 0.1.26"]
assert cls.lock_events == ["lock", "unlock"]
@@ -205,7 +205,7 @@ def test_parallel_install_runs_dependency_waves(tmp_path: Path) -> None:
mod = _load_script()
cls = _reset_fake(str(tmp_path))
cls.deps = {
"esphome/noise-c @ 0.1.24": [
"esphome/noise-c @ 0.1.26": [
{"owner": "esphome", "name": "libsodium", "version": "^1.0"},
{"name": "SPI"},
],
@@ -213,12 +213,12 @@ def test_parallel_install_runs_dependency_waves(tmp_path: Path) -> None:
{"owner": "esphome", "name": "libsodium", "version": "^1.0"},
],
}
mod.parallel_install(cls, ["esphome/noise-c @ 0.1.24", "esphome/wg @ 1.0"])
mod.parallel_install(cls, ["esphome/noise-c @ 0.1.26", "esphome/wg @ 1.0"])
assert len(cls.calls) == 3 # the shared dep installs exactly once
assert {mod.spec_key(c) for c in cls.calls} == {"noise-c", "wg", "libsodium"}
# Wave-1 strings carry no compatibility; the dependency wave does
compats = dict(cls.compat_calls)
assert compats["esphome/noise-c @ 0.1.24"] is None
assert compats["esphome/noise-c @ 0.1.26"] is None
dep_compat = next(v for k, v in cls.compat_calls if "libsodium" in k)
assert dep_compat is not None # mirrors pio's install_dependency
@@ -229,11 +229,11 @@ def test_dependency_wave_excludes_url_specs(tmp_path: Path) -> None:
mod = _load_script()
cls = _reset_fake(str(tmp_path))
cls.deps = {
"esphome/noise-c @ 0.1.24": [
"esphome/noise-c @ 0.1.26": [
{"name": "vendored", "version": "https://github.com/x/y.git"},
],
}
mod.parallel_install(cls, ["esphome/noise-c @ 0.1.24"])
mod.parallel_install(cls, ["esphome/noise-c @ 0.1.26"])
assert {mod.spec_key(c) for c in cls.calls} == {"noise-c"}
@@ -348,13 +348,13 @@ def test_warm_store_still_walks_dependencies(tmp_path: Path) -> None:
"""Already-installed top-level packages still feed the dependency
wave; a warm store can be missing a transitive dep."""
mod = _load_script()
cls = _reset_fake(str(tmp_path), installed={"esphome/noise-c @ 0.1.24"})
cls = _reset_fake(str(tmp_path), installed={"esphome/noise-c @ 0.1.26"})
cls.deps = {
"esphome/noise-c @ 0.1.24": [
"esphome/noise-c @ 0.1.26": [
{"owner": "esphome", "name": "libsodium", "version": "^1.0"},
],
}
mod.parallel_install(cls, ["esphome/noise-c @ 0.1.24"])
mod.parallel_install(cls, ["esphome/noise-c @ 0.1.26"])
assert [mod.spec_key(c) for c in cls.calls] == ["libsodium"]
+562
View File
@@ -0,0 +1,562 @@
"""Tests for script/setup.py."""
import importlib.util
import os
from pathlib import Path, PurePosixPath, PureWindowsPath
import runpy
import sys
from types import ModuleType
from unittest.mock import Mock, call, patch
import pytest
_SCRIPT = Path(__file__).parents[2] / "script" / "setup.py"
def _load_module() -> ModuleType:
spec = importlib.util.spec_from_file_location("script_setup", _SCRIPT)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
@pytest.fixture
def script_setup() -> ModuleType:
"""Fresh import of script/setup.py, isolated from other tests."""
return _load_module()
# --- bin_dir / venv_python / activate_hint -----------------------------------
def test_bin_dir_matches_host_layout(script_setup: ModuleType, tmp_path: Path) -> None:
"""The venv scheme resolves to Scripts on Windows and bin everywhere else."""
expected = "Scripts" if os.name == "nt" else "bin"
assert script_setup.bin_dir(tmp_path) == tmp_path / expected
# Both flavours are exercised on every host. Pure paths are used because a real
# Path refuses to change flavour: PosixPath cannot be built on Windows, and
# WindowsPath cannot be built on Unix.
def test_venv_python_posix(script_setup: ModuleType, tmp_path: Path) -> None:
with (
patch.object(
script_setup, "bin_dir", return_value=PurePosixPath("/x/venv/bin")
),
patch.object(script_setup.os, "name", "posix"),
):
result = script_setup.venv_python(tmp_path)
assert result == PurePosixPath("/x/venv/bin/python")
def test_venv_python_nt(script_setup: ModuleType, tmp_path: Path) -> None:
with (
patch.object(
script_setup, "bin_dir", return_value=PureWindowsPath(r"C:\x\venv\Scripts")
),
patch.object(script_setup.os, "name", "nt"),
):
result = script_setup.venv_python(tmp_path)
assert result == PureWindowsPath(r"C:\x\venv\Scripts\python.exe")
def test_activate_hint_posix(script_setup: ModuleType) -> None:
with (
patch.object(script_setup, "ROOT", PurePosixPath("/x")),
patch.object(
script_setup, "bin_dir", return_value=PurePosixPath("/x/venv/bin")
),
patch.object(script_setup.os, "name", "posix"),
):
hint = script_setup.activate_hint()
assert hint == "source venv/bin/activate"
def test_activate_hint_nt(script_setup: ModuleType) -> None:
with (
patch.object(script_setup, "ROOT", PureWindowsPath(r"C:\x")),
patch.object(
script_setup, "bin_dir", return_value=PureWindowsPath(r"C:\x\venv\Scripts")
),
patch.object(script_setup.os, "name", "nt"),
):
hint = script_setup.activate_hint()
# The nt branch returns str(activate) as-is, skipping the "source " prefix.
assert hint == r"venv\Scripts\activate"
# --- run -----------------------------------------------------------------
def test_run_success(script_setup: ModuleType) -> None:
with patch.object(
script_setup.subprocess, "run", return_value=Mock(returncode=0)
) as mock_run:
script_setup.run(["echo", "hi"])
mock_run.assert_called_once_with(
["echo", "hi"], cwd=script_setup.ROOT, env=None, check=False
)
def test_run_failure_raises_system_exit_with_code(
script_setup: ModuleType, capsys: pytest.CaptureFixture[str]
) -> None:
with (
patch.object(script_setup.subprocess, "run", return_value=Mock(returncode=7)),
pytest.raises(SystemExit) as excinfo,
):
script_setup.run(["false"])
assert excinfo.value.code == 7
assert "Failed with exit code 7: false" in capsys.readouterr().err
# --- git_output ------------------------------------------------------------
def test_git_output_success_strips_stdout(script_setup: ModuleType) -> None:
with patch.object(
script_setup.subprocess,
"run",
return_value=Mock(returncode=0, stdout=" /repo/.git \n"),
) as mock_run:
result = script_setup.git_output("rev-parse", "--absolute-git-dir")
assert result == "/repo/.git"
mock_run.assert_called_once_with(
["git", "rev-parse", "--absolute-git-dir"],
cwd=script_setup.ROOT,
capture_output=True,
text=True,
check=False,
)
def test_git_output_nonzero_returncode_is_empty(script_setup: ModuleType) -> None:
with patch.object(
script_setup.subprocess,
"run",
return_value=Mock(returncode=1, stdout="whatever"),
):
assert script_setup.git_output("status") == ""
def test_git_output_oserror_is_empty(script_setup: ModuleType) -> None:
with patch.object(script_setup.subprocess, "run", side_effect=OSError("no git")):
assert script_setup.git_output("status") == ""
# --- create_venv -----------------------------------------------------------
def test_create_venv_uses_uv_when_present(
script_setup: ModuleType, tmp_path: Path
) -> None:
venv = tmp_path / "venv"
with (
patch.object(script_setup.shutil, "which", return_value="/usr/bin/uv"),
patch.object(
script_setup.subprocess, "run", return_value=Mock(returncode=0)
) as mock_run,
):
script_setup.create_venv(venv)
mock_run.assert_called_once_with(
["/usr/bin/uv", "venv", "--clear", "--seed", str(venv)],
cwd=script_setup.ROOT,
env=None,
check=False,
)
def test_create_venv_falls_back_to_venv_module(
script_setup: ModuleType, tmp_path: Path
) -> None:
venv = tmp_path / "venv"
with (
patch.object(script_setup.shutil, "which", return_value=None),
patch.object(
script_setup.subprocess, "run", return_value=Mock(returncode=0)
) as mock_run,
):
script_setup.create_venv(venv)
mock_run.assert_called_once_with(
[sys.executable, "-m", "venv", "--clear", str(venv)],
cwd=script_setup.ROOT,
env=None,
check=False,
)
# --- venv_environment --------------------------------------------------------
def test_venv_environment_sets_virtual_env_and_prepends_path(
script_setup: ModuleType, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
venv = tmp_path / "venv"
monkeypatch.setenv("PYTHONHOME", "/somewhere")
monkeypatch.setenv("PATH", "/usr/bin:/bin")
env = script_setup.venv_environment(venv)
assert env["VIRTUAL_ENV"] == str(venv)
assert "PYTHONHOME" not in env
expected_prefix = str(script_setup.bin_dir(venv)) + os.pathsep
assert env["PATH"] == expected_prefix + "/usr/bin:/bin"
def test_venv_environment_path_fallback_when_unset(
script_setup: ModuleType, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
venv = tmp_path / "venv"
monkeypatch.delenv("PATH", raising=False)
env = script_setup.venv_environment(venv)
# No trailing separator: an empty PATH entry means "search the cwd".
assert env["PATH"] == str(script_setup.bin_dir(venv))
# --- find_uv -----------------------------------------------------------------
def test_find_uv_found_immediately(script_setup: ModuleType, tmp_path: Path) -> None:
venv = tmp_path / "venv"
env = {"PATH": "/usr/bin"}
with (
patch.object(script_setup.shutil, "which", return_value="/usr/bin/uv"),
patch.object(script_setup.subprocess, "run") as mock_run,
):
result = script_setup.find_uv(venv, env)
assert result == "/usr/bin/uv"
mock_run.assert_not_called()
def test_find_uv_installed_then_found(script_setup: ModuleType, tmp_path: Path) -> None:
venv = tmp_path / "venv"
env = {"PATH": "/usr/bin"}
with (
patch.object(script_setup.shutil, "which", side_effect=[None, "/usr/bin/uv"]),
patch.object(
script_setup.subprocess, "run", return_value=Mock(returncode=0)
) as mock_run,
):
result = script_setup.find_uv(venv, env)
assert result == "/usr/bin/uv"
mock_run.assert_called_once_with(
[str(script_setup.venv_python(venv)), "-m", "pip", "install", "uv"],
cwd=script_setup.ROOT,
env=env,
check=False,
)
def test_find_uv_still_missing_raises_system_exit(
script_setup: ModuleType, tmp_path: Path
) -> None:
venv = tmp_path / "venv"
env = {"PATH": "/usr/bin"}
with (
patch.object(script_setup.shutil, "which", side_effect=[None, None]),
patch.object(script_setup.subprocess, "run", return_value=Mock(returncode=0)),
pytest.raises(SystemExit, match="uv could not be installed"),
):
script_setup.find_uv(venv, env)
# --- install_dependencies -----------------------------------------------------
def test_install_dependencies_installs_setuptools_then_project(
script_setup: ModuleType, tmp_path: Path
) -> None:
venv = tmp_path / "venv"
env = {"PATH": "/usr/bin"}
with (
patch.object(script_setup.shutil, "which", return_value="/usr/bin/uv"),
patch.object(
script_setup.subprocess, "run", return_value=Mock(returncode=0)
) as mock_run,
):
script_setup.install_dependencies(venv, env)
assert mock_run.call_args_list == [
call(
["/usr/bin/uv", "pip", "install", "setuptools", "wheel"],
cwd=script_setup.ROOT,
env=env,
check=False,
),
call(
[
"/usr/bin/uv",
"pip",
"install",
"-e",
".[dev,test]",
"--config-settings",
"editable_mode=compat",
],
cwd=script_setup.ROOT,
env=env,
check=False,
),
]
# --- install_git_hooks ---------------------------------------------------------
def _fake_git_output(git_dir: str, common_dir: str):
def _run(*args: str) -> str:
if "--absolute-git-dir" in args:
return git_dir
return common_dir
return _run
def test_install_git_hooks_returns_early_when_git_dir_empty(
script_setup: ModuleType,
) -> None:
env = {"PATH": "/usr/bin"}
with (
patch.object(
script_setup, "git_output", side_effect=_fake_git_output("", "/repo/.git")
),
patch.object(script_setup.subprocess, "run") as mock_run,
):
script_setup.install_git_hooks(env)
mock_run.assert_not_called()
def test_install_git_hooks_returns_early_when_common_dir_empty(
script_setup: ModuleType,
) -> None:
env = {"PATH": "/usr/bin"}
with (
patch.object(
script_setup, "git_output", side_effect=_fake_git_output("/repo/.git", "")
),
patch.object(script_setup.subprocess, "run") as mock_run,
):
script_setup.install_git_hooks(env)
mock_run.assert_not_called()
def test_install_git_hooks_returns_early_for_worktree(
script_setup: ModuleType,
) -> None:
"""A worktree's git-dir differs from the shared common-dir."""
env = {"PATH": "/usr/bin"}
with (
patch.object(
script_setup,
"git_output",
side_effect=_fake_git_output("/repo/.git/worktrees/wt", "/repo/.git"),
),
patch.object(script_setup.subprocess, "run") as mock_run,
):
script_setup.install_git_hooks(env)
mock_run.assert_not_called()
def test_install_git_hooks_missing_prek_raises_system_exit(
script_setup: ModuleType,
) -> None:
env = {"PATH": "/usr/bin"}
with (
patch.object(
script_setup,
"git_output",
side_effect=_fake_git_output("/repo/.git", "/repo/.git"),
),
patch.object(script_setup.shutil, "which", return_value=None),
patch.object(script_setup.subprocess, "run") as mock_run,
pytest.raises(SystemExit, match="prek was not installed"),
):
script_setup.install_git_hooks(env)
mock_run.assert_not_called()
def test_install_git_hooks_happy_path_installs_hook(
script_setup: ModuleType, tmp_path: Path
) -> None:
env = {"PATH": "/usr/bin"}
common_dir = tmp_path / "repo" / ".git"
hooks_dir = common_dir / "hooks"
hooks_dir.mkdir(parents=True)
source_hook = tmp_path / "post-checkout"
source_hook.write_text("#!/bin/sh\necho post-checkout\n")
with (
patch.object(script_setup, "POST_CHECKOUT_HOOK", source_hook),
patch.object(
script_setup,
"git_output",
side_effect=_fake_git_output(str(common_dir), str(common_dir)),
),
patch.object(script_setup.shutil, "which", return_value="/usr/bin/prek"),
patch.object(
script_setup.subprocess, "run", return_value=Mock(returncode=0)
) as mock_run,
):
script_setup.install_git_hooks(env)
mock_run.assert_called_once_with(
["/usr/bin/prek", "install", "--overwrite"],
cwd=script_setup.ROOT,
env=env,
check=False,
)
installed = hooks_dir / "post-checkout"
assert installed.read_text() == source_hook.read_text()
if os.name != "nt":
# Windows has no POSIX permission bits for chmod to set.
assert (installed.stat().st_mode & 0o777) == 0o755
def test_install_git_hooks_skips_copy_when_hooks_dir_missing(
script_setup: ModuleType, tmp_path: Path
) -> None:
"""The prek install still runs when the hooks directory does not exist."""
env = {"PATH": "/usr/bin"}
common_dir = tmp_path / "repo" / ".git"
common_dir.mkdir(parents=True) # no "hooks" subdirectory created
with (
patch.object(
script_setup,
"git_output",
side_effect=_fake_git_output(str(common_dir), str(common_dir)),
),
patch.object(script_setup.shutil, "which", return_value="/usr/bin/prek"),
patch.object(
script_setup.subprocess, "run", return_value=Mock(returncode=0)
) as mock_run,
):
script_setup.install_git_hooks(env)
mock_run.assert_called_once()
assert not (common_dir / "hooks").exists()
# --- report ------------------------------------------------------------------
def test_report_active_state(
script_setup: ModuleType, capsys: pytest.CaptureFixture
) -> None:
venv = Path("/opt/esphome-venv")
script_setup.report(script_setup.VENV_ACTIVE, venv)
out = capsys.readouterr().out
assert "Dependencies installed into the active virtual environment:" in out
assert str(venv) in out
assert "is already active in this shell" in out
def test_report_reused_state(
script_setup: ModuleType, capsys: pytest.CaptureFixture
) -> None:
script_setup.report(script_setup.VENV_REUSED, script_setup.DEFAULT_VENV)
out = capsys.readouterr().out
assert "Dependencies updated in the existing ./venv" in out
def test_report_created_state(
script_setup: ModuleType, capsys: pytest.CaptureFixture
) -> None:
script_setup.report(script_setup.VENV_CREATED, script_setup.DEFAULT_VENV)
out = capsys.readouterr().out
assert "Virtual environment created at ./venv" in out
# --- main --------------------------------------------------------------------
def test_main_raises_system_exit_when_python_too_old(
script_setup: ModuleType,
) -> None:
with (
patch.object(script_setup.sys, "version_info", (3, 11, 5)),
pytest.raises(SystemExit, match="ESPHome needs Python 3.12"),
):
script_setup.main()
def test_main_uses_active_virtual_env(
script_setup: ModuleType, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
active_venv = tmp_path / "active-venv"
monkeypatch.setenv("VIRTUAL_ENV", str(active_venv))
with (
patch.object(script_setup, "ROOT", tmp_path),
patch.object(script_setup, "create_venv") as mock_create_venv,
patch.object(script_setup, "install_dependencies") as mock_install_deps,
patch.object(script_setup, "install_git_hooks") as mock_install_hooks,
patch.object(script_setup, "report") as mock_report,
):
script_setup.main()
mock_create_venv.assert_not_called()
mock_install_deps.assert_called_once()
mock_install_hooks.assert_called_once()
mock_report.assert_called_once_with(script_setup.VENV_ACTIVE, active_venv)
assert (tmp_path / ".temp").is_dir()
def test_main_reuses_existing_venv(
script_setup: ModuleType, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.delenv("VIRTUAL_ENV", raising=False)
default_venv = tmp_path / "venv"
python_path = script_setup.venv_python(default_venv)
python_path.parent.mkdir(parents=True)
python_path.touch()
with (
patch.object(script_setup, "ROOT", tmp_path),
patch.object(script_setup, "DEFAULT_VENV", default_venv),
patch.object(script_setup, "create_venv") as mock_create_venv,
patch.object(script_setup, "install_dependencies") as mock_install_deps,
patch.object(script_setup, "install_git_hooks") as mock_install_hooks,
patch.object(script_setup, "report") as mock_report,
):
script_setup.main()
mock_create_venv.assert_not_called()
mock_install_deps.assert_called_once()
mock_install_hooks.assert_called_once()
mock_report.assert_called_once_with(script_setup.VENV_REUSED, default_venv)
assert (tmp_path / ".temp").is_dir()
def test_main_creates_new_venv(
script_setup: ModuleType, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.delenv("VIRTUAL_ENV", raising=False)
default_venv = tmp_path / "venv" # does not exist yet
with (
patch.object(script_setup, "ROOT", tmp_path),
patch.object(script_setup, "DEFAULT_VENV", default_venv),
patch.object(script_setup, "create_venv") as mock_create_venv,
patch.object(script_setup, "install_dependencies") as mock_install_deps,
patch.object(script_setup, "install_git_hooks") as mock_install_hooks,
patch.object(script_setup, "report") as mock_report,
):
script_setup.main()
mock_create_venv.assert_called_once_with(default_venv)
mock_install_deps.assert_called_once()
mock_install_hooks.assert_called_once()
mock_report.assert_called_once_with(script_setup.VENV_CREATED, default_venv)
assert (tmp_path / ".temp").is_dir()
def test_run_as_script_calls_main(tmp_path: Path) -> None:
"""The __main__ guard runs the whole flow, with every side effect stubbed."""
completed = Mock(returncode=0, stdout="")
with (
patch("subprocess.run", return_value=completed) as mock_run,
patch("shutil.which", return_value="/usr/bin/uv"),
patch("pathlib.Path.mkdir") as mock_mkdir,
patch.dict(os.environ, {"VIRTUAL_ENV": str(tmp_path / "env")}),
):
runpy.run_path(str(_SCRIPT), run_name="__main__")
# The dependency install ran, and git reported no hooks directory to touch.
assert mock_run.called
mock_mkdir.assert_called_once_with(exist_ok=True)
@@ -0,0 +1,219 @@
"""Unit tests for script/sync_dependency_versions.py."""
from pathlib import Path
import subprocess
import sys
import pytest
import yamlrocks
sys.path.insert(0, str((Path(__file__).parent / ".." / ".." / "script").resolve()))
import sync_dependency_versions as sync_mod # noqa: E402
PRECOMMIT = """\
# See https://pre-commit.com for more information
repos:
- repo: https://github.com/astral-sh/ruff-pre-commit
# Ruff version.
rev: v0.1.0
hooks:
- id: ruff
- repo: https://github.com/PyCQA/flake8
rev: 7.0.0
hooks:
- id: flake8
- repo: https://github.com/asottile/pyupgrade
rev: v3.0.0
hooks:
- id: pyupgrade
- repo: https://github.com/pre-commit/mirrors-clang-format
rev: v13.0.1
hooks:
- id: clang-format
- repo: https://github.com/adrienverge/yamllint.git
rev: v1.0.0
hooks:
- id: yamllint
- repo: local
hooks:
- id: pylint
"""
REQ_TEST = """\
pylint==4.0.8
flake8==7.1.0
ruff==0.2.0 # comment
pyupgrade==3.0.0
"""
REQ_DEV = """\
clang-format==13.0.1
yamllint==1.0.0
"""
RUFF_REPO = "https://github.com/astral-sh/ruff-pre-commit"
DUPLICATE_RUFF_BLOCK = f" - repo: {RUFF_REPO}\n rev: v0.3.0\n hooks: []\n"
EXPECTED_DRIFT = ["ruff: 0.1.0 -> 0.2.0", "flake8: 7.0.0 -> 7.1.0"]
EXPECTED_PRECOMMIT = PRECOMMIT.replace("rev: v0.1.0", "rev: v0.2.0").replace(
"rev: 7.0.0", "rev: 7.1.0"
)
@pytest.fixture
def root(tmp_path: Path) -> Path:
"""A fake checkout where ruff (v-prefixed) and flake8 (bare) have drifted."""
(tmp_path / ".pre-commit-config.yaml").write_text(PRECOMMIT)
(tmp_path / "requirements_test.txt").write_text(REQ_TEST)
(tmp_path / "requirements_dev.txt").write_text(REQ_DEV)
return tmp_path
def _load(text: str) -> object:
return yamlrocks.loads(text.encode(), option=yamlrocks.OPT_ROUND_TRIP)
@pytest.mark.parametrize(
("requirements", "expected"),
[
("prek==0.5.1 # comment\n", "0.5.1"),
("Prek==0.5.1\n", "0.5.1"),
("other==1.0\nprek==0.5.1\n", "0.5.1"),
("prek>=0.5.1\n", None),
("prek-extra==0.5.1\n", None),
("", None),
],
)
def test_read_requirement_version(requirements: str, expected: str | None) -> None:
assert sync_mod.read_requirement_version(requirements, "prek") == expected
def test_find_repo_entry() -> None:
entry = sync_mod.find_repo_entry(_load(PRECOMMIT), RUFF_REPO)
assert entry["rev"] == "v0.1.0"
@pytest.mark.parametrize(
("text", "message"),
[
("hooks: []\n", "missing key 'repos'"),
("repos:\n - rev: 1.0.0\n", "missing key 'repo'"),
(PRECOMMIT + DUPLICATE_RUFF_BLOCK, "found 2"),
("repos:\n - repo: other\n rev: 1.0.0\n", "found 0"),
],
)
def test_find_repo_entry_errors(text: str, message: str) -> None:
with pytest.raises(sync_mod.SyncError, match=message):
sync_mod.find_repo_entry(_load(text), RUFF_REPO)
@pytest.mark.parametrize(
("rev", "expected"),
[("v0.1.0", ("v", "0.1.0")), ("7.0.0", ("", "7.0.0")), ("'1.0'", ("", "1.0"))],
)
def test_current_rev(rev: str, expected: tuple[str, str]) -> None:
doc = _load(f"repos:\n - repo: {RUFF_REPO}\n rev: {rev}\n")
assert sync_mod.current_rev(doc["repos"][0], RUFF_REPO) == expected
@pytest.mark.parametrize(
("block", "message"),
[(" hooks: []\n", "has no rev"), (" rev: 1.0\n", "not a string: 1.0")],
)
def test_current_rev_errors(block: str, message: str) -> None:
doc = _load(f"repos:\n - repo: {RUFF_REPO}\n{block}")
with pytest.raises(sync_mod.SyncError, match=message):
sync_mod.current_rev(doc["repos"][0], RUFF_REPO)
def test_sync_reports_without_writing(root: Path) -> None:
assert sync_mod.sync(root, write=False) == EXPECTED_DRIFT
assert (root / ".pre-commit-config.yaml").read_text() == PRECOMMIT
def test_sync_writes_keeps_layout_and_is_idempotent(root: Path) -> None:
assert sync_mod.sync(root, write=True) == EXPECTED_DRIFT
assert (root / ".pre-commit-config.yaml").read_text() == EXPECTED_PRECOMMIT
assert sync_mod.sync(root, write=True) == []
def test_sync_does_not_touch_a_config_that_matches(root: Path) -> None:
(root / ".pre-commit-config.yaml").write_text(EXPECTED_PRECOMMIT)
before = (root / ".pre-commit-config.yaml").stat().st_mtime_ns
assert sync_mod.sync(root, write=True) == []
assert (root / ".pre-commit-config.yaml").stat().st_mtime_ns == before
def test_sync_missing_requirement_pin(root: Path) -> None:
(root / "requirements_dev.txt").write_text("")
with pytest.raises(sync_mod.SyncError, match="no 'clang-format==' pin"):
sync_mod.sync(root, write=True)
def test_sync_propagates_config_errors(root: Path) -> None:
(root / ".pre-commit-config.yaml").write_text(PRECOMMIT + DUPLICATE_RUFF_BLOCK)
with pytest.raises(sync_mod.SyncError, match="found 2"):
sync_mod.sync(root, write=True)
def test_main_check_reports_drift(
root: Path, capsys: pytest.CaptureFixture[str]
) -> None:
assert sync_mod.main(["--check", "--root", str(root)]) == 1
assert capsys.readouterr().out.splitlines() == EXPECTED_DRIFT
assert (root / ".pre-commit-config.yaml").read_text() == PRECOMMIT
def test_main_writes_then_check_is_clean(
root: Path, capsys: pytest.CaptureFixture[str]
) -> None:
assert sync_mod.main(["--root", str(root)]) == 0
assert capsys.readouterr().out.splitlines() == EXPECTED_DRIFT
assert sync_mod.main(["--check", "--root", str(root)]) == 0
assert capsys.readouterr().out == ""
def test_main_reports_sync_error(
root: Path, capsys: pytest.CaptureFixture[str]
) -> None:
(root / "requirements_dev.txt").write_text("")
assert sync_mod.main(["--root", str(root)]) == 1
assert (
"error: requirements_dev.txt: no 'clang-format==' pin"
in capsys.readouterr().err
)
def test_main_defaults_to_repo_root(monkeypatch: pytest.MonkeyPatch) -> None:
seen: dict[str, object] = {}
def fake_sync(root: Path, *, write: bool) -> list[str]:
seen["root"] = root
seen["write"] = write
return []
monkeypatch.setattr(sync_mod, "sync", fake_sync)
assert sync_mod.main([]) == 0
assert seen == {"root": sync_mod.REPO_ROOT, "write": True}
def test_repository_is_in_sync() -> None:
"""The real checkout must match; a failure here means a rev has drifted.
Also proves every SYNC_TARGETS entry still resolves in the real files.
"""
assert sync_mod.sync(sync_mod.REPO_ROOT, write=False) == []
def test_cli_entry_point(root: Path) -> None:
"""Run the script the way the workflow does, as a subprocess."""
script = Path(sync_mod.__file__)
result = subprocess.run(
[sys.executable, str(script), "--check", "--root", str(root)],
capture_output=True,
text=True,
check=False,
)
assert result.returncode == 1
assert result.stdout.splitlines() == EXPECTED_DRIFT
@@ -0,0 +1,37 @@
"""Tests for the udp component configuration schema."""
from __future__ import annotations
import pytest
from esphome.components import udp
from esphome.components.packet_transport import (
CONF_BINARY_SENSORS,
CONF_ENCRYPTION,
CONF_PING_PONG_ENABLE,
CONF_PROVIDERS,
CONF_ROLLING_CODE_ENABLE,
CONF_SENSORS,
)
import esphome.config_validation as cv
@pytest.mark.parametrize(
"option",
[
CONF_PROVIDERS,
CONF_ENCRYPTION,
CONF_PING_PONG_ENABLE,
CONF_ROLLING_CODE_ENABLE,
CONF_SENSORS,
CONF_BINARY_SENSORS,
],
)
def test_relocated_option_rejected(option: str) -> None:
"""Options that moved to packet_transport raise a pointing error."""
with pytest.raises(cv.Invalid) as exc_info:
udp.CONFIG_SCHEMA({option: True})
assert (
f"The '{option}' option should now be configured in the 'packet_transport' component"
in str(exc_info.value)
)
@@ -1394,6 +1394,35 @@ def test_entity_metadata_visibility_hints() -> None:
assert web["web_server"].visibility is advanced
def test_with_visibility_remarks_keys() -> None:
"""``with_visibility`` re-marks the named keys, preserving each field's
default and validator, without touching the other keys or the input schema.
"""
base = cv.Schema(
{
cv.Optional("a", default=7): cv.int_,
cv.Optional("b", visibility=cv.Visibility.ADVANCED): cv.string,
}
)
promoted = cv.with_visibility(base, cv.Visibility.UI, "a")
pm = {str(k): k for k in promoted.schema}
assert pm["a"].visibility is cv.Visibility.UI # re-marked
assert pm["a"].default() == 7 # default preserved
assert pm["b"].visibility is cv.Visibility.ADVANCED # sibling untouched
assert promoted({}) == {"a": 7} # validator/default still applied
# The input schema is left untouched (no shared-marker mutation).
assert {str(k): k for k in base.schema}["a"].visibility is None
def test_with_visibility_unknown_key_raises() -> None:
"""A key not present in the schema is a typo — fail at build time."""
base = cv.Schema({cv.Optional("a"): cv.int_})
with pytest.raises(ValueError, match="not in schema"):
cv.with_visibility(base, cv.Visibility.UI, "nope")
def _wrap_str(value: str) -> ESPHomeDataBase:
"""Wrap a raw string as an ESPHomeDataBase, mimicking a YAML-loaded value."""
return make_data_base(value)
+3
View File
@@ -416,6 +416,9 @@ def test_perform_ota_no_auth(
"Update took 14.00 seconds (prepare 2.00, upload 5.00, commit 7.00)"
in caplog.text
)
# The data phase timeout must outlast the device's 105 s data timeout
mock_socket.settimeout.assert_any_call(espota2.DATA_PHASE_TIMEOUT)
assert espota2.DATA_PHASE_TIMEOUT > 105.0
@pytest.mark.usefixtures("mock_time")
@@ -7,6 +7,7 @@ exercised in their own test modules)."""
import json
import logging
from pathlib import Path
from unittest.mock import Mock
import pytest
@@ -228,6 +229,24 @@ def test_resolve_registry_version_raises_without_pkg_file(monkeypatch):
_resolve_registry_version("owner", "pkg", set())
def test_make_registry_client_skips_private_package_probe(monkeypatch):
"""Our client answers the probe locally without patching PlatformIO's class."""
from platformio.account.client import AccountClient
from platformio.registry.client import RegistryClient
pio_probe = RegistryClient.__dict__["allowed_private_packages"]
monkeypatch.setattr(
AccountClient,
"get_account_info",
Mock(side_effect=AssertionError("account probe must not run")),
)
client = lib._make_registry_client().get_registry_client_instance()
assert client.allowed_private_packages() is False
assert RegistryClient.__dict__["allowed_private_packages"] is pio_probe
def _patch_registry_resolve(monkeypatch: pytest.MonkeyPatch) -> None:
"""Stub the registry lookup so tests never touch the network."""
monkeypatch.setattr(
+16 -2
View File
@@ -1225,6 +1225,20 @@ def test_main_runs_prefetch(tmp_path: Path) -> None:
mock_prefetch.assert_called_once_with(tmp_path, "testenv")
def test_main_skips_private_package_probe_before_prefetch(tmp_path: Path) -> None:
"""The registry probe patch is applied before any package manager runs."""
order: list[str] = []
with (
patch.object(pf, "_prefetch", side_effect=lambda *_: order.append("prefetch")),
patch(
"esphome.platformio.runner.patch_registry_private_packages",
side_effect=lambda: order.append("patch"),
),
):
assert pf.main([str(tmp_path), "testenv"]) == 0
assert order == ["patch", "prefetch"]
def test_main_bad_argv_is_a_distinct_exit(
caplog: pytest.LogCaptureFixture,
) -> None:
@@ -1649,7 +1663,7 @@ def test_preinstall_runs_dependency_waves(tmp_path: Path) -> None:
{"name": "SPI"},
]
m.dependency_to_spec.side_effect = lambda dep: _FakeSpec(name=dep["name"])
pf._preinstall(m, [("noise-c@0.1.24", _FakeSpec(name="noise-c"))])
pf._preinstall(m, [("noise-c@0.1.26", _FakeSpec(name="noise-c"))])
assert installed == ["noise-c", "libsodium"] # dep deduped, SPI left out
# The dep wave carries its compatibility so _install searches qualified
dep_call = m._install.call_args_list[-1]
@@ -1669,7 +1683,7 @@ def test_preinstall_dependency_wave_skips_seen_names(tmp_path: Path) -> None:
m._install.side_effect = lambda spec, skip_dependencies, compatibility=None: (
installed.append(getattr(spec, "name", str(spec)))
)
pf._preinstall(m, [("noise-c@0.1.24", _FakeSpec(name="noise-c"))])
pf._preinstall(m, [("noise-c@0.1.26", _FakeSpec(name="noise-c"))])
assert installed == ["noise-c"]
@@ -6,7 +6,9 @@ from collections.abc import Callable
import io
import sys
from types import ModuleType
from unittest.mock import Mock
from platformio.registry.client import RegistryClient
import pytest
from esphome.platformio import runner
@@ -30,6 +32,7 @@ def _prepare_main(
monkeypatch.setattr(sys, "stderr", stream)
monkeypatch.setattr(runner, "patch_structhash", lambda: None)
monkeypatch.setattr(runner, "patch_file_downloader", lambda: None)
monkeypatch.setattr(runner, "patch_registry_private_packages", lambda: None)
platformio = ModuleType("platformio")
platformio_main = ModuleType("platformio.__main__")
@@ -91,3 +94,40 @@ def test_main_still_filters_a_drained_partial_line(
assert runner.main() == 0
assert buf.getvalue() == b""
def test_main_applies_registry_private_packages_patch(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""The probe is patched before PlatformIO runs."""
order: list[str] = []
_prepare_main(monkeypatch, lambda: order.append("pio") or 0)
monkeypatch.setattr(
runner, "patch_registry_private_packages", lambda: order.append("patch")
)
assert runner.main() == 0
assert order == ["patch", "pio"]
# Snapshot PlatformIO's own probe at import, before any test can patch it
_PIO_PROBE = RegistryClient.__dict__["allowed_private_packages"]
def test_patch_registry_private_packages_skips_account_probe(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Answers False without touching the account client."""
from platformio.account.client import AccountClient
monkeypatch.setattr(RegistryClient, "allowed_private_packages", _PIO_PROBE)
monkeypatch.setattr(
AccountClient,
"get_account_info",
Mock(side_effect=AssertionError("account probe must not run")),
)
runner.patch_registry_private_packages()
assert RegistryClient.allowed_private_packages() is False
assert RegistryClient().allowed_private_packages() is False