esphome/tests/unit_tests/test_core.py

import os
from pathlib import Path
from unittest.mock import patch

from hypothesis import given
import pytest
from strategies import mac_addr_strings

from esphome import const, core
from esphome.cpp_generator import (
    AssignmentExpression,
    CallExpression,
    ComponentMarker,
    ExpressionStatement,
    IIFEUnsafeStatement,
    MockObj,
    RawExpression,
    RawStatement,
)


class TestHexInt:
    @pytest.mark.parametrize(
        "value, expected",
        (
            (1, "0x01"),
            (255, "0xFF"),
            (128, "0x80"),
            (256, "0x100"),
            (-1, "-0x01"),  # TODO: this currently fails
        ),
    )
    def test_str(self, value, expected):
        target = core.HexInt(value)

        actual = str(target)

        assert actual == expected


class TestMACAddress:
    @given(value=mac_addr_strings())
    def test_init__valid(self, value):
        core.MACAddress(*value.split(":"))

    @pytest.mark.parametrize("value", ("1:2:3:4:5", "localhost", ""))
    def test_init__invalid(self, value):
        with pytest.raises(ValueError, match="MAC Address must consist of 6 items"):
            core.MACAddress(*value.split(":"))

    @given(value=mac_addr_strings())
    def test_str(self, value):
        target = core.MACAddress(*(int(v, 16) for v in value.split(":")))

        actual = str(target)

        assert actual == value

    def test_as_hex(self):
        target = core.MACAddress(0xDE, 0xAD, 0xBE, 0xEF, 0x00, 0xFF)

        actual = target.as_hex

        assert actual.text == "0xDEADBEEF00FFULL"


@pytest.mark.parametrize("value", (1, 2, -1, 0, 1.0, -1.0, 42.0009, -42.0009))
def test_is_approximately_integer__in_range(value):
    actual = core.is_approximately_integer(value)

    assert actual is True


@pytest.mark.parametrize("value", (42.01, -42.01, 1.5))
def test_is_approximately_integer__not_in_range(value):
    actual = core.is_approximately_integer(value)

    assert actual is False


class TestTimePeriod:
    @pytest.mark.parametrize(
        "kwargs, expected",
        (
            ({}, {}),
            ({"microseconds": 1}, {"microseconds": 1}),
            ({"microseconds": 1.0001}, {"microseconds": 1}),
            ({"milliseconds": 2}, {"milliseconds": 2}),
            ({"milliseconds": 2.0001}, {"milliseconds": 2}),
            ({"milliseconds": 2.01}, {"milliseconds": 2, "microseconds": 10}),
            ({"seconds": 3}, {"seconds": 3}),
            ({"seconds": 3.0001}, {"seconds": 3}),
            ({"seconds": 3.01}, {"seconds": 3, "milliseconds": 10}),
            ({"minutes": 4}, {"minutes": 4}),
            ({"minutes": 4.0001}, {"minutes": 4}),
            ({"minutes": 4.1}, {"minutes": 4, "seconds": 6}),
            ({"hours": 5}, {"hours": 5}),
            ({"hours": 5.0001}, {"hours": 5}),
            ({"hours": 5.1}, {"hours": 5, "minutes": 6}),
            ({"days": 6}, {"days": 6}),
            ({"days": 6.0001}, {"days": 6}),
            ({"days": 6.1}, {"days": 6, "hours": 2, "minutes": 24}),
        ),
    )
    def test_init(self, kwargs, expected):
        target = core.TimePeriod(**kwargs)

        actual = target.as_dict()

        assert actual == expected

    def test_init__nanoseconds_with_fraction(self):
        with pytest.raises(ValueError, match="Maximum precision is nanoseconds"):
            core.TimePeriod(nanoseconds=1.1)

    @pytest.mark.parametrize(
        "kwargs, expected",
        (
            ({}, "0s"),
            ({"nanoseconds": 1}, "1ns"),
            ({"nanoseconds": 1.0001}, "1ns"),
            ({"microseconds": 1}, "1us"),
            ({"microseconds": 1.0001}, "1us"),
            ({"milliseconds": 2}, "2ms"),
            ({"milliseconds": 2.0001}, "2ms"),
            ({"milliseconds": 2.01}, "2010us"),
            ({"seconds": 3}, "3s"),
            ({"seconds": 3.0001}, "3s"),
            ({"seconds": 3.01}, "3010ms"),
            ({"minutes": 4}, "4min"),
            ({"minutes": 4.0001}, "4min"),
            ({"minutes": 4.1}, "246s"),
            ({"hours": 5}, "5h"),
            ({"hours": 5.0001}, "5h"),
            ({"hours": 5.1}, "306min"),
            ({"days": 6}, "6d"),
            ({"days": 6.0001}, "6d"),
            ({"days": 6.1}, "8784min"),
        ),
    )
    def test_str(self, kwargs, expected):
        target = core.TimePeriod(**kwargs)

        actual = str(target)

        assert actual == expected

    @pytest.mark.parametrize(
        "comparison, other, expected",
        (
            ("__eq__", core.TimePeriod(microseconds=900), False),
            ("__eq__", core.TimePeriod(milliseconds=1), True),
            ("__eq__", core.TimePeriod(microseconds=1100), False),
            ("__eq__", 1000, NotImplemented),
            ("__eq__", "1000", NotImplemented),
            ("__eq__", True, NotImplemented),
            ("__eq__", object(), NotImplemented),
            ("__eq__", None, NotImplemented),
            ("__ne__", core.TimePeriod(microseconds=900), True),
            ("__ne__", core.TimePeriod(milliseconds=1), False),
            ("__ne__", core.TimePeriod(microseconds=1100), True),
            ("__ne__", 1000, NotImplemented),
            ("__ne__", "1000", NotImplemented),
            ("__ne__", True, NotImplemented),
            ("__ne__", object(), NotImplemented),
            ("__ne__", None, NotImplemented),
            ("__lt__", core.TimePeriod(microseconds=900), False),
            ("__lt__", core.TimePeriod(milliseconds=1), False),
            ("__lt__", core.TimePeriod(microseconds=1100), True),
            ("__lt__", 1000, NotImplemented),
            ("__lt__", "1000", NotImplemented),
            ("__lt__", True, NotImplemented),
            ("__lt__", object(), NotImplemented),
            ("__lt__", None, NotImplemented),
            ("__gt__", core.TimePeriod(microseconds=900), True),
            ("__gt__", core.TimePeriod(milliseconds=1), False),
            ("__gt__", core.TimePeriod(microseconds=1100), False),
            ("__gt__", 1000, NotImplemented),
            ("__gt__", "1000", NotImplemented),
            ("__gt__", True, NotImplemented),
            ("__gt__", object(), NotImplemented),
            ("__gt__", None, NotImplemented),
            ("__le__", core.TimePeriod(microseconds=900), False),
            ("__le__", core.TimePeriod(milliseconds=1), True),
            ("__le__", core.TimePeriod(microseconds=1100), True),
            ("__le__", 1000, NotImplemented),
            ("__le__", "1000", NotImplemented),
            ("__le__", True, NotImplemented),
            ("__le__", object(), NotImplemented),
            ("__le__", None, NotImplemented),
            ("__ge__", core.TimePeriod(microseconds=900), True),
            ("__ge__", core.TimePeriod(milliseconds=1), True),
            ("__ge__", core.TimePeriod(microseconds=1100), False),
            ("__ge__", 1000, NotImplemented),
            ("__ge__", "1000", NotImplemented),
            ("__ge__", True, NotImplemented),
            ("__ge__", object(), NotImplemented),
            ("__ge__", None, NotImplemented),
        ),
    )
    def test_comparison(self, comparison, other, expected):
        target = core.TimePeriod(microseconds=1000)

        actual = getattr(target, comparison)(other)

        assert actual == expected


