Merge branch 'schedule_use_after_free' into integration

esphome/components/packages/__init__.py

@@ -1,3 +1,4 @@
+import logging
 from pathlib import Path
 
 from esphome import git, yaml_util
@@ -20,18 +21,41 @@ from esphome.const import (
 )
 from esphome.core import EsphomeError
 
+_LOGGER = logging.getLogger(__name__)
+
 DOMAIN = CONF_PACKAGES
 
 
-def validate_git_package(config: dict):
-    if CONF_URL not in config:
-        return config
-    config = BASE_SCHEMA(config)
-    new_config = config
+def valid_package_contents(package_config: dict):
+    """Validates that a package_config that will be merged looks as much as possible to a valid config
+    to fail early on obvious mistakes."""
+    if isinstance(package_config, dict):
+        if CONF_URL in package_config:
+            # If a URL key is found, then make sure the config conforms to a remote package schema:
+            return REMOTE_PACKAGE_SCHEMA(package_config)
+
+        # Validate manually since Voluptuous would regenerate dicts and lose metadata
+        # such as ESPHomeDataBase
+        for k, v in package_config.items():
+            if not isinstance(k, str):
+                raise cv.Invalid("Package content keys must be strings")
+            if isinstance(v, (dict, list)):
+                continue  # e.g. script: [] or logger: {level: debug}
+            if v is None:
+                continue  # e.g. web_server:
+            raise cv.Invalid("Invalid component content in package definition")
+        return package_config
+
+    raise cv.Invalid("Package contents must be a dict")
+
+
+def expand_file_to_files(config: dict):
     if CONF_FILE in config:
+        new_config = config
         new_config[CONF_FILES] = [config[CONF_FILE]]
         del new_config[CONF_FILE]
-    return new_config
+        return new_config
+    return config
 
 
 def validate_yaml_filename(value):
@@ -45,7 +69,7 @@ def validate_yaml_filename(value):
 
 def validate_source_shorthand(value):
     if not isinstance(value, str):
-        raise cv.Invalid("Shorthand only for strings")
+        raise cv.Invalid("Git URL shorthand only for strings")
 
     git_file = git.GitFile.from_shorthand(value)
 
@@ -56,10 +80,17 @@ def validate_source_shorthand(value):
     if git_file.ref:
         conf[CONF_REF] = git_file.ref
 
