[packet_interface] Reject a layer type registered after the schemas are built

Such a type would be missing from the link schema, and selecting it would
fail with an unrelated error.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
clydebarrow
2026-10-19 22:20:49 +11:00
co-authored by Claude Opus 5.5
parent 554c2b5660
commit 60415975b4
2 changed files with 27 additions and 20 deletions
@@ -44,6 +44,8 @@ class LayerKind:
self.key = key
self.default_type = default_type
self.types: dict[str, LayerType] = {}
# Set once the schema is built, as types registered after that would not be in it
self.frozen = False
def register(
self, name: str, *, schema: Any
@@ -51,6 +53,11 @@ class LayerKind:
"""Register a type with its schema."""
def decorator(build: LayerBuilder) -> LayerBuilder:
if self.frozen:
raise ValueError(
f"{self.key} layer type '{name}' is registered after the schemas were built; register it "
"before the packet_interface component is imported"
)
if name in self.types:
raise ValueError(
f"{self.key} layer type '{name}' is already registered"
@@ -66,7 +73,8 @@ class LayerKind:
return {} if value is None else value
def schema(self) -> Callable[[Any], ConfigType]:
"""The schema for this kind, from the types registered so far."""
"""The schema for this kind; no type can be registered after this."""
self.frozen = True
return cv.All(
self._expand,
cv.typed_schema(
+18 -19
View File
@@ -11,6 +11,7 @@ from esphome.components.packet_interface.layers import (
CONF_RELIABILITY,
RELIABILITY,
STACK_ORDER,
LayerKind,
)
from esphome.components.packet_interface.layers.hdlc_reliability import (
CONF_MAX_QUEUE_LENGTH,
@@ -90,30 +91,28 @@ def test_layer_types_registered() -> None:
}
async def _build(config: dict, lower: Any) -> Any:
return lower
def test_registered_type_selectable(setup_core) -> None:
"""A newly registered layer type can be selected with type."""
async def build(config: dict, lower: Any) -> Any:
return lower
RELIABILITY.register("test", schema=cv.Schema({}))(build)
try:
config = RELIABILITY.schema()({"type": "test"})
assert config["type"] == "test"
assert RELIABILITY.get_type(config).build is build
finally:
del RELIABILITY.types["test"]
kind = LayerKind("test_kind", "test")
kind.register("test", schema=cv.Schema({}))(_build)
config = kind.schema()({"type": "test"})
assert config["type"] == "test"
assert kind.get_type(config).build is _build
def test_duplicate_layer_type_rejected(setup_core) -> None:
"""Registering a type under a name already in use is an error, not a silent replacement."""
kind = LayerKind("test_kind", "test")
kind.register("test", schema=cv.Schema({}))(_build)
with pytest.raises(ValueError, match="already registered"):
kind.register("test", schema=cv.Schema({}))(_build)
async def build(config: dict, lower: Any) -> Any:
return lower
RELIABILITY.register("test", schema=cv.Schema({}))(build)
try:
with pytest.raises(ValueError, match="already registered"):
RELIABILITY.register("test", schema=cv.Schema({}))(build)
finally:
del RELIABILITY.types["test"]
def test_late_registration_rejected(setup_core) -> None:
"""A type registered after the schemas were built would be missing from them, so it is an error."""
with pytest.raises(ValueError, match="after the schemas were built"):
RELIABILITY.register("late", schema=cv.Schema({}))(_build)