SAMPLE_LAMBDA = """
it.strftime(64, 0, id(my_font), TextAlign::TOP_CENTER, "%H:%M:%S", id(esptime).now());
it.printf(64, 16, id(my_font2), TextAlign::TOP_CENTER, "%.1f°C (%.1f%%)", id( office_tmp ).state, id(office_hmd).state);
//id(my_commented_id)
int x = 4;/* id(my_commented_id2)
id(my_commented_id3)
*/
"""


class TestLambda:
    def test_init__copy_initializer(self):
        value = core.Lambda("foo")
        target = core.Lambda(value)

        assert str(target) is value.value

    def test_parts(self):
        target = core.Lambda(SAMPLE_LAMBDA.strip())

        # Check cache
        assert target._parts is None
        actual = target.parts
        assert target._parts is actual
        assert target.parts is actual

        assert actual == [
            "it.strftime(64, 0, ",
            "my_font",
            "",
            ', TextAlign::TOP_CENTER, "%H:%M:%S", ',
            "esptime",
            ".",
            "now());\nit.printf(64, 16, ",
            "my_font2",
            "",
            ', TextAlign::TOP_CENTER, "%.1f°C (%.1f%%)", ',
            "office_tmp",
            ".",
            "state, ",
            "office_hmd",
            ".",
            "state);\n \nint x = 4; ",
        ]

    def test_requires_ids(self):
        target = core.Lambda(SAMPLE_LAMBDA.strip())

        # Check cache
        assert target._requires_ids is None
        actual = target.requires_ids
        assert target._requires_ids is actual
        assert target.requires_ids is actual

        assert actual == [
            core.ID("my_font"),
            core.ID("esptime"),
            core.ID("my_font2"),
            core.ID("office_tmp"),
            core.ID("office_hmd"),
        ]

    def test_value_setter(self):
        target = core.Lambda("")

        # Populate cache
        _ = target.parts
        _ = target.requires_ids

        target.value = SAMPLE_LAMBDA

        # Check cache has been cleared
        assert target._parts is None
        assert target._requires_ids is None

        assert target.value == SAMPLE_LAMBDA

    def test_repr(self):
        target = core.Lambda("id(var).value == 1")

        assert repr(target) == "Lambda<id(var).value == 1>"