-    return BASE_SCHEMA(conf)
+    return REMOTE_PACKAGE_SCHEMA(conf)
 
 
-BASE_SCHEMA = cv.All(
+def deprecate_single_package(config):
+    _LOGGER.warning(
+        "Including a single package under `packages:` is deprecated. Use a list instead."
+    )
+    return config
+
+
+REMOTE_PACKAGE_SCHEMA = cv.All(
     cv.Schema(
         {
             cv.Required(CONF_URL): cv.url,
@@ -90,23 +121,30 @@ BASE_SCHEMA = cv.All(
         }
     ),
     cv.has_at_least_one_key(CONF_FILE, CONF_FILES),
+    expand_file_to_files,
 )
 
-PACKAGE_SCHEMA = cv.All(
-    cv.Any(validate_source_shorthand, BASE_SCHEMA, dict), validate_git_package
+PACKAGE_SCHEMA = cv.Any(  # A package definition is either:
+    validate_source_shorthand,  # A git URL shorthand string that expands to a remote package schema, or
+    REMOTE_PACKAGE_SCHEMA,  # a valid remote package schema, or
+    valid_package_contents,  # Something that at least looks like an actual package, e.g. {wifi:{ssid: xxx}}
+    # which will have to be fully validated later as per each component's schema.
 )
 
-CONFIG_SCHEMA = cv.Any(
+CONFIG_SCHEMA = cv.Any(  # under `packages:` we can have either:
     cv.Schema(
         {
-            str: PACKAGE_SCHEMA,
+            str: PACKAGE_SCHEMA,  # a named dict of package definitions, or
         }
     ),
-    [PACKAGE_SCHEMA],
+    [PACKAGE_SCHEMA],  # a list of package definitions, or
+    cv.All(  # a single package definition (deprecated)
+        cv.ensure_list(PACKAGE_SCHEMA), deprecate_single_package
+    ),
 )
 
 
-def _process_base_package(config: dict, skip_update: bool = False) -> dict:
+def _process_remote_package(config: dict, skip_update: bool = False) -> dict:
     # When skip_update is True, use NEVER_REFRESH to prevent updates
     actual_refresh = git.NEVER_REFRESH if skip_update else config[CONF_REFRESH]
     repo_dir, revert = git.clone_or_update(
@@ -185,7 +223,7 @@ def _process_base_package(config: dict, skip_update: bool = False) -> dict:
 def _process_package(package_config, config, skip_update: bool = False):
     recursive_package = package_config
     if CONF_URL in package_config:
-        package_config = _process_base_package(package_config, skip_update)
+        package_config = _process_remote_package(package_config, skip_update)
     if isinstance(package_config, dict):
         recursive_package = do_packages_pass(package_config, skip_update)
     return merge_config(recursive_package, config)

esphome/core/scheduler.cpp

@@ -359,8 +359,7 @@ void HOT Scheduler::call(uint32_t now) {
       std::unique_ptr<SchedulerItem> item;
       {
         LockGuard guard{this->lock_};
-        item = std::move(this->items_[0]);
-        this->pop_raw_();
+        item = this->pop_raw_locked_();
       }
 
       const char *name = item->get_name();
@@ -401,7 +400,7 @@ void HOT Scheduler::call(uint32_t now) {
     // Don't run on failed components
     if (item->component != nullptr && item->component->is_failed()) {
       LockGuard guard{this->lock_};
-      this->pop_raw_();
+      this->recycle_item_(this->pop_raw_locked_());
       continue;
     }
 
@@ -414,7 +413,7 @@ void HOT Scheduler::call(uint32_t now) {
     {
       LockGuard guard{this->lock_};
       if (is_item_removed_(item.get())) {
-        this->pop_raw_();
+        this->recycle_item_(this->pop_raw_locked_());
         this->to_remove_--;
         continue;
       }
@@ -423,7 +422,7 @@ void HOT Scheduler::call(uint32_t now) {
     // Single-threaded or multi-threaded with atomics: can check without lock
     if (is_item_removed_(item.get())) {
       LockGuard guard{this->lock_};
-      this->pop_raw_();
+      this->recycle_item_(this->pop_raw_locked_());
       this->to_remove_--;
       continue;
     }
@@ -443,14 +442,14 @@ void HOT Scheduler::call(uint32_t now) {
 
     LockGuard guard{this->lock_};
 
-    auto executed_item = std::move(this->items_[0]);
     // Only pop after function call, this ensures we were reachable
     // during the function call and know if we were cancelled.
-    this->pop_raw_();
+    auto executed_item = this->pop_raw_locked_();
 
     if (executed_item->remove) {
-      // We were removed/cancelled in the function call, stop
+      // We were removed/cancelled in the function call, recycle and continue
       this->to_remove_--;
+      this->recycle_item_(std::move(executed_item));
       continue;
     }
 
@@ -497,7 +496,7 @@ size_t HOT Scheduler::cleanup_() {
     return this->items_.size();
 
   // We must hold the lock for the entire cleanup operation because:
-  // 1. We're modifying items_ (via pop_raw_) which requires exclusive access
+  // 1. We're modifying items_ (via pop_raw_locked_) which requires exclusive access
   // 2. We're decrementing to_remove_ which is also modified by other threads
   //    (though all modifications are already under lock)
   // 3. Other threads read items_ when searching for items to cancel in cancel_item_locked_()
@@ -510,17 +509,18 @@ size_t HOT Scheduler::cleanup_() {
     if (!item->remove)
       break;
     this->to_remove_--;
-    this->pop_raw_();
+    this->recycle_item_(this->pop_raw_locked_());
   }
   return this->items_.size();
 }
-void HOT Scheduler::pop_raw_() {
+std::unique_ptr<Scheduler::SchedulerItem> HOT Scheduler::pop_raw_locked_() {
   std::pop_heap(this->items_.begin(), this->items_.end(), SchedulerItem::cmp);
 
-  // Instead of destroying, recycle the item
-  this->recycle_item_(std::move(this->items_.back()));
+  // Move the item out before popping - this is the item that was at the front of the heap
+  auto item = std::move(this->items_.back());
 
   this->items_.pop_back();
+  return item;
 }
 
 // Helper to execute a scheduler item

esphome/core/scheduler.h

@@ -219,7 +219,9 @@ class Scheduler {
   // Returns the number of items remaining after cleanup
   // IMPORTANT: This method should only be called from the main thread (loop task).
   size_t cleanup_();
-  void pop_raw_();
+  // Remove and return the front item from the heap
+  // IMPORTANT: Caller must hold the scheduler lock before calling this function.
+  std::unique_ptr<SchedulerItem> pop_raw_locked_();
 
  private:
   // Helper to cancel items by name - must be called with lock held

esphome/cpp_generator.py

@@ -19,21 +19,11 @@ from esphome.core import (
     TimePeriodNanoseconds,
     TimePeriodSeconds,
 )
-from esphome.coroutine import CoroPriority, coroutine_with_priority
 from esphome.helpers import cpp_string_escape, indent_all_but_first_and_last
 from esphome.types import Expression, SafeExpType, TemplateArgsType
 from esphome.util import OrderedDict
 from esphome.yaml_util import ESPHomeDataBase
 
-# Keys for lambda deduplication storage in CORE.data
-_KEY_LAMBDA_DEDUP = "lambda_dedup"
-_KEY_LAMBDA_DEDUP_DECLARATIONS = "lambda_dedup_declarations"
-
-# Regex patterns for static variable detection (compiled once)
-_RE_CPP_SINGLE_LINE_COMMENT = re.compile(r"//.*?$", re.MULTILINE)
-_RE_CPP_MULTI_LINE_COMMENT = re.compile(r"/\*.*?\*/", re.DOTALL)
-_RE_STATIC_VARIABLE = re.compile(r"\bstatic\s+(?!cast|assert|pointer_cast)\w+\s+\w+")
-
 
 class RawExpression(Expression):
     __slots__ = ("text",)
@@ -198,7 +188,7 @@ class LambdaExpression(Expression):
 
     def __init__(
         self, parts, parameters, capture: str = "=", return_type=None, source=None
-    ) -> None:
+    ):
         self.parts = parts
         if not isinstance(parameters, ParameterListExpression):
             parameters = ParameterListExpression(*parameters)
@@ -207,21 +197,16 @@ class LambdaExpression(Expression):
         self.capture = capture
         self.return_type = safe_exp(return_type) if return_type is not None else None
 
-    def format_body(self) -> str:
-        """Format the lambda body with source directive and content."""
-        body = ""
-        if self.source is not None:
-            body += f"{self.source.as_line_directive}\n"
-        body += self.content
-        return body
-
-    def __str__(self) -> str:
+    def __str__(self):
         # Stateless lambdas (empty capture) implicitly convert to function pointers
         # when assigned to function pointer types - no unary + needed
         cpp = f"[{self.capture}]({self.parameters})"
         if self.return_type is not None:
             cpp += f" -> {self.return_type}"
-        cpp += f" {{\n{self.format_body()}\n}}"
+        cpp += " {\n"
+        if self.source is not None:
+            cpp += f"{self.source.as_line_directive}\n"
+        cpp += f"{self.content}\n}}"
         return indent_all_but_first_and_last(cpp)
 
     @property
@@ -229,37 +214,6 @@ class LambdaExpression(Expression):
         return "".join(str(part) for part in self.parts)
 
 
-class SharedFunctionLambdaExpression(LambdaExpression):
-    """A lambda expression that references a shared deduplicated function.
-
-    This class wraps a function pointer but maintains the LambdaExpression
-    interface so calling code works unchanged.
-    """
-
-    __slots__ = ("_func_name",)
-
-    def __init__(
-        self,
-        func_name: str,
-        parameters: TemplateArgsType,
-        return_type: SafeExpType | None = None,
-    ) -> None:
-        # Initialize parent with empty parts since we're just a function reference
-        super().__init__(
-            [], parameters, capture="", return_type=return_type, source=None
-        )
-        self._func_name = func_name
-
-    def __str__(self) -> str:
-        # Just return the function name - it's already a function pointer
-        return self._func_name
-
-    @property
-    def content(self) -> str:
-        # No content, just a function reference
-        return ""
-
-
 # pylint: disable=abstract-method
 class Literal(Expression, metaclass=abc.ABCMeta):
     __slots__ = ()
@@ -629,25 +583,6 @@ def add_global(expression: SafeExpType | Statement, prepend: bool = False):
     CORE.add_global(expression, prepend)
 
 
-@coroutine_with_priority(CoroPriority.FINAL)
-async def flush_lambda_dedup_declarations() -> None:
-    """Flush all deferred lambda deduplication declarations to global scope.
-
-    This is a coroutine that runs with FINAL priority (after all components)
-    to ensure all referenced variables are declared before the shared
-    lambda functions that use them.
-    """
-    if _KEY_LAMBDA_DEDUP_DECLARATIONS not in CORE.data:
-        return
-
-    declarations = CORE.data[_KEY_LAMBDA_DEDUP_DECLARATIONS]
-    for func_declaration in declarations:
-        add_global(RawStatement(func_declaration))
-
-    # Clear the list so we don't add them again
-    CORE.data[_KEY_LAMBDA_DEDUP_DECLARATIONS] = []
-
-
 def add_library(name: str, version: str | None, repository: str | None = None):
     """Add a library to the codegen library storage.
 
@@ -721,93 +656,6 @@ async def get_variable_with_full_id(id_: ID) -> tuple[ID, "MockObj"]:
     return await CORE.get_variable_with_full_id(id_)
 
 
-def _has_static_variables(code: str) -> bool:
-    """Check if code contains static variable definitions.
-
-    Static variables in lambdas should not be deduplicated because each lambda
-    instance should have its own static variable state.
-
-    Args:
-        code: The lambda body code to check
-
-    Returns:
-        True if code contains static variable definitions
-    """
-    # Remove C++ comments to avoid false positives
-    # Remove single-line comments (// ...)
-    code_no_comments = _RE_CPP_SINGLE_LINE_COMMENT.sub("", code)
-    # Remove multi-line comments (/* ... */)
-    code_no_comments = _RE_CPP_MULTI_LINE_COMMENT.sub("", code_no_comments)
-
-    # Match: static <type> <identifier>
-    # But not: static_cast, static_assert, static_pointer_cast
-    return bool(_RE_STATIC_VARIABLE.search(code_no_comments))
-
-
-def _get_shared_lambda_name(lambda_expr: LambdaExpression) -> str | None:
-    """Get the shared function name for a lambda expression.
-
-    If an identical lambda was already generated, returns the existing shared
-    function name. Otherwise, creates a new shared function and returns its name.
-
-    Lambdas with static variables are not deduplicated to preserve their
-    independent state.
-
-    Args:
-        lambda_expr: The lambda expression to deduplicate
-
-    Returns:
-        The name of the shared function for this lambda (either existing or newly created),
-        or None if the lambda should not be deduplicated (e.g., contains static variables)
-    """
-    # Create a unique key from the lambda content, parameters, and return type
-    content = lambda_expr.content
-
-    # Don't deduplicate lambdas with static variables - each instance needs its own state
-    if _has_static_variables(content):
-        return None
-    param_str = str(lambda_expr.parameters)
-    return_str = (
-        str(lambda_expr.return_type) if lambda_expr.return_type is not None else "void"
-    )
-
-    # Use tuple of (content, params, return_type) as key
-    lambda_key = (content, param_str, return_str)
-
-    # Initialize deduplication storage in CORE.data if not exists
-    if _KEY_LAMBDA_DEDUP not in CORE.data:
-        CORE.data[_KEY_LAMBDA_DEDUP] = {}
-        # Register the flush job to run after all components (FINAL priority)
-        # This ensures all variables are declared before shared lambda functions
-        CORE.add_job(flush_lambda_dedup_declarations)
-
-    lambda_cache = CORE.data[_KEY_LAMBDA_DEDUP]
-
-    # Check if we've seen this lambda before
-    if lambda_key in lambda_cache:
-        # Return name of existing shared function
-        return lambda_cache[lambda_key]
-
-    # First occurrence - create a shared function
-    # Use the cache size as the function number
-    func_name = f"shared_lambda_{len(lambda_cache)}"
-
-    # Build the function declaration using lambda's body formatting
-    func_declaration = (
-        f"{return_str} {func_name}({param_str}) {{\n{lambda_expr.format_body()}\n}}"
-    )
-
-    # Store the declaration to be added later (after all variable declarations)
-    # We can't add it immediately because it might reference variables not yet declared
-    CORE.data.setdefault(_KEY_LAMBDA_DEDUP_DECLARATIONS, []).append(func_declaration)
-
-    # Store in cache
-    lambda_cache[lambda_key] = func_name
-
-    # Return the function name (this is the first occurrence, but we still generate shared function)
-    return func_name
-
-
 async def process_lambda(
     value: Lambda,
     parameters: TemplateArgsType,
@@ -865,19 +713,6 @@ async def process_lambda(
         location.line += value.content_offset
     else:
         location = None
-
-    # Lambda deduplication: Only deduplicate stateless lambdas (empty capture).
-    # Stateful lambdas cannot be shared as they capture different contexts.
-    # Lambdas with static variables are also not deduplicated to preserve independent state.
-    if capture == "":
-        lambda_expr = LambdaExpression(
-            parts, parameters, capture, return_type, location
-        )
-        func_name = _get_shared_lambda_name(lambda_expr)
-        if func_name is not None:
-            # Return a shared function reference instead of inline lambda
-            return SharedFunctionLambdaExpression(func_name, parameters, return_type)
-
     return LambdaExpression(parts, parameters, capture, return_type, location)
 
 

tests/component_tests/packages/test_packages.py

@@ -95,7 +95,7 @@ def test_package_invalid_dict(basic_esphome, basic_wifi):
 
 
 @pytest.mark.parametrize(
-    "package",
+    "packages",
     [
         {"package1": "github://esphome/non-existant-repo/file1.yml@main"},
         {"package2": "github://esphome/non-existant-repo/file1.yml"},
@@ -107,12 +107,12 @@ def test_package_invalid_dict(basic_esphome, basic_wifi):
         ],
     ],
 )
-def test_package_shorthand(package):
-    CONFIG_SCHEMA(package)
+def test_package_shorthand(packages):
+    CONFIG_SCHEMA(packages)
 
 
 @pytest.mark.parametrize(
-    "package",
+    "packages",
     [
         # not github
         {"package1": "someplace://esphome/non-existant-repo/file1.yml@main"},
@@ -133,9 +133,9 @@ def test_package_shorthand(package):
         [3],
     ],
 )
-def test_package_invalid(package):
+def test_package_invalid(packages):
     with pytest.raises(cv.Invalid):
-        CONFIG_SCHEMA(package)
+        CONFIG_SCHEMA(packages)
 
 
 def test_package_include(basic_wifi, basic_esphome):
@@ -155,6 +155,33 @@ def test_package_include(basic_wifi, basic_esphome):
     assert actual == expected
 
 
+def test_single_package(
+    basic_esphome,
+    basic_wifi,
+    caplog: pytest.LogCaptureFixture,
+):
+    """
+    Tests the simple case where a single package is added to the top-level config as is.
+    In this test, the CONF_WIFI config is expected to be simply added to the top-level config.
+    This tests the case where the user just put packages: !include package.yaml, not
+    part of a list or mapping of packages.
+    This behavior is deprecated, the test also checks if a warning is issued.
+    """
+    config = {CONF_ESPHOME: basic_esphome, CONF_PACKAGES: {CONF_WIFI: basic_wifi}}
+
+    expected = {CONF_ESPHOME: basic_esphome, CONF_WIFI: basic_wifi}
+
+    with caplog.at_level("WARNING"):
+        actual = packages_pass(config)
+
+    assert actual == expected
+
+    assert (
+        "Including a single package under `packages:` is deprecated. Use a list instead."
+        in caplog.text
+    )
+
+
 def test_package_append(basic_wifi, basic_esphome):
     """
     Tests the case where a key is present in both a package and top-level config.

tests/component_tests/text/test_text.py

@@ -1,6 +1,4 @@
-"""Tests for the text component."""
-
-from esphome.core import CORE
+"""Tests for the binary sensor component."""
 
 
 def test_text_is_setup(generate_main):
@@ -58,22 +56,15 @@ def test_text_config_value_mode_set(generate_main):
     assert "it_3->traits.set_mode(text::TEXT_MODE_PASSWORD);" in main_cpp
 
 
-def test_text_config_lambda_is_set(generate_main) -> None:
+def test_text_config_lamda_is_set(generate_main):
     """
-    Test if lambda is set for lambda mode (optimized with stateless lambda and deduplication)
+    Test if lambda is set for lambda mode (optimized with stateless lambda)
     """
     # Given
 
     # When
     main_cpp = generate_main("tests/component_tests/text/test_text.yaml")
 
-    # Get both global and main sections to find the shared lambda definition
-    full_cpp = CORE.cpp_global_section + main_cpp
-
     # Then
-    # Lambda is deduplicated into a shared function (reference in main section)
-    assert "it_4->set_template(shared_lambda_" in main_cpp
-    # Lambda body should be in the code somewhere
-    assert 'return std::string{"Hello"};' in full_cpp
-    # Verify the shared lambda function is defined (in global section)
-    assert "esphome::optional<std::string> shared_lambda_" in full_cpp
+    assert "it_4->set_template([]() -> esphome::optional<std::string> {" in main_cpp
+    assert 'return std::string{"Hello"};' in main_cpp

tests/unit_tests/test_lambda_dedup.py

@@ -1,286 +0,0 @@
-"""Tests for lambda deduplication in cpp_generator."""
-
-from esphome import cpp_generator as cg
-from esphome.core import CORE
-
-
-def test_deduplicate_identical_lambdas() -> None:
-    """Test that identical stateless lambdas are deduplicated."""
-    # Create two identical lambda expressions
-    lambda1 = cg.LambdaExpression(
-        parts=["return 42;"],
-        parameters=[],
-        capture="",
-        return_type=cg.RawExpression("int"),
-    )
-
-    lambda2 = cg.LambdaExpression(
-        parts=["return 42;"],
-        parameters=[],
-        capture="",
-        return_type=cg.RawExpression("int"),
-    )
-
-    # Try to deduplicate them
-    func_name1 = cg._get_shared_lambda_name(lambda1)
-    func_name2 = cg._get_shared_lambda_name(lambda2)
-
-    # Both should get the same function name (deduplication happened)
-    assert func_name1 == func_name2
-    assert func_name1 == "shared_lambda_0"
-
-
-def test_different_lambdas_not_deduplicated() -> None:
-    """Test that different lambdas get different function names."""
-    lambda1 = cg.LambdaExpression(
-        parts=["return 42;"],
-        parameters=[],
-        capture="",
-        return_type=cg.RawExpression("int"),
-    )
-
-    lambda2 = cg.LambdaExpression(
-        parts=["return 24;"],  # Different content
-        parameters=[],
-        capture="",
-        return_type=cg.RawExpression("int"),
-    )
-
-    func_name1 = cg._get_shared_lambda_name(lambda1)
-    func_name2 = cg._get_shared_lambda_name(lambda2)
-
-    # Different lambdas should get different function names
-    assert func_name1 != func_name2
-    assert func_name1 == "shared_lambda_0"
-    assert func_name2 == "shared_lambda_1"
-
-
-def test_different_return_types_not_deduplicated() -> None:
-    """Test that lambdas with different return types are not deduplicated."""
-    lambda1 = cg.LambdaExpression(
-        parts=["return 42;"],
-        parameters=[],
-        capture="",
-        return_type=cg.RawExpression("int"),
-    )
-
-    lambda2 = cg.LambdaExpression(
-        parts=["return 42;"],  # Same content
-        parameters=[],
-        capture="",
-        return_type=cg.RawExpression("float"),  # Different return type
-    )
-
-    func_name1 = cg._get_shared_lambda_name(lambda1)
-    func_name2 = cg._get_shared_lambda_name(lambda2)
-
-    # Different return types = different functions
-    assert func_name1 != func_name2
-
-
-def test_different_parameters_not_deduplicated() -> None:
-    """Test that lambdas with different parameters are not deduplicated."""
-    lambda1 = cg.LambdaExpression(
-        parts=["return x;"],
-        parameters=[("int", "x")],
-        capture="",
-        return_type=cg.RawExpression("int"),
-    )
-
-    lambda2 = cg.LambdaExpression(
-        parts=["return x;"],  # Same content
-        parameters=[("float", "x")],  # Different parameter type
-        capture="",
-        return_type=cg.RawExpression("int"),
-    )
-
-    func_name1 = cg._get_shared_lambda_name(lambda1)
-    func_name2 = cg._get_shared_lambda_name(lambda2)
-
-    # Different parameters = different functions
-    assert func_name1 != func_name2
-
-
-def test_flush_lambda_dedup_declarations() -> None:
-    """Test that deferred declarations are properly stored for later flushing."""
-    # Create a lambda which will create a deferred declaration
-    lambda1 = cg.LambdaExpression(
-        parts=["return 42;"],
-        parameters=[],
-        capture="",
-        return_type=cg.RawExpression("int"),
-    )
-
-    cg._get_shared_lambda_name(lambda1)
-
-    # Check that declaration was stored
-    assert cg._KEY_LAMBDA_DEDUP_DECLARATIONS in CORE.data
-    assert len(CORE.data[cg._KEY_LAMBDA_DEDUP_DECLARATIONS]) == 1
-
-    # Verify the declaration content is correct
-    declaration = CORE.data[cg._KEY_LAMBDA_DEDUP_DECLARATIONS][0]
-    assert "shared_lambda_0" in declaration
-    assert "return 42;" in declaration
-
-    # Note: The actual flushing happens via CORE.add_job with FINAL priority
-    # during real code generation, so we don't test that here
-
-
-def test_shared_function_lambda_expression() -> None:
-    """Test SharedFunctionLambdaExpression behaves correctly."""
-    shared_lambda = cg.SharedFunctionLambdaExpression(
-        func_name="shared_lambda_0",
-        parameters=[],
-        return_type=cg.RawExpression("int"),
-    )
-
-    # Should output just the function name
-    assert str(shared_lambda) == "shared_lambda_0"
-
-    # Should have empty capture (stateless)
-    assert shared_lambda.capture == ""
-
-    # Should have empty content (just a reference)
-    assert shared_lambda.content == ""
-
-
-def test_lambda_deduplication_counter() -> None:
-    """Test that lambda counter increments correctly."""
-    # Create 3 different lambdas
-    for i in range(3):
-        lambda_expr = cg.LambdaExpression(
-            parts=[f"return {i};"],
-            parameters=[],
-            capture="",
-            return_type=cg.RawExpression("int"),
-        )
-        func_name = cg._get_shared_lambda_name(lambda_expr)
-        assert func_name == f"shared_lambda_{i}"
-
-
-def test_lambda_format_body() -> None:
-    """Test that format_body correctly formats lambda body with source."""
-    # Without source
-    lambda1 = cg.LambdaExpression(
-        parts=["return 42;"],
-        parameters=[],
-        capture="",
-        return_type=None,
-        source=None,
-    )
-    assert lambda1.format_body() == "return 42;"
-
-    # With source would need a proper source object, skip for now
-
-
-def test_stateful_lambdas_not_deduplicated() -> None:
-    """Test that stateful lambdas (non-empty capture) are not deduplicated."""
-    # _get_shared_lambda_name is only called for stateless lambdas (capture == "")
-    # Stateful lambdas bypass deduplication entirely in process_lambda
-
-    # Verify that a stateful lambda would NOT get deduplicated
-    # by checking it's not in the stateless dedup cache
-    stateful_lambda = cg.LambdaExpression(
-        parts=["return x + y;"],
-        parameters=[],
-        capture="=",  # Non-empty capture means stateful
-        return_type=cg.RawExpression("int"),
-    )
-
-    # Stateful lambdas should NOT be passed to _get_shared_lambda_name
-    # This is enforced by the `if capture == ""` check in process_lambda
-    # We verify the lambda has a non-empty capture
-    assert stateful_lambda.capture != ""
-    assert stateful_lambda.capture == "="
-
-
-def test_static_variable_detection() -> None:
-    """Test detection of static variables in lambda code."""
-    # Should detect static variables
-    assert cg._has_static_variables("static int counter = 0;")
-    assert cg._has_static_variables("static bool flag = false; return flag;")
-    assert cg._has_static_variables("  static  float  value  =  1.0;  ")
-
-    # Should NOT detect static_cast, static_assert, etc. (with underscores)
-    assert not cg._has_static_variables("return static_cast<int>(value);")
-    assert not cg._has_static_variables("static_assert(sizeof(int) == 4);")
-    assert not cg._has_static_variables("auto ptr = static_pointer_cast<Foo>(bar);")
-
-    # Edge case: 'cast', 'assert', 'pointer_cast' are NOT C++ keywords
-    # Someone could use them as type names, but we should NOT flag them
-    # because they're not actually static variables with state
-    # NOTE: These are valid C++ but extremely unlikely in ESPHome lambdas
-    assert not cg._has_static_variables("static cast obj;")  # 'cast' as type name
-    assert not cg._has_static_variables("static assert value;")  # 'assert' as type name
-    assert not cg._has_static_variables(
-        "static pointer_cast ptr;"
-    )  # 'pointer_cast' as type
-
-    # Should NOT detect in comments
-    assert not cg._has_static_variables("// static int x = 0;\nreturn 42;")
-    assert not cg._has_static_variables("/* static int y = 0; */ return 42;")
-
-    # Should detect even with comments elsewhere
-    assert cg._has_static_variables("// comment\nstatic int x = 0;\nreturn x;")
-
-    # Should NOT detect non-static code
-    assert not cg._has_static_variables("int counter = 0; return counter++;")
-    assert not cg._has_static_variables("return 42;")
-
-    # Should handle newlines between static and type/variable
-    assert cg._has_static_variables("static int\nfoo = 0;")
-    assert cg._has_static_variables("static\nint\nbar = 0;")
-    assert cg._has_static_variables(
-        "static  int  \n  foo  =  0;"
-    )  # Mixed spaces/newlines
-
-
-def test_lambdas_with_static_not_deduplicated() -> None:
-    """Test that lambdas with static variables are not deduplicated."""
-    # Two identical lambdas with static variables
-    lambda1 = cg.LambdaExpression(
-        parts=["static int counter = 0; return counter++;"],
-        parameters=[],
-        capture="",
-        return_type=cg.RawExpression("int"),
-    )
-
-    lambda2 = cg.LambdaExpression(
-        parts=["static int counter = 0; return counter++;"],
-        parameters=[],
-        capture="",
-        return_type=cg.RawExpression("int"),
-    )
-
-    # Should return None (not deduplicated)
-    func_name1 = cg._get_shared_lambda_name(lambda1)
-    func_name2 = cg._get_shared_lambda_name(lambda2)
-
-    assert func_name1 is None
-    assert func_name2 is None
-
-
-def test_lambdas_without_static_still_deduplicated() -> None:
-    """Test that lambdas without static variables are still deduplicated."""
-    # Two identical lambdas WITHOUT static variables
-    lambda1 = cg.LambdaExpression(
-        parts=["int counter = 0; return counter++;"],  # No static
-        parameters=[],
-        capture="",
-        return_type=cg.RawExpression("int"),
-    )
-
-    lambda2 = cg.LambdaExpression(
-        parts=["int counter = 0; return counter++;"],  # No static
-        parameters=[],
-        capture="",
-        return_type=cg.RawExpression("int"),
-    )
-
-    # Should be deduplicated (same function name)
-    func_name1 = cg._get_shared_lambda_name(lambda1)
-    func_name2 = cg._get_shared_lambda_name(lambda2)
-
-    assert func_name1 is not None
-    assert func_name2 is not None
-    assert func_name1 == func_name2