Consolidate zigbee schema factory and shared integration setup

esphome/components/binary_sensor/__init__.py

@@ -3,7 +3,6 @@ from logging import getLogger
 from esphome import automation, core
 from esphome.automation import Condition, maybe_simple_id
 import esphome.codegen as cg
-from esphome.components import mqtt, web_server, zigbee
 from esphome.components.const import CONF_ON_STATE_CHANGE
 import esphome.config_validation as cv
 from esphome.const import (
@@ -28,7 +27,6 @@ from esphome.const import (
     CONF_STATE,
     CONF_TIMING,
     CONF_TRIGGER_ID,
-    CONF_WEB_SERVER,
     DEVICE_CLASS_BATTERY,
     DEVICE_CLASS_BATTERY_CHARGING,
     DEVICE_CLASS_CARBON_MONOXIDE,
@@ -59,9 +57,11 @@ from esphome.const import (
     DEVICE_CLASS_VIBRATION,
     DEVICE_CLASS_WINDOW,
 )
-from esphome.core import CORE, CoroPriority, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority, entity_helpers
 from esphome.core.entity_helpers import (
     entity_duplicate_validator,
+    lazy_load_validator,
+    mqtt_component_class,
     queue_entity_register,
     setup_device_class,
     setup_entity,
@@ -433,14 +433,14 @@ def validate_publish_initial_state(value):
 
 
 _BINARY_SENSOR_SCHEMA = (
-    cv.ENTITY_BASE_SCHEMA.extend(web_server.WEBSERVER_SORTING_SCHEMA)
+    cv.ENTITY_BASE_SCHEMA.extend(entity_helpers.WEBSERVER_SORTING_SCHEMA)
     .extend(cv.MQTT_COMPONENT_SCHEMA)
-    .extend(zigbee.BINARY_SENSOR_SCHEMA)
+    .extend(entity_helpers.ZIGBEE_BINARY_SENSOR_SCHEMA)
     .extend(
         {
             cv.GenerateID(): cv.declare_id(BinarySensor),
             cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(
-                mqtt.MQTTBinarySensorComponent
+                mqtt_component_class("MQTTBinarySensorComponent")
             ),
             cv.Exclusive(
                 CONF_PUBLISH_INITIAL_STATE, CONF_TRIGGER_ON_INITIAL_STATE
@@ -505,7 +505,7 @@ _BINARY_SENSOR_SCHEMA = (
 
 
 _BINARY_SENSOR_SCHEMA.add_extra(entity_duplicate_validator("binary_sensor"))
-_BINARY_SENSOR_SCHEMA.add_extra(zigbee.validate_binary_sensor)
+_BINARY_SENSOR_SCHEMA.add_extra(lazy_load_validator("zigbee", "validate_binary_sensor"))
 
 
 def binary_sensor_schema(
@@ -607,14 +607,12 @@ async def setup_binary_sensor_core_(var, config):
 
     CORE.add_job(_build_binary_sensor_automations, var, config)
 
-    if mqtt_id := config.get(CONF_MQTT_ID):
+    await entity_helpers.setup_entity_integrations(var, config)
-        mqtt_ = cg.new_Pvariable(mqtt_id, var)
-        await mqtt.register_mqtt_component(mqtt_, config)
 
-    if web_server_config := config.get(CONF_WEB_SERVER):
+    if "zigbee" in CORE.loaded_integrations:
-        await web_server.add_entity_config(var, web_server_config)
+        from esphome.components import zigbee
 
-    await zigbee.setup_binary_sensor(var, config)
+        await zigbee.setup_binary_sensor(var, config)
 
 
 async def register_binary_sensor(var, config):

esphome/components/web_server/__init__.py

@@ -10,7 +10,11 @@ from esphome.components import web_server_base
 from esphome.components.logger import request_log_listener
 from esphome.components.web_server_base import CONF_WEB_SERVER_BASE_ID
 import esphome.config_validation as cv
-from esphome.const import (
+
+# Re-exported so entity components and external components keep importing
+# these from web_server; defined outside this package so entity base
+# schemas do not need to import it.
+from esphome.const import (  # noqa: F401
     CONF_AUTH,
     CONF_COMPRESSION,
     CONF_CSS_INCLUDE,
@@ -26,6 +30,8 @@ from esphome.const import (
     CONF_OTA,
     CONF_PASSWORD,
     CONF_PORT,
+    CONF_SORTING_GROUP_ID,
+    CONF_SORTING_WEIGHT,
     CONF_TYPE,
     CONF_USERNAME,
     CONF_VERSION,
@@ -39,6 +45,7 @@ from esphome.const import (
     PLATFORM_RTL87XX,
 )
 from esphome.core import CORE, CoroPriority, coroutine_with_priority
+from esphome.core.entity_helpers import WEBSERVER_SORTING_SCHEMA  # noqa: F401
 import esphome.final_validate as fv
 from esphome.types import ConfigType
 
@@ -49,9 +56,7 @@ AUTO_LOAD = ["json", "web_server_base"]
 AUTH_TYPE_BASIC = "basic"
 AUTH_TYPE_DIGEST = "digest"
 
-CONF_SORTING_GROUP_ID = "sorting_group_id"
 CONF_SORTING_GROUPS = "sorting_groups"
-CONF_SORTING_WEIGHT = "sorting_weight"
 CONF_ALLOWED_ORIGINS = "allowed_origins"
 
 
@@ -225,27 +230,6 @@ sorting_group = {
     cv.Optional(CONF_SORTING_WEIGHT): cv.float_,
 }
 
-WEBSERVER_SORTING_SCHEMA = cv.Schema(
-    {
-        # The per-entity web_server block is cosmetic dashboard ordering —
-        # mark the whole block advanced; the children inherit via the cascade.
-        cv.Optional(CONF_WEB_SERVER, visibility=cv.Visibility.ADVANCED): cv.Schema(
-            {
-                cv.OnlyWith(CONF_WEB_SERVER_ID, "web_server"): cv.use_id(WebServer),
-                cv.Optional(CONF_SORTING_WEIGHT): cv.All(
-                    cv.requires_component("web_server"),
-                    cv.float_,
-                ),
-                cv.Optional(CONF_SORTING_GROUP_ID): cv.All(
-                    cv.requires_component("web_server"),
-                    cv.use_id(cg.int_),
-                ),
-            }
-        )
-    }
-)
-
-
 CONFIG_SCHEMA = cv.All(
     cv.Schema(
         {

esphome/components/zigbee/__init__.py

@@ -14,10 +14,12 @@ from esphome.components.esp32.const import (
 )
 import esphome.config_validation as cv
 from esphome.const import CONF_ID, CONF_INTERNAL, CONF_MODEL, CONF_NAME, CONF_ON_START
-from esphome.core import CORE, CoroPriority, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority, entity_helpers
 from esphome.types import ConfigType
 
-from .const import (
+# CONF_ENDPOINT, CONF_MAX_EP_NUMBER, CONF_REPORT, CONF_USE_DEVICE_TYPE and
+# REPORT are re-exported for existing consumers of this package's namespace.
+from .const import (  # noqa: F401
     CONF_ENDPOINT,
     CONF_MAX_EP_NUMBER,
     CONF_ON_JOIN,
@@ -45,12 +47,7 @@ from .zigbee_esp32 import (
     validate_sensor_esp32,
     zigbee_require_vfs_select,
 )
-from .zigbee_zephyr import (
+from .zigbee_zephyr import zephyr_number, zephyr_sensor, zephyr_switch
-    zephyr_binary_sensor,
-    zephyr_number,
-    zephyr_sensor,
-    zephyr_switch,
-)
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -59,35 +56,10 @@ CODEOWNERS = ["@luar123", "@tomaszduda23"]
 CONFLICTS_WITH = ["openthread"]
 
 
-def _check_report_deprecation(value: str) -> str:
+# Defined in esphome.core.entity_helpers so entity base schemas can reference
-    if str(value).lower() in ("coordinator", "enable"):
+# them without importing this package; re-exported here for existing consumers.
-        _LOGGER.warning(
+BASE_SCHEMA = entity_helpers.ZIGBEE_BASE_ENTITY_SCHEMA
-            "Report options 'coordinator' and 'enable' are deprecated and will be removed in a future release. Use 'default' instead."
+BINARY_SENSOR_SCHEMA = entity_helpers.ZIGBEE_BINARY_SENSOR_SCHEMA
-        )
-    return value
-
-
-BASE_SCHEMA = cv.Schema(
-    {
-        cv.Optional(CONF_REPORT): cv.All(
-            cv.requires_component("zigbee"),
-            cv.requires_component("esp32"),
-            _check_report_deprecation,
-            cv.enum(REPORT, lower=True),
-        ),
-        cv.Optional(CONF_ENDPOINT): cv.All(
-            cv.requires_component("zigbee"),
-            cv.requires_component("esp32"),
-            cv.int_range(1, CONF_MAX_EP_NUMBER),
-        ),
-        cv.Optional(CONF_USE_DEVICE_TYPE): cv.All(
-            cv.requires_component("zigbee"),
-            cv.requires_component("esp32"),
-            cv.boolean,
-        ),
-    }
-)
-BINARY_SENSOR_SCHEMA = cv.Schema({}).extend(BASE_SCHEMA).extend(zephyr_binary_sensor)
 SENSOR_SCHEMA = cv.Schema({}).extend(BASE_SCHEMA).extend(zephyr_sensor)
 SWITCH_SCHEMA = cv.Schema({}).extend(zephyr_switch)
 NUMBER_SCHEMA = cv.Schema({}).extend(zephyr_number)

esphome/components/zigbee/const.py

@@ -1,7 +1,14 @@
 from enum import IntEnum
 
 import esphome.codegen as cg
-from esphome.const import (
+
+# The entity schema keys and report enum live outside this package so
+# entity base schemas can use them without importing it; re-imported here
+# so zigbee code keeps its existing import paths.
+from esphome.const import (  # noqa: F401  # pylint: disable=unused-import
+    CONF_ENDPOINT,
+    CONF_REPORT,
+    CONF_USE_DEVICE_TYPE,
     DEVICE_CLASS_CURRENT,
     DEVICE_CLASS_DURATION,
     DEVICE_CLASS_ENERGY,
@@ -48,6 +55,10 @@ from esphome.const import (
     UNIT_WATT,
     UNIT_WATT_HOURS,
 )
+from esphome.core.entity_helpers import (  # noqa: F401  # pylint: disable=unused-import
+    ZIGBEE_MAX_EP_NUMBER as CONF_MAX_EP_NUMBER,
+    ZIGBEE_REPORT as REPORT,
+)
 
 zigbee_ns = cg.esphome_ns.namespace("zigbee")
 ZigbeeComponent = zigbee_ns.class_("ZigbeeComponent", cg.Component)
@@ -56,22 +67,10 @@ BinaryAttrs = zigbee_ns.struct("BinaryAttrs")
 AnalogAttrs = zigbee_ns.struct("AnalogAttrs")
 AnalogAttrsOutput = zigbee_ns.struct("AnalogAttrsOutput")
 
-report = zigbee_ns.enum("ZigbeeReportT")
-REPORT = {
-    "coordinator": report.ZIGBEE_REPORT_COORDINATOR,
-    "enable": report.ZIGBEE_REPORT_ENABLE,
-    "force": report.ZIGBEE_REPORT_FORCE,
-    "default": report.ZIGBEE_REPORT_DEFAULT,
-}
-
-CONF_ENDPOINT = "endpoint"
-CONF_MAX_EP_NUMBER = 239
 CONF_ON_JOIN = "on_join"
 CONF_WIPE_ON_BOOT = "wipe_on_boot"
-CONF_REPORT = "report"
 CONF_ROUTER = "router"
 CONF_POWER_SOURCE = "power_source"
-CONF_USE_DEVICE_TYPE = "use_device_type"
 POWER_SOURCE = {
     "UNKNOWN": 0x00,  # ZB_ZCL_BASIC_POWER_SOURCE_UNKNOWN
     "MAINS_SINGLE_PHASE": 0x01,  # ZB_ZCL_BASIC_POWER_SOURCE_MAINS_SINGLE_PHASE

esphome/components/zigbee/const_zephyr.py

@@ -1,6 +1,11 @@
+# These two schema keys live in esphome.components.const so entity base
+# schemas can use them without importing this package.
+from esphome.const import (  # noqa: F401  # pylint: disable=unused-import
+    CONF_ZIGBEE_BINARY_SENSOR,
+    CONF_ZIGBEE_ID,
+)
+
 CONF_MAX_EP_NUMBER_ZEPHYR = 8
-CONF_ZIGBEE_ID = "zigbee_id"
-CONF_ZIGBEE_BINARY_SENSOR = "zigbee_binary_sensor"
 CONF_ZIGBEE_SENSOR = "zigbee_sensor"
 CONF_ZIGBEE_SWITCH = "zigbee_switch"
 CONF_ZIGBEE_NUMBER = "zigbee_number"

esphome/components/zigbee/zigbee_zephyr.py

@@ -57,15 +57,6 @@ ZigbeeSensor = zigbee_ns.class_("ZigbeeSensor", cg.Component)
 ZigbeeSwitch = zigbee_ns.class_("ZigbeeSwitch", cg.Component)
 ZigbeeNumber = zigbee_ns.class_("ZigbeeNumber", cg.Component)
 
-zephyr_binary_sensor = cv.Schema(
-    {
-        cv.OnlyWith(CONF_ZIGBEE_ID, ["nrf52", "zigbee"]): cv.use_id(ZigbeeComponent),
-        cv.OnlyWith(CONF_ZIGBEE_BINARY_SENSOR, ["nrf52", "zigbee"]): cv.declare_id(
-            ZigbeeBinarySensor
-        ),
-    }
-)
-
 zephyr_sensor = cv.Schema(
     {
         cv.OnlyWith(CONF_ZIGBEE_ID, ["nrf52", "zigbee"]): cv.use_id(ZigbeeComponent),

esphome/const.py

@@ -387,6 +387,7 @@ CONF_ENABLE_PRIVATE_NETWORK_ACCESS = "enable_private_network_access"
 CONF_ENABLE_RRM = "enable_rrm"
 CONF_ENABLE_TIME = "enable_time"
 CONF_ENCRYPTION = "encryption"
+CONF_ENDPOINT = "endpoint"
 CONF_ENERGY = "energy"
 CONF_ENTITY_CATEGORY = "entity_category"
 CONF_ENTITY_ID = "entity_id"
@@ -885,6 +886,7 @@ CONF_REFERENCE_VOLTAGE = "reference_voltage"
 CONF_REFRESH = "refresh"
 CONF_RELABEL = "relabel"
 CONF_REPEAT = "repeat"
+CONF_REPORT = "report"
 CONF_REPOSITORY = "repository"
 CONF_RESET = "reset"
 CONF_RESET_DURATION = "reset_duration"
@@ -962,6 +964,8 @@ CONF_SLEEP_DURATION = "sleep_duration"
 CONF_SLEEP_PIN = "sleep_pin"
 CONF_SLEEP_WHEN_DONE = "sleep_when_done"
 CONF_SONY = "sony"
+CONF_SORTING_GROUP_ID = "sorting_group_id"
+CONF_SORTING_WEIGHT = "sorting_weight"
 CONF_SOURCE = "source"
 CONF_SOURCE_ID = "source_id"
 CONF_SPEAKER = "speaker"
@@ -1093,6 +1097,7 @@ CONF_UPDATE_ON_BOOT = "update_on_boot"
 CONF_URL = "url"
 CONF_USE_ABBREVIATIONS = "use_abbreviations"
 CONF_USE_ADDRESS = "use_address"
+CONF_USE_DEVICE_TYPE = "use_device_type"
 CONF_USE_DMA = "use_dma"
 CONF_USE_FAHRENHEIT = "use_fahrenheit"
 CONF_USERNAME = "username"
@@ -1144,6 +1149,8 @@ CONF_Y = "y"
 CONF_Y_GRID = "y_grid"
 CONF_YEAR = "year"
 CONF_ZERO = "zero"
+CONF_ZIGBEE_BINARY_SENSOR = "zigbee_binary_sensor"
+CONF_ZIGBEE_ID = "zigbee_id"
 
 TYPE_GIT = "git"
 TYPE_LOCAL = "local"

esphome/core/entity_helpers.py

@@ -1,6 +1,8 @@
 from collections.abc import Callable
 from dataclasses import dataclass, field
 import functools
+from importlib import import_module
+from importlib.util import find_spec
 import logging
 
 import esphome.codegen as cg
@@ -9,12 +11,22 @@ from esphome.const import (
     CONF_DEVICE_CLASS,
     CONF_DEVICE_ID,
     CONF_DISABLED_BY_DEFAULT,
+    CONF_ENDPOINT,
     CONF_ENTITY_CATEGORY,
     CONF_ICON,
     CONF_ID,
     CONF_INTERNAL,
+    CONF_MQTT_ID,
     CONF_NAME,
+    CONF_REPORT,
+    CONF_SORTING_GROUP_ID,
+    CONF_SORTING_WEIGHT,
     CONF_UNIT_OF_MEASUREMENT,
+    CONF_USE_DEVICE_TYPE,
+    CONF_WEB_SERVER,
+    CONF_WEB_SERVER_ID,
+    CONF_ZIGBEE_BINARY_SENSOR,
+    CONF_ZIGBEE_ID,
 )
 from esphome.core import CORE, ID, CoroPriority, coroutine_with_priority
 from esphome.core.config import (
@@ -22,7 +34,7 @@ from esphome.core.config import (
     ICON_MAX_LENGTH,
     UNIT_OF_MEASUREMENT_MAX_LENGTH,
 )
-from esphome.cpp_generator import MockObj, RawStatement, add, get_variable
+from esphome.cpp_generator import MockObj, MockObjClass, RawStatement, add, get_variable
 from esphome.cpp_types import App
 import esphome.final_validate as fv
 from esphome.helpers import (
@@ -635,3 +647,156 @@ def entity_duplicate_validator(platform: str) -> Callable[[ConfigType], ConfigTy
         return config
 
     return validator
+
+
+# ---------------------------------------------------------------------------
+# Cross-integration entity schema fragments
+# ---------------------------------------------------------------------------
+#
+# Entity base schemas offer mqtt/web_server/zigbee options, but importing
+# those packages pulls their full dependency chains (mqtt and zigbee both
+# import the esp32 package) into every entity component import. The
+# fragments below are built from cheap primitives instead: MockObjClass
+# identity is string-based, so the class handles here are interchangeable
+# with the ones the integrations declare, and every key is guarded by
+# ``cv.requires_component``/``cv.OnlyWith``, which consult
+# ``CORE.loaded_integrations`` at validation time without importing. The
+# owning integrations re-export the shared schema names and key strings so
+# each stays defined once.
+
+_WebServer = cg.esphome_ns.namespace("web_server").class_(
+    "WebServer", cg.Component, cg.Controller
+)
+
+WEBSERVER_SORTING_SCHEMA = cv.Schema(
+    {
+        # The per-entity web_server block is cosmetic dashboard ordering —
+        # mark the whole block advanced; the children inherit via the cascade.
+        cv.Optional(CONF_WEB_SERVER, visibility=cv.Visibility.ADVANCED): cv.Schema(
+            {
+                cv.OnlyWith(CONF_WEB_SERVER_ID, "web_server"): cv.use_id(_WebServer),
+                cv.Optional(CONF_SORTING_WEIGHT): cv.All(
+                    cv.requires_component("web_server"),
+                    cv.float_,
+                ),
+                cv.Optional(CONF_SORTING_GROUP_ID): cv.All(
+                    cv.requires_component("web_server"),
+                    cv.use_id(cg.int_),
+                ),
+            }
+        )
+    }
+)
+
+_mqtt_ns = cg.esphome_ns.namespace("mqtt")
+_MQTTComponent = _mqtt_ns.class_("MQTTComponent", cg.Component)
+
+
+def mqtt_component_class(name: str) -> MockObjClass:
+    """Handle for a per-entity mqtt::<name> companion class."""
+    return _mqtt_ns.class_(name, _MQTTComponent)
+
+
+ZIGBEE_MAX_EP_NUMBER = 239
+
+_zigbee_ns = cg.esphome_ns.namespace("zigbee")
+_ZigbeeComponent = _zigbee_ns.class_("ZigbeeComponent", cg.Component)
+_zigbee_report = _zigbee_ns.enum("ZigbeeReportT")
+ZIGBEE_REPORT = {
+    "coordinator": _zigbee_report.ZIGBEE_REPORT_COORDINATOR,
+    "enable": _zigbee_report.ZIGBEE_REPORT_ENABLE,
+    "force": _zigbee_report.ZIGBEE_REPORT_FORCE,
+    "default": _zigbee_report.ZIGBEE_REPORT_DEFAULT,
+}
+
+
+def _check_report_deprecation(value: str) -> str:
+    if str(value).lower() in ("coordinator", "enable"):
+        _LOGGER.warning(
+            "Report options 'coordinator' and 'enable' are deprecated and will be removed in a future release. Use 'default' instead."
+        )
+    return value
+
+
+ZIGBEE_BASE_ENTITY_SCHEMA = cv.Schema(
+    {
+        cv.Optional(CONF_REPORT): cv.All(
+            cv.requires_component("zigbee"),
+            cv.requires_component("esp32"),
+            _check_report_deprecation,
+            cv.enum(ZIGBEE_REPORT, lower=True),
+        ),
+        cv.Optional(CONF_ENDPOINT): cv.All(
+            cv.requires_component("zigbee"),
+            cv.requires_component("esp32"),
+            cv.int_range(1, ZIGBEE_MAX_EP_NUMBER),
+        ),
+        cv.Optional(CONF_USE_DEVICE_TYPE): cv.All(
+            cv.requires_component("zigbee"),
+            cv.requires_component("esp32"),
+            cv.boolean,
+        ),
+    }
+)
+
+# Entity platform -> (config key, zigbee C++ class). A unit test checks each
+# class against the owning declaration in zigbee_zephyr.
+_ZIGBEE_ENTITY_CLASSES = {
+    "binary_sensor": (CONF_ZIGBEE_BINARY_SENSOR, "ZigbeeBinarySensor"),
+}
+
+
+def _zigbee_entity_schema(platform: str) -> cv.Schema:
+    conf_key, class_name = _ZIGBEE_ENTITY_CLASSES[platform]
+    return ZIGBEE_BASE_ENTITY_SCHEMA.extend(
+        {
+            cv.OnlyWith(CONF_ZIGBEE_ID, ["nrf52", "zigbee"]): cv.use_id(
+                _ZigbeeComponent
+            ),
+            cv.OnlyWith(conf_key, ["nrf52", "zigbee"]): cv.declare_id(
+                _zigbee_ns.class_(class_name, cg.Component)
+            ),
+        }
+    )
+
+
+ZIGBEE_BINARY_SENSOR_SCHEMA = _zigbee_entity_schema("binary_sensor")
+
+
+def lazy_load_validator(
+    component: str, name: str
+) -> Callable[[ConfigType], ConfigType]:
+    """Schema extra delegating to ``components.<component>.<name>`` when loaded."""
+    if find_spec(f"esphome.components.{component}") is None:
+        raise ValueError(f"No such component {component!r}")
+
+    def validator(config: ConfigType) -> ConfigType:
+        if component not in CORE.loaded_integrations:
+            return config
+        module = import_module(f"esphome.components.{component}")
+        if (delegate := getattr(module, name, None)) is None:
+            raise ValueError(f"{component} has no validator {name!r}")
+        return delegate(config)
+
+    return validator
+
+
+async def setup_entity_integrations(var: MockObj, config: ConfigType) -> MockObj | None:
+    """Register the mqtt companion and web_server entry for an entity.
+
+    Imports the integrations lazily; returns the mqtt companion (or None)
+    so callers can apply integration specific options to it.
+    """
+    mqtt_ = None
+    if (mqtt_id := config.get(CONF_MQTT_ID)) is not None:
+        from esphome.components import mqtt
+
+        mqtt_ = cg.new_Pvariable(mqtt_id, var)
+        await mqtt.register_mqtt_component(mqtt_, config)
+
+    if web_server_config := config.get(CONF_WEB_SERVER):
+        from esphome.components import web_server
+
+        await web_server.add_entity_config(var, web_server_config)
+
+    return mqtt_

script/ci-custom.py

@@ -557,7 +557,7 @@ def lint_constants_usage():
 # Maximum allowed CONF_ constants in esphome/const.py.
 # This file is frozen — new constants go in esphome/components/const/__init__.py.
 # Decrease this number when constants are moved out of const.py.
-CONST_PY_MAX_CONF = 1017
+CONST_PY_MAX_CONF = 1024
 
 
 @lint_content_check(include=["esphome/const.py"])

tests/unit_tests/core/test_entity_helpers.py

@@ -1,9 +1,11 @@
 """Test get_base_entity_object_id function matches C++ behavior."""
 
 from collections.abc import Callable, Generator
+import logging
 from pathlib import Path
 import re
 from typing import Any
+from unittest.mock import patch
 
 import pytest
 
@@ -21,11 +23,14 @@ from esphome.const import (
 )
 from esphome.core import CORE, ID, entity_helpers
 from esphome.core.entity_helpers import (
+    _check_report_deprecation,
     _register_string,
     _setup_entity_impl,
     entity_duplicate_validator,
     finalize_entity_strings,
     get_base_entity_object_id,
+    lazy_load_validator,
+    mqtt_component_class,
     register_device_class,
     register_icon,
     register_unit_of_measurement,
@@ -1236,3 +1241,82 @@ async def test_finalize_comment_sanitization(
     # Newline must be replaced to prevent breaking out of comment
     assert "\n" not in comment_line
     assert "INJECTED_CODE" in comment_line  # still visible but safe in comment
+
+
+@pytest.mark.parametrize(
+    ("value", "warns"),
+    [
+        ("coordinator", True),
+        ("enable", True),
+        ("force", False),
+        ("default", False),
+    ],
+)
+def test_check_report_deprecation(
+    value: str, warns: bool, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Deprecated zigbee report options warn; the value always passes through."""
+    with caplog.at_level(logging.WARNING):
+        assert _check_report_deprecation(value) == value
+    assert ("deprecated" in caplog.text) is warns
+
+
+def test_lazy_load_validator_defers_import() -> None:
+    """The validator no-ops without importing unless the component is loaded."""
+    validator = lazy_load_validator("zigbee", "validate_binary_sensor")
+    config = {CONF_NAME: "test"}
+
+    with (
+        patch.object(CORE, "loaded_integrations", set()),
+        patch("esphome.core.entity_helpers.import_module") as import_mock,
+    ):
+        assert validator(config) is config
+        import_mock.assert_not_called()
+
+        CORE.loaded_integrations.add("zigbee")
+        delegate = import_mock.return_value.validate_binary_sensor
+        delegate.return_value = {CONF_NAME: "validated"}
+        assert validator(config) == {CONF_NAME: "validated"}
+        import_mock.assert_called_once_with("esphome.components.zigbee")
+        delegate.assert_called_once_with(config)
+
+
+def test_lazy_load_validator_rejects_unknown_component() -> None:
+    """A typo in the component name fails at schema construction."""
+    with pytest.raises(ValueError, match="no_such_component"):
+        lazy_load_validator("no_such_component", "validate_binary_sensor")
+
+
+def test_lazy_load_validator_names_missing_hook() -> None:
+    """A missing hook raises a clear error naming the component and hook."""
+    validator = lazy_load_validator("zigbee", "no_such_hook")
+
+    with (
+        patch.object(CORE, "loaded_integrations", {"zigbee"}),
+        patch("esphome.core.entity_helpers.import_module") as import_mock,
+        pytest.raises(ValueError, match="no_such_hook"),
+    ):
+        del import_mock.return_value.no_such_hook
+        validator({})
+
+
+def test_integration_class_handles_match_owning_definitions() -> None:
+    """The cheap class handles must stay string-equal to the integrations'
+    own declarations, or use_id/declare_id resolution silently drifts."""
+    from esphome.components import mqtt, web_server
+    from esphome.components.zigbee import zigbee_zephyr
+    from esphome.components.zigbee.const import ZigbeeComponent
+
+    mqtt_handle = mqtt_component_class("MQTTBinarySensorComponent")
+    assert str(mqtt_handle) == str(mqtt.MQTTBinarySensorComponent)
+    assert mqtt_handle.inherits_from(mqtt.MQTTComponent)
+    assert mqtt.MQTTBinarySensorComponent.inherits_from(entity_helpers._MQTTComponent)
+
+    assert str(entity_helpers._WebServer) == str(web_server.WebServer)
+    assert entity_helpers._WebServer.inherits_from(web_server.WebServer)
+    assert web_server.WebServer.inherits_from(entity_helpers._WebServer)
+
+    assert str(entity_helpers._ZigbeeComponent) == str(ZigbeeComponent)
+    for _conf_key, class_name in entity_helpers._ZIGBEE_ENTITY_CLASSES.values():
+        owning = getattr(zigbee_zephyr, class_name)
+        assert str(owning) == f"zigbee::{class_name}"

tests/unit_tests/test_lazy_imports.py

@@ -190,6 +190,40 @@ def test_api_client_does_not_import_heavy_modules() -> None:
     )
 
 
+def test_binary_sensor_does_not_import_integrations() -> None:
+    """An entity component must not drag in its optional integrations.
+
+    binary_sensor's mqtt/web_server/zigbee schema fragments live in
+    ``esphome.core.entity_helpers``; the integration packages (and the
+    esp32/logger chains mqtt and zigbee pull in) must only load when the
+    user's config actually uses them.
+    """
+    allowed = (
+        "esphome.components",
+        "esphome.components.binary_sensor",
+        "esphome.components.const",
+    )
+    check = (
+        "import sys; import esphome.components.binary_sensor; "
+        f"leaked = [m for m in sys.modules "
+        f"if m.startswith('esphome.components') and m not in {allowed!r}]; "
+        "print(','.join(leaked))"
+    )
+    result = subprocess.run(
+        [sys.executable, "-c", check],
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+    leaked = result.stdout.strip()
+    assert not leaked, (
+        f"esphome.components.binary_sensor imports integration packages at "
+        f"top level: {leaked}. Keep the shared schema fragments in "
+        "esphome.core.entity_helpers and import the integration inside "
+        "to_code instead."
+    )
+
+
 def test_stacktrace_does_not_import_heavy_modules() -> None:
     """``esphome.stacktrace`` guards its own docstring's contract.