class TestID:
    @pytest.fixture
    def target(self):
        return core.ID(None, is_declaration=True, type="binary_sensor::Example")

    @pytest.mark.parametrize(
        "id, is_manual, expected",
        (
            ("foo", None, True),
            (None, None, False),
            ("foo", True, True),
            ("foo", False, False),
            (None, True, True),
        ),
    )
    def test_init__resolve_is_manual(self, id, is_manual, expected):
        target = core.ID(id, is_manual=is_manual)

        assert target.is_manual == expected

    @pytest.mark.parametrize(
        "registered_ids, expected",
        (
            ([], "binary_sensor_example"),
            (["binary_sensor_example"], "binary_sensor_example_2"),
            (["foo"], "binary_sensor_example"),
            (
                ["binary_sensor_example", "foo", "binary_sensor_example_2"],
                "binary_sensor_example_3",
            ),
        ),
    )
    def test_resolve(self, target, registered_ids, expected):
        actual = target.resolve(registered_ids)

        assert actual == expected
        assert str(target) == expected

    def test_copy(self, target):
        target.resolve([])

        actual = target.copy()

        assert actual is not target
        assert all(
            getattr(actual, n) == getattr(target, n)
            for n in ("id", "is_declaration", "type", "is_manual")
        )

    @pytest.mark.parametrize(
        "comparison, other, expected",
        (
            ("__eq__", core.ID(id="foo"), True),
            ("__eq__", core.ID(id="bar"), False),
            ("__eq__", 1000, NotImplemented),
            ("__eq__", "1000", NotImplemented),
            ("__eq__", True, NotImplemented),
            ("__eq__", object(), NotImplemented),
            ("__eq__", None, NotImplemented),
        ),
    )
    def test_comparison(self, comparison, other, expected):
        target = core.ID(id="foo")

        actual = getattr(target, comparison)(other)

        assert actual == expected


class TestDocumentLocation:
    @pytest.fixture
    def target(self):
        return core.DocumentLocation(
            document="foo.txt",
            line=10,
            column=20,
        )

    def test_str(self, target):
        actual = str(target)

        assert actual == "foo.txt 10:20"


class TestDocumentRange:
    @pytest.fixture
    def target(self):
        return core.DocumentRange(
            core.DocumentLocation(
                document="foo.txt",
                line=10,
                column=20,
            ),
            core.DocumentLocation(
                document="foo.txt",
                line=15,
                column=12,
            ),
        )

    def test_str(self, target):
        actual = str(target)

        assert actual == "[foo.txt 10:20 - foo.txt 15:12]"


class TestDefine:
    @pytest.mark.parametrize(
        "name, value, prop, expected",
        (
            ("ANSWER", None, "as_build_flag", "-DANSWER"),
            ("ANSWER", None, "as_macro", "#define ANSWER"),
            ("ANSWER", None, "as_tuple", ("ANSWER", None)),
            ("ANSWER", 42, "as_build_flag", "-DANSWER=42"),
            ("ANSWER", 42, "as_macro", "#define ANSWER 42"),
            ("ANSWER", 42, "as_tuple", ("ANSWER", 42)),
        ),
    )
    def test_properties(self, name, value, prop, expected):
        target = core.Define(name, value)

        actual = getattr(target, prop)

        assert actual == expected

    @pytest.mark.parametrize(
        "comparison, other, expected",
        (
            ("__eq__", core.Define(name="FOO", value=42), True),
            ("__eq__", core.Define(name="FOO", value=13), False),
            ("__eq__", core.Define(name="FOO"), False),
            ("__eq__", core.Define(name="BAR", value=42), False),
            ("__eq__", core.Define(name="BAR"), False),
            ("__eq__", 1000, NotImplemented),
            ("__eq__", "1000", NotImplemented),
            ("__eq__", True, NotImplemented),
            ("__eq__", object(), NotImplemented),
            ("__eq__", None, NotImplemented),
        ),
    )
    def test_comparison(self, comparison, other, expected):
        target = core.Define(name="FOO", value=42)

        actual = getattr(target, comparison)(other)

        assert actual == expected


class TestLibrary:
    @pytest.mark.parametrize(
        "name, version, repository, prop, expected",
        (
            ("mylib", None, None, "as_lib_dep", "mylib"),
            ("mylib", None, None, "as_tuple", ("mylib", None, None)),
            ("mylib", "1.2.3", None, "as_lib_dep", "mylib@1.2.3"),
            ("mylib", "1.2.3", None, "as_tuple", ("mylib", "1.2.3", None)),
            ("mylib", None, "file:///test", "as_lib_dep", "mylib=file:///test"),
            (
                "mylib",
                None,
                "file:///test",
                "as_tuple",
                ("mylib", None, "file:///test"),
            ),
        ),
    )
    def test_properties(self, name, version, repository, prop, expected):
        target = core.Library(name, version, repository)

        actual = getattr(target, prop)

        assert actual == expected

    @pytest.mark.parametrize(
        "comparison, other, expected",
        (
            ("__eq__", core.Library(name="libfoo", version="1.2.3"), True),
            ("__eq__", core.Library(name="libfoo", version="1.2.4"), False),
            ("__eq__", core.Library(name="libbar", version="1.2.3"), False),
            (
                "__eq__",
                core.Library(name="libbar", version=None, repository="file:///test"),
                False,
            ),
            ("__eq__", 1000, NotImplemented),
            ("__eq__", "1000", NotImplemented),
            ("__eq__", True, NotImplemented),
            ("__eq__", object(), NotImplemented),
            ("__eq__", None, NotImplemented),
        ),
    )
    def test_comparison(self, comparison, other, expected):
        target = core.Library(name="libfoo", version="1.2.3")

        actual = getattr(target, comparison)(other)

        assert actual == expected

    @pytest.mark.parametrize(
        "target, other, result, exception",
        (
            (core.Library("libfoo", None), core.Library("libfoo", None), True, None),
            (
                core.Library("libfoo", "1.2.3"),
                core.Library("libfoo", "1.2.3"),
                True,  # target is unchanged
                None,
            ),
            (
                core.Library("libfoo", None),
                core.Library("libfoo", "1.2.3"),
                False,  # Use version from other
                None,
            ),
            (
                core.Library("libfoo", "1.2.3"),
                core.Library("libfoo", "1.2.4"),
                False,
                ValueError,  # Version mismatch
            ),
            (
                core.Library("libfoo", "1.2.3"),
                core.Library("libbar", "1.2.3"),
                False,
                ValueError,  # Name mismatch
            ),
            (
                core.Library(
                    "libfoo", "1.2.4", "https://github.com/esphome/ESPAsyncWebServer"
                ),
                core.Library("libfoo", "1.2.3"),
                True,  # target is unchanged due to having a repository
                None,
            ),
            (
                core.Library("libfoo", "1.2.3"),
                core.Library(
                    "libfoo", "1.2.4", "https://github.com/esphome/ESPAsyncWebServer"
                ),
                False,  # use other due to having a repository
                None,
            ),
        ),
    )
    def test_reconcile(self, target, other, result, exception):
        if exception is not None:
            with pytest.raises(exception):
                target.reconcile_with(other)
        else:
            expected = target if result else other
            actual = target.reconcile_with(other)
            assert actual == expected


