[core] Drop unused return values from FINAL_VALIDATE_SCHEMA validators

loader.py declares final_validate_schema as Callable[[ConfigType], None]
and config.py discards whatever it returns, so a validator wired straight
to FINAL_VALIDATE_SCHEMA that ends in "return config" implies the value
matters when it does not. Drop those returns and annotate the functions
-> None. Validators composed into a cv.All still return their config,
since each one feeds the next. Returned validation calls are kept as
plain statements so no validation step is lost.
This commit is contained in:
Jesse Hills
2026-10-15 00:15:51 +13:00
parent 56cc03c4d4
commit 78a0ffb812
29 changed files with 59 additions and 74 deletions
@@ -29,7 +29,7 @@ def _request_micro_decoder(config: ConfigType) -> ConfigType:
return config
def _inherit_ca_certificate_path(config: ConfigType) -> ConfigType:
def _inherit_ca_certificate_path(config: ConfigType) -> None:
# Default to the CA certificate configured on the http_request component so
# HTTPS playback verifies against the same trust anchor without repeating
# the option on every source. Needed because audio_http sources are often
@@ -37,15 +37,15 @@ def _inherit_ca_certificate_path(config: ConfigType) -> ConfigType:
# The PEM replaces the built-in certificate bundle as the sole trust anchor,
# so make the inheritance visible in the log.
if CONF_CA_CERTIFICATE_PATH in config:
return config
return
fconf = fv.full_config.get()
if CONF_HTTP_REQUEST not in fconf:
return config
return
http_request_config = fconf[CONF_HTTP_REQUEST]
# Mirror http_request's own semantics: it only applies its CA when SSL
# verification is enabled, so neither does the inheritance.
if not http_request_config.get(CONF_VERIFY_SSL, True):
return config
return
if ca_cert_path := http_request_config.get(CONF_CA_CERTIFICATE_PATH):
_LOGGER.info(
"audio_http source '%s' is inheriting ca_certificate_path from the "
@@ -55,7 +55,6 @@ def _inherit_ca_certificate_path(config: ConfigType) -> ConfigType:
)
# Already validated by http_request's schema; the value is a resolved Path.
config[CONF_CA_CERTIFICATE_PATH] = ca_cert_path
return config
FINAL_VALIDATE_SCHEMA = _inherit_ca_certificate_path
@@ -48,7 +48,7 @@ def _reject_debug(uart_conf: ConfigType) -> ConfigType:
return uart_conf
def _final_validate(config: ConfigType) -> ConfigType:
def _final_validate(config: ConfigType) -> None:
full_config = fv.full_config.get()
# Bridges of any platform must own their interfaces exclusively; shared ring
# buffers and overwritten callbacks would corrupt both streams silently. The
@@ -81,7 +81,6 @@ def _final_validate(config: ConfigType) -> ConfigType:
)
fv.id_declaration_match_schema(_reject_debug)(config[CONF_UART_ID])
return config
FINAL_VALIDATE_SCHEMA = _final_validate
+1 -2
View File
@@ -57,7 +57,7 @@ async def to_code(config: ConfigType) -> None:
cg.add(var.set_time(await cg.get_variable(time_id)))
def _final_validate(config: ConfigType) -> ConfigType:
def _final_validate(config: ConfigType) -> None:
# The unit only talks over an encrypted, bonded link
ble_conf = fv.full_config.get()[esp32_ble.DOMAIN]
if (
@@ -69,7 +69,6 @@ def _final_validate(config: ConfigType) -> ConfigType:
"ecocomfort2 needs esp32_ble configured with io_capability: none, "
"auth_req_mode: sc_bond and max_key_size: 16"
)
return config
FINAL_VALIDATE_SCHEMA = _final_validate
@@ -206,7 +206,7 @@ async def to_code(config: ConfigType) -> None:
add_idf_sdkconfig_option("CONFIG_IEEE802154_SLEEP_ENABLE", True)
def _final_validate(config: ConfigType) -> ConfigType:
def _final_validate(config: ConfigType) -> None:
full_config = fv.full_config.get()
entries = full_config.get(power_management.DOMAIN, [])
if sum(entry.get(CONF_PLATFORM) == DOMAIN for entry in entries) > 1:
@@ -248,7 +248,6 @@ def _final_validate(config: ConfigType) -> ConfigType:
f"{CONF_POWER_DOWN_FLASH} is not allowed when the device has PSRAM",
path=[CONF_POWER_DOWN_FLASH],
)
return config
FINAL_VALIDATE_SCHEMA = _final_validate
+3 -4
View File
@@ -148,21 +148,20 @@ CONFIG_SCHEMA = cv.All(
)
def _validate_variant(config: ConfigType) -> ConfigType:
def _validate_variant(config: ConfigType) -> None:
# ESP-NOW rides the Wi-Fi PHY. Radio-less esp32 variants have no native
# ESP-NOW; only the ESP32-P4 has a path, via the esp32_hosted shim that
# supplies the esp_now_* symbols. Fail here with a clear message instead of
# letting the build reach an "undefined reference to esp_now_*" link error.
if not CORE.is_esp32:
return config
return
variant = get_esp32_variant()
if wifi.variant_has_wifi(variant):
return config
return
if variant != VARIANT_ESP32P4:
raise cv.Invalid(f"ESP-NOW is not supported on {variant} (no Wi-Fi radio)")
if ESP32_HOSTED_DOMAIN not in fv.full_config.get():
raise cv.Invalid(f"ESP-NOW on {variant} requires the esp32_hosted component")
return config
FINAL_VALIDATE_SCHEMA = _validate_variant
+2 -3
View File
@@ -44,10 +44,10 @@ HOME_ASSISTANT_IMPORT_CONTROL_SCHEMA = cv.Schema(
)
def final_validate_entity_id(config: ConfigType) -> ConfigType:
def final_validate_entity_id(config: ConfigType) -> None:
"""Without an entity_id, the entity must be a wizard input, as Home Assistant then supplies it."""
if CONF_ENTITY_ID in config:
return config
return
from esphome.components.api import wizard
if config[CONF_ID].id not in wizard.wizard_input_ids(
@@ -56,7 +56,6 @@ def final_validate_entity_id(config: ConfigType) -> ConfigType:
raise cv.Invalid(
f"{CONF_ENTITY_ID} is required unless this entity is a wizard input"
)
return config
def setup_home_assistant_entity(var: MockObj, config: ConfigType) -> None:
+2 -3
View File
@@ -599,7 +599,7 @@ def legacy_platform_migration_warning(
domain: str, platform: str, removal_version: str
) -> tuple[
Callable[[ConfigType], ConfigType],
Callable[[ConfigType], ConfigType],
Callable[[ConfigType], None],
]:
"""Build the per-entry capture and one-shot warning validators for a
deprecated top-level component that is now an ``image:`` platform.
@@ -618,7 +618,7 @@ def legacy_platform_migration_warning(
data.setdefault(entries_key, []).append(dict(config))
return config
def finalize(config: ConfigType) -> ConfigType:
def finalize(config: ConfigType) -> None:
data = CORE.data.setdefault(domain, {})
if not data.get(shown_key):
data[shown_key] = True
@@ -639,7 +639,6 @@ def legacy_platform_migration_warning(
domain,
yaml_util.dump({DOMAIN: migrated}),
)
return config
return capture, finalize
+1 -1
View File
@@ -264,7 +264,7 @@ def _final_validate(config: ConfigType) -> None:
# If no drawing methods are configured, and LVGL is not enabled, show a test card
config[CONF_SHOW_TEST_CARD] = True
if CONF_SPI_ID in config:
config = spi.final_validate_device_schema(
spi.final_validate_device_schema(
"mipi_rgb", require_miso=False, require_mosi=True
)(config)
+1 -2
View File
@@ -317,7 +317,7 @@ def _has_pairing_method(config: ConfigType) -> bool:
return CONF_STATIC_PAIRING_CODE in config or _offers_dynamic_pairing_code(config)
def _final_validate(config: ConfigType) -> ConfigType:
def _final_validate(config: ConfigType) -> None:
dynamic_code = _offers_dynamic_pairing_code(config)
# The protocol allows only one pairing code method.
if dynamic_code and CONF_STATIC_PAIRING_CODE in config:
@@ -334,7 +334,6 @@ def _final_validate(config: ConfigType) -> ConfigType:
CONF_UNPAIRED_ACCESS,
CONF_STATIC_PAIRING_CODE,
)
return config
FINAL_VALIDATE_SCHEMA = _final_validate
@@ -65,7 +65,7 @@ CONFIG_SCHEMA = cv.All(
)
def _final_validate(config: ConfigType) -> ConfigType:
def _final_validate(config: ConfigType) -> None:
full_config = fv.full_config.get()
switch_type = config[CONF_TYPE]
same_type = [
@@ -77,7 +77,7 @@ def _final_validate(config: ConfigType) -> ConfigType:
if len(same_type) > 1:
raise cv.Invalid(f"Only one sendspin '{switch_type}' switch is allowed")
if switch_type != CONF_UNPAIRED_ACCESS:
return config
return
hub_config = full_config.get(DOMAIN, {})
if CONF_UNPAIRED_ACCESS in hub_config:
raise cv.Invalid(
@@ -91,7 +91,6 @@ def _final_validate(config: ConfigType) -> ConfigType:
CONF_UNPAIRED_ACCESS,
CONF_STATIC_PAIRING_CODE,
)
return config
FINAL_VALIDATE_SCHEMA = _final_validate
+1 -2
View File
@@ -76,13 +76,12 @@ CONFIG_SCHEMA = (
)
def _final_validate(config: ConfigType) -> ConfigType:
def _final_validate(config: ConfigType) -> None:
is_usb = is_usb_uart_channel(config[CONF_UART_ID], fv.full_config.get())
if config[CONF_PORT_TYPE] == "USB_SERIAL" and not is_usb:
raise cv.Invalid(
f"{CONF_PORT_TYPE} USB_SERIAL requires {CONF_UART_ID} to be a usb_uart channel"
)
return config
FINAL_VALIDATE_SCHEMA = _final_validate
+2 -3
View File
@@ -539,13 +539,13 @@ def _walk_config(value: Any, path: tuple[Any, ...] = ()):
yield from _walk_config(child, (*path, index))
def _final_validate(config: Any) -> Any:
def _final_validate(config: Any) -> None:
buses = config if isinstance(config, list) else [config]
software_bus_ids = {
bus[CONF_ID] for bus in buses if CONF_INTERFACE_INDEX not in bus
}
if not software_bus_ids:
return config
return
for candidate, path in _walk_config(fv.full_config.get()):
if (
candidate.get(CONF_PSRAM_DMA)
@@ -553,7 +553,6 @@ def _final_validate(config: Any) -> Any:
):
with cv.prepend_path([cv.ROOT_CONFIG_PATH, *path, CONF_PSRAM_DMA]):
raise cv.Invalid("psram_dma requires a hardware SPI interface")
return config
FINAL_VALIDATE_SCHEMA = _final_validate
+2 -3
View File
@@ -55,13 +55,13 @@ CONFIG_SCHEMA = cv.Schema(
)
def _final_validate(config: ConfigType) -> ConfigType:
def _final_validate(config: ConfigType) -> None:
fconf = fv.full_config.get()
hub_path = fconf.get_path_for_id(config[CONF_TAS58XX_ID])
hub_conf = fconf.get_config_for_path(hub_path[:-1])
if hub_conf[CONF_MODEL] == DAC_TAS5825M:
return config
return
unsupported = sorted(frozenset(TAS5825M_ONLY_FAULT_SENSORS).intersection(config))
if unsupported:
@@ -69,7 +69,6 @@ def _final_validate(config: ConfigType) -> ConfigType:
f"binary_sensor platform tas58xx: '{', '.join(unsupported)}' is only available for 'model: {DAC_TAS5825M}' "
f"- Remove from YAML for 'model: {hub_conf[CONF_MODEL]}'"
)
return config
FINAL_VALIDATE_SCHEMA = _final_validate
+1 -2
View File
@@ -36,7 +36,7 @@ CONFIG_SCHEMA = (
)
def _final_validate(config: ConfigType) -> ConfigType:
def _final_validate(config: ConfigType) -> None:
# Historical mode runs at 1200 baud, standard mode at 9600 baud.
baud_rate = 1200 if config[CONF_HISTORICAL_MODE] else 9600
uart.final_validate_device_schema(
@@ -46,7 +46,6 @@ def _final_validate(config: ConfigType) -> ConfigType:
parity="EVEN",
stop_bits=1,
)(config)
return config
FINAL_VALIDATE_SCHEMA = _final_validate
+1 -2
View File
@@ -49,12 +49,11 @@ def _validate_factory_default_address(config: ConfigType) -> ConfigType:
return config
def _final_validate(config: ConfigType) -> ConfigType:
def _final_validate(config: ConfigType) -> None:
if CONF_FACTORY_RESET in config:
fv.id_declaration_match_schema(_validate_factory_default_address)(
config[CONF_TFLUNA_ID]
)
return config
FINAL_VALIDATE_SCHEMA = _final_validate
@@ -56,9 +56,9 @@ CONFIG_SCHEMA = cv.Schema(
).extend(cv.polling_component_schema("60s"))
def _final_validate(config: ConfigType) -> ConfigType:
def _final_validate(config: ConfigType) -> None:
if CONF_ACTUAL_CURRENT not in config:
return config
return
full_config = fv.full_config.get()
path = full_config.get_path_for_id(config[CONF_TMC2209_ID])[:-1]
if full_config.get_config_for_path(path).get(CONF_ANALOG_CURRENT_SCALE):
@@ -67,7 +67,6 @@ def _final_validate(config: ConfigType) -> ConfigType:
"the current then depends on VREF",
path=[CONF_ACTUAL_CURRENT],
)
return config
FINAL_VALIDATE_SCHEMA = _final_validate
+1 -2
View File
@@ -110,7 +110,7 @@ def _validate_analog_current_scale(config: ConfigType) -> ConfigType:
return config
def final_validate(config: ConfigType) -> ConfigType:
def final_validate(config: ConfigType) -> None:
"""Reject two drivers of the family with the same address on one UART bus."""
for other in fv.full_config.get().get("stepper", []):
if other is config:
@@ -123,7 +123,6 @@ def final_validate(config: ConfigType) -> ConfigType:
f"Address {config[CONF_ADDRESS]} is already used by '{other[CONF_ID]}' "
"on the same UART bus"
)
return config
async def new_tmc22xx(config: ConfigType) -> cg.Pvariable:
+2 -2
View File
@@ -14,8 +14,8 @@ CONFIG_SCHEMA = binary_sensor.binary_sensor_schema().extend(
)
def _final_validate(config: ConfigType) -> ConfigType:
return validate_unique_child(config, "binary_sensor")
def _final_validate(config: ConfigType) -> None:
validate_unique_child(config, "binary_sensor")
FINAL_VALIDATE_SCHEMA = _final_validate
+1 -2
View File
@@ -51,12 +51,11 @@ CONFIG_SCHEMA = cv.All(
)
def _final_validate(config: ConfigType) -> ConfigType:
def _final_validate(config: ConfigType) -> None:
validate_unique_child(config, "number")
full = fv.full_config.get()
parent = full.get_config_for_path(full.get_path_for_id(config[CONF_TMP102_ID])[:-1])
add_number_parent(config[CONF_TMP102_ID], parent.get(CONF_EXTENDED_MODE, False))
return config
FINAL_VALIDATE_SCHEMA = _final_validate
+2 -2
View File
@@ -17,8 +17,8 @@ CONFIG_SCHEMA = text_sensor.text_sensor_schema(
)
def _final_validate(config: ConfigType) -> ConfigType:
return validate_unique_child(config, "text_sensor")
def _final_validate(config: ConfigType) -> None:
validate_unique_child(config, "text_sensor")
FINAL_VALIDATE_SCHEMA = _final_validate
+1 -2
View File
@@ -37,7 +37,7 @@ CONFIG_SCHEMA = cv.All(
)
def _final_validate(config: ConfigType) -> ConfigType:
def _final_validate(config: ConfigType) -> None:
# Two muxes on one bridge would each believe they own the bus.
owned = fv.full_config.get().data.setdefault(DOMAIN, set())
bridge_id = str(config[CONF_BRIDGE_ID])
@@ -48,7 +48,6 @@ def _final_validate(config: ConfigType) -> ConfigType:
[CONF_BRIDGE_ID],
)
owned.add(bridge_id)
return config
FINAL_VALIDATE_SCHEMA = _final_validate
+1 -2
View File
@@ -81,7 +81,7 @@ def _reject_dummy_receiver(uart_conf: ConfigType) -> ConfigType:
return uart_conf
def _final_validate(config: ConfigType) -> ConfigType:
def _final_validate(config: ConfigType) -> None:
# A second reader would split the bytes with this one, and every connect
# discards what the other reader has not read yet.
full_config = fv.full_config.get()
@@ -122,7 +122,6 @@ def _final_validate(config: ConfigType) -> ConfigType:
[CONF_PORT],
)
ports.add(port)
return config
FINAL_VALIDATE_SCHEMA = _final_validate
+1 -2
View File
@@ -93,7 +93,7 @@ def validate_usb_clients(configs: list[ConfigType]) -> list[ConfigType]:
return configs
def _final_validate(config: ConfigType) -> ConfigType:
def _final_validate(config: ConfigType) -> None:
# Any two overlapping clients could otherwise open the same device
devices = config.get(CONF_DEVICES) or []
uarts = fv.full_config.get().get("usb_uart") or []
@@ -117,7 +117,6 @@ def _final_validate(config: ConfigType) -> ConfigType:
if redundant
else "narrow its filter so it no longer matches that device",
)
return config
FINAL_VALIDATE_SCHEMA = _final_validate
+4 -2
View File
@@ -28,6 +28,7 @@ def test_warn_legacy_animation_warns_once(
) -> None:
"""The deprecation warning fires exactly once and never mutates the config."""
config: ConfigType = {"id": "test_animation", "file": "anim.gif", "type": "rgb565"}
unchanged = dict(config)
# A per-entry capture (CONFIG_SCHEMA step) records the raw entry so the
# one-shot warning can print a pasteable migrated block.
@@ -35,10 +36,11 @@ def test_warn_legacy_animation_warns_once(
with caplog.at_level(logging.WARNING):
# First call: flag not yet set -> warns and records the flag.
assert _warn_legacy_animation(config) is config
_warn_legacy_animation(config)
# Second call: flag already set -> stays silent (the dedup branch).
assert _warn_legacy_animation(config) is config
_warn_legacy_animation(config)
assert config == unchanged
assert CORE.data[DOMAIN]["legacy_warning_shown"] is True
warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
assert len(warnings) == 1
@@ -28,6 +28,7 @@ def test_warn_legacy_online_image_warns_once(
) -> None:
"""The deprecation warning fires exactly once and never mutates the config."""
config: ConfigType = {"id": "test_online_image", "url": "http://example.com/i.png"}
unchanged = dict(config)
# A per-entry capture (CONFIG_SCHEMA step) records the raw entry so the
# one-shot warning can print a pasteable migrated block.
@@ -35,10 +36,11 @@ def test_warn_legacy_online_image_warns_once(
with caplog.at_level(logging.WARNING):
# First call: flag not yet set -> warns and records the flag.
assert _warn_legacy_online_image(config) is config
_warn_legacy_online_image(config)
# Second call: flag already set -> stays silent (the dedup branch).
assert _warn_legacy_online_image(config) is config
_warn_legacy_online_image(config)
assert config == unchanged
assert CORE.data[DOMAIN]["legacy_warning_shown"] is True
warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
assert len(warnings) == 1
@@ -114,7 +114,8 @@ def test_switch_without_hub_key_accepted(
set_core_config(PlatformFramework.ESP32_IDF, full_config={DOMAIN: {"id": HUB_ID}})
config = _switch_config(CONF_UNPAIRED_ACCESS)
assert SWITCH_FINAL_VALIDATE_SCHEMA(config) is config
# Should not raise.
SWITCH_FINAL_VALIDATE_SCHEMA(config)
def test_switch_without_pairing_method_warns(
@@ -153,7 +154,8 @@ def test_enabled_switch_ignores_hub_key(set_core_config: SetCoreConfigCallable)
)
config = _switch_config("enabled")
assert SWITCH_FINAL_VALIDATE_SCHEMA(config) is config
# Should not raise.
SWITCH_FINAL_VALIDATE_SCHEMA(config)
@pytest.mark.parametrize("switch_type", ["enabled", CONF_UNPAIRED_ACCESS])
@@ -15,7 +15,8 @@ def _source_config() -> ConfigType:
def _run(full_config: dict, config: ConfigType) -> ConfigType:
token = fv.full_config.set(full_config)
try:
return _inherit_ca_certificate_path(config)
_inherit_ca_certificate_path(config)
return config
finally:
fv.full_config.reset(token)
+7 -7
View File
@@ -27,20 +27,20 @@ def _run(
full_config: dict,
config: ConfigType,
platform: str = PLATFORM_ESP32,
) -> ConfigType:
) -> None:
monkeypatch.setattr("esphome.components.espnow.get_esp32_variant", lambda: variant)
monkeypatch.setitem(CORE.data, KEY_CORE, {KEY_TARGET_PLATFORM: platform})
token = fv.full_config.set(full_config)
try:
return _validate_variant(config)
_validate_variant(config)
finally:
fv.full_config.reset(token)
def test_variant_with_native_wifi_passes(monkeypatch) -> None:
"""A variant with a native Wi-Fi PHY needs no shim; config passes through."""
"""A variant with a native Wi-Fi PHY needs no shim, so validation passes."""
config = {"id": "espnow"}
assert _run(monkeypatch, VARIANT_ESP32C3, {}, config) is config
_run(monkeypatch, VARIANT_ESP32C3, {}, config)
def test_radioless_non_p4_variant_rejected(monkeypatch) -> None:
@@ -56,15 +56,15 @@ def test_p4_without_esp32_hosted_rejected(monkeypatch) -> None:
def test_p4_with_esp32_hosted_passes(monkeypatch) -> None:
"""The P4 with esp32_hosted present validates; config passes through."""
"""The P4 with esp32_hosted present validates."""
config = {"id": "espnow"}
assert _run(monkeypatch, VARIANT_ESP32P4, {"esp32_hosted": {}}, config) is config
_run(monkeypatch, VARIANT_ESP32P4, {"esp32_hosted": {}}, config)
def test_esp8266_skips_variant_check(monkeypatch) -> None:
"""The ESP8266 has one Wi-Fi radio and no variants, so the check does not run."""
config = {"id": "espnow"}
assert _run(monkeypatch, "", {}, config, platform=PLATFORM_ESP8266) is config
_run(monkeypatch, "", {}, config, platform=PLATFORM_ESP8266)
def _esp8266_arduino(monkeypatch) -> None:
+2 -4
View File
@@ -114,13 +114,11 @@ def test_every_pair_is_compared_not_just_neighbours() -> None:
validate_usb_clients(configs)
def _run_final_validate(
devices: list[ConfigType], uarts: list[ConfigType]
) -> ConfigType:
def _run_final_validate(devices: list[ConfigType], uarts: list[ConfigType]) -> None:
config = {"id": "usb_host", "devices": devices}
token = fv.full_config.set({"usb_host": config, "usb_uart": uarts})
try:
return _final_validate(config)
_final_validate(config)
finally:
fv.full_config.reset(token)