class TestEsphomeCore:
    @pytest.fixture
    def target(self, fixture_path):
        target = core.EsphomeCore()
        target.build_path = Path("foo/build")
        target.config_path = Path("foo/config")
        return target

    def test_reset(self, target):
        """Call reset on target and compare to new instance"""
        other = core.EsphomeCore().__dict__

        target.reset()
        t = target.__dict__
        # ignore event loop
        del other["event_loop"]
        del t["event_loop"]

        assert t == other

    def test_address__none(self, target):
        target.config = {}
        assert target.address is None

    def test_address__wifi(self, target):
        target.config = {}
        target.config[const.CONF_WIFI] = {const.CONF_USE_ADDRESS: "1.2.3.4"}
        target.config[const.CONF_ETHERNET] = {const.CONF_USE_ADDRESS: "4.3.2.1"}

        assert target.address == "1.2.3.4"

    def test_address__ethernet(self, target):
        target.config = {}
        target.config[const.CONF_ETHERNET] = {const.CONF_USE_ADDRESS: "4.3.2.1"}

        assert target.address == "4.3.2.1"

    def test_address__openthread(self, target):
        target.config = {}
        target.config[const.CONF_OPENTHREAD] = {
            const.CONF_USE_ADDRESS: "test-device.local"
        }
        target.name = "test-device"

        assert target.address == "test-device.local"

    def test_is_esp32(self, target):
        target.data[const.KEY_CORE] = {const.KEY_TARGET_PLATFORM: "esp32"}

        assert target.is_esp32 is True
        assert target.is_esp8266 is False

    def test_is_esp8266(self, target):
        target.data[const.KEY_CORE] = {const.KEY_TARGET_PLATFORM: "esp8266"}

        assert target.is_esp32 is False
        assert target.is_esp8266 is True

    @pytest.mark.skipif(os.name == "nt", reason="Unix-specific test")
    def test_data_dir_default_unix(self, target):
        """Test data_dir returns .esphome in config directory by default on Unix."""
        target.config_path = Path("/home/user/config.yaml")
        assert target.data_dir == Path("/home/user/.esphome")

    @pytest.mark.skipif(os.name != "nt", reason="Windows-specific test")
    def test_data_dir_default_windows(self, target):
        """Test data_dir returns .esphome in config directory by default on Windows."""
        target.config_path = Path("D:\\home\\user\\config.yaml")
        assert target.data_dir == Path("D:\\home\\user\\.esphome")

    def test_data_dir_ha_addon(self, target):
        """Test data_dir returns /data when running as Home Assistant addon."""
        target.config_path = Path("/config/test.yaml")

        with patch.dict(os.environ, {"ESPHOME_IS_HA_ADDON": "true"}):
            assert target.data_dir == Path("/data")

    def test_data_dir_env_override(self, target):
        """Test data_dir uses ESPHOME_DATA_DIR environment variable when set."""
        target.config_path = Path("/home/user/config.yaml")

        with patch.dict(os.environ, {"ESPHOME_DATA_DIR": "/custom/data/path"}):
            assert target.data_dir == Path("/custom/data/path")

    @pytest.mark.skipif(os.name == "nt", reason="Unix-specific test")
    def test_data_dir_priority_unix(self, target):
        """Test data_dir priority on Unix: HA addon > env var > default."""
        target.config_path = Path("/config/test.yaml")
        expected_default = "/config/.esphome"

        # Test HA addon takes priority over env var
        with patch.dict(
            os.environ,
            {"ESPHOME_IS_HA_ADDON": "true", "ESPHOME_DATA_DIR": "/custom/path"},
        ):
            assert target.data_dir == Path("/data")

        # Test env var is used when not HA addon
        with patch.dict(
            os.environ,
            {"ESPHOME_IS_HA_ADDON": "false", "ESPHOME_DATA_DIR": "/custom/path"},
        ):
            assert target.data_dir == Path("/custom/path")

        # Test default when neither is set
        with patch.dict(os.environ, {}, clear=True):
            # Ensure these env vars are not set
            os.environ.pop("ESPHOME_IS_HA_ADDON", None)
            os.environ.pop("ESPHOME_DATA_DIR", None)
            assert target.data_dir == Path(expected_default)

    @pytest.mark.skipif(os.name != "nt", reason="Windows-specific test")
    def test_data_dir_priority_windows(self, target):
        """Test data_dir priority on Windows: HA addon > env var > default."""
        target.config_path = Path("D:\\config\\test.yaml")
        expected_default = "D:\\config\\.esphome"

        # Test HA addon takes priority over env var
        with patch.dict(
            os.environ,
            {"ESPHOME_IS_HA_ADDON": "true", "ESPHOME_DATA_DIR": "/custom/path"},
        ):
            assert target.data_dir == Path("/data")

        # Test env var is used when not HA addon
        with patch.dict(
            os.environ,
            {"ESPHOME_IS_HA_ADDON": "false", "ESPHOME_DATA_DIR": "/custom/path"},
        ):
            assert target.data_dir == Path("/custom/path")

        # Test default when neither is set
        with patch.dict(os.environ, {}, clear=True):
            # Ensure these env vars are not set
            os.environ.pop("ESPHOME_IS_HA_ADDON", None)
            os.environ.pop("ESPHOME_DATA_DIR", None)
            assert target.data_dir == Path(expected_default)

    def test_web_port__none(self, target):
        """Test web_port returns None when web_server is not configured."""
        target.config = {}
        assert target.web_port is None

    def test_web_port__explicit_web_server_default_port(self, target):
        """Test web_port returns 80 when web_server is explicitly configured without port."""
        target.config = {const.CONF_WEB_SERVER: {}}
        assert target.web_port == 80

    def test_web_port__explicit_web_server_custom_port(self, target):
        """Test web_port returns custom port when web_server is configured with port."""
        target.config = {const.CONF_WEB_SERVER: {const.CONF_PORT: 8080}}
        assert target.web_port == 8080

    def test_web_port__ota_web_server_platform_only(self, target):
        """
        Test web_port returns None when ota.web_server platform is explicitly configured.

        This is a critical test for Dashboard Issue #766:
        https://github.com/esphome/dashboard/issues/766

        When ota: platform: web_server is explicitly configured (or auto-loaded by captive_portal):
        - "web_server" appears in loaded_integrations (platform name added to integrations)
        - "ota/web_server" appears in loaded_platforms
        - But CONF_WEB_SERVER is NOT in config (only the platform is loaded, not the component)
        - web_port MUST return None (no web UI available)
        - Dashboard should NOT show VISIT button

        This test ensures web_port only checks CONF_WEB_SERVER in config, not loaded_integrations.
        """
        # Simulate config with ota.web_server platform but no web_server component
        # This happens when:
        # 1. User explicitly configures: ota: - platform: web_server
        # 2. OR captive_portal auto-loads ota.web_server
        target.config = {
            const.CONF_OTA: [
                {
                    "platform": "web_server",
                    # OTA web_server platform config would be here
                }
            ],
            # Note: CONF_WEB_SERVER is NOT in config - only the OTA platform
        }
        # Even though "web_server" is in loaded_integrations due to the platform,
        # web_port must return None because the full web_server component is not configured
        assert target.web_port is None

    def test_has_at_least_one_component__none_configured(self, target):
        """Test has_at_least_one_component returns False when none of the components are configured."""
        target.config = {const.CONF_ESPHOME: {"name": "test"}, "logger": {}}

        assert target.has_at_least_one_component("wifi", "ethernet") is False

    def test_has_at_least_one_component__one_configured(self, target):
        """Test has_at_least_one_component returns True when one component is configured."""
        target.config = {const.CONF_WIFI: {}, "logger": {}}

        assert target.has_at_least_one_component("wifi", "ethernet") is True

    def test_has_at_least_one_component__multiple_configured(self, target):
        """Test has_at_least_one_component returns True when multiple components are configured."""
        target.config = {
            const.CONF_WIFI: {},
            const.CONF_ETHERNET: {},
            "logger": {},
        }

        assert (
            target.has_at_least_one_component("wifi", "ethernet", "bluetooth") is True
        )

    def test_has_at_least_one_component__single_component(self, target):
        """Test has_at_least_one_component works with a single component."""
        target.config = {const.CONF_MQTT: {}}

        assert target.has_at_least_one_component("mqtt") is True
        assert target.has_at_least_one_component("wifi") is False

    def test_has_at_least_one_component__config_not_loaded(self, target):
        """Test has_at_least_one_component raises ValueError when config is not loaded."""
        target.config = None

        with pytest.raises(ValueError, match="Config has not been loaded yet"):
            target.has_at_least_one_component("wifi")

    def test_has_networking__with_wifi(self, target):
        """Test has_networking returns True when wifi is configured."""
        target.config = {const.CONF_WIFI: {}}

        assert target.has_networking is True

    def test_has_networking__with_ethernet(self, target):
        """Test has_networking returns True when ethernet is configured."""
        target.config = {const.CONF_ETHERNET: {}}

        assert target.has_networking is True

    def test_has_networking__with_openthread(self, target):
        """Test has_networking returns True when openthread is configured."""
        target.config = {const.CONF_OPENTHREAD: {}}

        assert target.has_networking is True

    def test_has_networking__without_networking(self, target):
        """Test has_networking returns False when no networking component is configured."""
        target.config = {const.CONF_ESPHOME: {"name": "test"}, "logger": {}}

        assert target.has_networking is False

    def test_add_library__esp32_arduino_enables_disabled_library(self, target):
        """Test add_library auto-enables Arduino libraries on ESP32 Arduino builds."""
        target.data[const.KEY_CORE] = {
            const.KEY_TARGET_PLATFORM: "esp32",
            const.KEY_TARGET_FRAMEWORK: "arduino",
        }

        library = core.Library("WiFi", None)

        with patch("esphome.components.esp32._enable_arduino_library") as mock_enable:
            target.add_library(library)
            mock_enable.assert_called_once_with("WiFi")

        assert "WiFi" in target.platformio_libraries

    def test_add_library__esp32_arduino_ignores_non_arduino_library(self, target):
        """Test add_library doesn't enable libraries not in ARDUINO_DISABLED_LIBRARIES."""
        target.data[const.KEY_CORE] = {
            const.KEY_TARGET_PLATFORM: "esp32",
            const.KEY_TARGET_FRAMEWORK: "arduino",
        }

        library = core.Library("SomeOtherLib", "1.0.0")

        with patch("esphome.components.esp32._enable_arduino_library") as mock_enable:
            target.add_library(library)
            mock_enable.assert_not_called()

        assert "SomeOtherLib" in target.platformio_libraries

    def test_add_library__esp32_idf_does_not_enable_arduino_library(self, target):
        """Test add_library doesn't auto-enable Arduino libraries on ESP32 IDF builds."""
        target.data[const.KEY_CORE] = {
            const.KEY_TARGET_PLATFORM: "esp32",
            const.KEY_TARGET_FRAMEWORK: "esp-idf",
        }

        library = core.Library("WiFi", None)

        with patch("esphome.components.esp32._enable_arduino_library") as mock_enable:
            target.add_library(library)
            mock_enable.assert_not_called()

        assert "WiFi" in target.platformio_libraries

    def test_add_library__esp8266_does_not_enable_arduino_library(self, target):
        """Test add_library doesn't auto-enable Arduino libraries on ESP8266."""
        target.data[const.KEY_CORE] = {
            const.KEY_TARGET_PLATFORM: "esp8266",
            const.KEY_TARGET_FRAMEWORK: "arduino",
        }

        library = core.Library("WiFi", None)

        with patch("esphome.components.esp32._enable_arduino_library") as mock_enable:
            target.add_library(library)
            mock_enable.assert_not_called()

        assert "WiFi" in target.platformio_libraries

    def test_testing_ensure_platform_registered__sets_count(self, target):
        """Test testing_ensure_platform_registered sets count to 1 for new platform."""
        assert target.platform_counts["sensor"] == 0
        target.testing_ensure_platform_registered("sensor")
        assert target.platform_counts["sensor"] == 1

    def test_testing_ensure_platform_registered__does_not_overwrite(self, target):
        """Test testing_ensure_platform_registered preserves existing count."""
        target.platform_counts["sensor"] = 3
        target.testing_ensure_platform_registered("sensor")
        assert target.platform_counts["sensor"] == 3

    def test_add_library__extracts_short_name_from_path(self, target):
        """Test add_library extracts short name from library paths like owner/lib."""
        target.data[const.KEY_CORE] = {
            const.KEY_TARGET_PLATFORM: "esp32",
            const.KEY_TARGET_FRAMEWORK: "arduino",
        }

        library = core.Library("arduino/Wire", None)

        with patch("esphome.components.esp32._enable_arduino_library") as mock_enable:
            target.add_library(library)
            mock_enable.assert_called_once_with("Wire")

        assert "Wire" in target.platformio_libraries


def test_wrap_in_iifes_empty_input() -> None:
    assert core._wrap_in_iifes([], max_statements=10) == []


def test_wrap_in_iifes_fewer_lines_than_limit() -> None:
    lines = ["a();", "b();", "c();"]
    assert core._wrap_in_iifes(lines, max_statements=10) == [
        "[]() {",
        "a();",
        "b();",
        "c();",
        "}();",
    ]


def test_wrap_in_iifes_splits_at_max_statements() -> None:
    lines = [f"s{i}();" for i in range(5)]
    result = core._wrap_in_iifes(lines, max_statements=2)
    # With max=2 and 5 lines: chunks of 2, 2, 1 → 3 IIFEs.
    assert sum(1 for line in result if line.startswith("[]()")) == 3


def test_wrap_in_iifes_never_splits_inside_braces() -> None:
    # max=2 would naively split after "{" but brace guard keeps block whole.
    lines = ["a();", "{", "inner();", "}", "b();"]
    assert core._wrap_in_iifes(lines, max_statements=2) == [
        "[]() {",
        "a();",
        "{",
        "inner();",
        "}",
        "}();",
        "[]() {",
        "b();",
        "}();",
    ]


def test_wrap_in_iifes_nested_braces() -> None:
    lines = ["{", "{", "deep();", "}", "}", "after();"]
    assert core._wrap_in_iifes(lines, max_statements=1) == [
        "[]() {",
        "{",
        "{",
        "deep();",
        "}",
        "}",
        "}();",
        "[]() {",
        "after();",
        "}();",
    ]


def test_wrap_in_iifes_unbalanced_braces_fall_through() -> None:
    # Pathological input where "}" appears before "{": don't crash; emit
    # a single IIFE with all lines rather than splitting mid-flight.
    lines = ["a();", "}", "b();"]
    result = core._wrap_in_iifes(lines, max_statements=1)
    assert result[0] == "[]() {"
    assert result[-1] == "}();"
    assert [line for line in result if line in lines] == lines


def test_wrap_in_iifes_negative_depth_stays_poisoned() -> None:
    # Once depth goes negative the brace tracker is unreliable; a later
    # arithmetic return to depth 0 must not re-enable splitting. Here
    # "}" drives depth to -1 immediately, then "{" later brings depth
    # back to 0 arithmetically. With max=1 every statement would flush
    # if splitting were still enabled — assert everything emits as one
    # IIFE because the poisoned flag stayed set.
    lines = ["}", "a();", "{", "b();", "}", "c();"]
    result = core._wrap_in_iifes(lines, max_statements=1)
    assert sum(1 for line in result if line == "[]() {") == 1
    assert result[0] == "[]() {"
    assert result[-1] == "}();"


def test_wrap_in_iifes_never_splits_inline_brace_lines() -> None:
    # Defensive: if codegen ever emits control flow with braces on the
    # same line (if/else/for), the depth tracker should keep the whole
    # scoped block together even with aggressive max_statements.
    lines = [
        "before();",
        "if (cond) {",
        "then_branch();",
        "} else {",
        "for (;;) {",
        "loop_body();",
        "}",
        "}",
        "after();",
    ]
    assert core._wrap_in_iifes(lines, max_statements=1) == [
        "[]() {",
        "before();",
        "}();",
        "[]() {",
        "if (cond) {",
        "then_branch();",
        "} else {",
        "for (;;) {",
        "loop_body();",
        "}",
        "}",
        "}();",
        "[]() {",
        "after();",
        "}();",
    ]


def test_wrap_in_iifes_skips_comment_only_chunks() -> None:
    # A chunk with no C++ statements (only comments, e.g. a component's
    # config dump) should be emitted verbatim without a no-op IIFE.
    lines = ["// sha256:", "//   {}"]
    assert core._wrap_in_iifes(lines, max_statements=50) == lines


def test_wrap_in_iifes_ignores_iife_pattern_in_comment() -> None:
    # A comment whose text mentions "[]()" (e.g. a YAML dump of a
    # lambda) must not fool the comment-only detector into wrapping.
    lines = [
        "// on_value:",
        "//   - !lambda |-",
        "//       return []() { return 5; };",
    ]
    assert core._wrap_in_iifes(lines, max_statements=50) == lines


def test_cpp_main_section_no_components_emits_flat() -> None:
    target = core.EsphomeCore()
    target.main_statements = [RawStatement("a();"), RawStatement("b();")]
    out = target.cpp_main_section
    assert "[]() {" not in out
    assert "a();" in out
    assert "b();" in out


def test_cpp_main_section_component_marker_wraps_in_iife() -> None:
    target = core.EsphomeCore()
    target.main_statements = [
        ComponentMarker("logger"),
        RawStatement("new_logger();"),
        ComponentMarker("wifi"),
        RawStatement("new_wifi();"),
    ]
    out = target.cpp_main_section
    # One IIFE per component that emits C++ statements.
    assert out.count("[]() {") == 2
    assert out.count("}();") == 2
    # ComponentMarker produces no output of its own.
    assert "component-marker" not in out


_CHUNK_COUNT = 3  # number of sub-chunks to force when testing splitting
# Statement count that produces exactly _CHUNK_COUNT IIFEs when sub-split
# at IIFE_MAX_STATEMENTS (full chunks at the cap, no partial).
_STATEMENTS_OVER_CAP = core.IIFE_MAX_STATEMENTS * _CHUNK_COUNT


def test_cpp_main_section_scope_brace_raw_disables_sub_split() -> None:
    # A group containing scope-brace RawStatements (e.g. `{` / `}` from
    # with_local_variable) must stay in one IIFE regardless of size so
    # the scope bounds and any locals between them stay together.
    target = core.EsphomeCore()
    stmts: list = [ComponentMarker("wifi"), RawStatement("{")]
    stmts.extend(RawStatement(f"s{i}();") for i in range(_STATEMENTS_OVER_CAP))
    stmts.append(RawStatement("}"))
    target.main_statements = stmts
    out = target.cpp_main_section
    assert out.count("[]() {") == 1
    assert out.count("}();") == 1


def test_cpp_main_section_inline_comment_raw_still_sub_splits() -> None:
    # entity_helpers emits `call();  // flags` as RawStatement for inline
    # comments. Those shouldn't flag the group as scope-using — the
    # content-aware check only triggers on bare `{` / `}`.
    target = core.EsphomeCore()
    stmts: list = [ComponentMarker("sensor")]
    stmts.extend(
        RawStatement(f"s{i}();  // flags") for i in range(_STATEMENTS_OVER_CAP)
    )
    target.main_statements = stmts
    out = target.cpp_main_section
    assert out.count("[]() {") == _CHUNK_COUNT


def test_cpp_main_section_raw_expression_disables_sub_split() -> None:
    # cg.add(RawExpression(...)) — e.g. `time::ParsedTimezone tz` followed
    # by `tz.field = ...` — is raw bare text that may reference a local
    # declared elsewhere in the same group. Keep the group in one IIFE.
    target = core.EsphomeCore()
    stmts: list = [
        ComponentMarker("time"),
        ExpressionStatement(RawExpression("time::ParsedTimezone tz{}")),
    ]
    stmts.extend(
        ExpressionStatement(RawExpression(f"tz.field_{i} = {i}"))
        for i in range(_STATEMENTS_OVER_CAP)
    )
    target.main_statements = stmts
    out = target.cpp_main_section
    assert out.count("[]() {") == 1


def test_cpp_main_section_raw_expression_as_call_arg_still_sub_splits() -> None:
    # RawExpression passed as an argument to a method call (e.g.
    # `var.set_program(RawExpression("&foo"))`) produces
    # `ExpressionStatement(CallExpression(..., RawExpression))` — the
    # outer expression is a CallExpression, not a RawExpression, so
    # the group is still sub-splittable.
    target = core.EsphomeCore()
    stmts: list = [ComponentMarker("rp2040_pio_led_strip")]
    stmts.extend(
        ExpressionStatement(
            CallExpression(MockObj(f"var_{i}.set_program"), RawExpression("&foo"))
        )
        for i in range(_STATEMENTS_OVER_CAP)
    )
    target.main_statements = stmts
    out = target.cpp_main_section
    assert out.count("[]() {") == _CHUNK_COUNT


def test_cpp_main_section_typed_assignment_disables_sub_split() -> None:
    # cg.variable(id, rhs) emits `Type id = rhs;` via
    # ExpressionStatement(AssignmentExpression(type=..., ...)). That's a
    # function-local whose name must stay visible across all uses in
    # the component — no sub-split.
    target = core.EsphomeCore()
    typed_assign = ExpressionStatement(
        AssignmentExpression(MockObj("int"), "", MockObj("x"), MockObj("42"))
    )
    stmts: list = [ComponentMarker("custom"), typed_assign]
    stmts.extend(RawStatement(f"use_x_{i}();") for i in range(_STATEMENTS_OVER_CAP))
    target.main_statements = stmts
    out = target.cpp_main_section
    assert out.count("[]() {") == 1


def test_cpp_main_section_iife_unsafe_wins_over_no_split() -> None:
    # A group that triggers BOTH flags (IIFEUnsafeStatement present AND
    # a bare-local emission) must still be emitted flat — the unsafe
    # flag wins because a `return` inside any IIFE, even a single big
    # one, only exits the lambda.
    target = core.EsphomeCore()
    target.main_statements = [
        ComponentMarker("safe_mode_with_local"),
        RawStatement("{"),  # would trigger no_split
        RawStatement("new_foo();"),
        RawStatement("}"),
        IIFEUnsafeStatement(RawExpression("if (cond) return")),
    ]
    out = target.cpp_main_section
    assert "[]() {" not in out
    assert "if (cond) return;" in out


def test_cpp_main_section_iife_unsafe_statement_emits_component_flat() -> None:
    # A component that emits IIFEUnsafeStatement (e.g. safe_mode with
    # `if (...) return;`) must be emitted flat — a `return` inside an
    # IIFE would only exit the lambda, not setup().
    target = core.EsphomeCore()
    target.main_statements = [
        ComponentMarker("logger"),
        RawStatement("new_logger();"),
        ComponentMarker("safe_mode"),
        RawStatement("new_safe_mode();"),
        IIFEUnsafeStatement(RawExpression("if (entering) return")),
        ComponentMarker("sensor"),
        RawStatement("new_sensor();"),
    ]
    out = target.cpp_main_section
    # logger and sensor wrapped; safe_mode flat.
    assert out.count("[]() {") == 2
    # safe_mode's statements appear at top level, not indented in a lambda.
    assert "new_safe_mode();" in out
    assert "if (entering) return;" in out
    # The IIFEUnsafeStatement wrapper picks up the trailing semicolon
    # via statement() when inner is a bare Expression.
    assert "if (entering) return\n" not in out


def test_cpp_main_section_comment_only_component_omits_iife() -> None:
    # A component that emits only a ComponentMarker (no statements) adds
    # nothing to the generated output. A neighboring component with
    # actual code still gets its own IIFE.
    target = core.EsphomeCore()
    target.main_statements = [
        ComponentMarker("sha256"),
        ComponentMarker("wifi"),
        RawStatement("new_wifi();"),
    ]
    out = target.cpp_main_section
    assert out.count("[]() {") == 1
    assert "new_wifi();" in out


def test_cpp_main_section_prefix_statements_stay_outside_iife() -> None:
    target = core.EsphomeCore()
    target.main_statements = [
        RawStatement("prefix();"),
        ComponentMarker("c"),
        RawStatement("body();"),
    ]
    out = target.cpp_main_section
    assert out.index("prefix();") < out.index("[]() {")