[core] Add unit tests for the smaller core modules (#21333)

tests/unit_tests/build_helpers/test_idedata.py

@@ -2,10 +2,13 @@
 
 # pylint: disable=protected-access
 
+import ctypes
 import json
 import logging
 import os
 from pathlib import Path
+import shlex
+from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -128,6 +131,65 @@ def test_pick_entry_prefers_esphome_tu() -> None:
     assert idedata._pick_entry(entries)["file"].endswith("app.cpp")
 
 
+def test_pick_entry_without_cxx_tu_raises() -> None:
+    """A compile DB holding only C files has no usable translation unit."""
+    entries = [_entry(f"{ABS}build", f"{ABS}build/src/a.c", "gcc -c a.c")]
+    with pytest.raises(ValueError, match="no C\\+\\+ translation unit"):
+        idedata._pick_entry(entries)
+
+
+def test_expand_response_files_absolute_path(tmp_path: Path) -> None:
+    """An absolute response file path is read as is, not joined to the directory."""
+    rsp = tmp_path / "flags.rsp"
+    rsp.write_text("-DFROM_RSP=1")
+    other = tmp_path / "elsewhere"
+    other.mkdir()
+
+    assert idedata._expand_response_files([f"@{rsp}"], other) == ["-DFROM_RSP=1"]
+
+
+def test_parse_entry_drops_bare_input_files() -> None:
+    """A source file given without -c is not kept as a compiler flag."""
+    entry = _entry(
+        f"{ABS}build",
+        f"{ABS}build/src/esphome/x.cpp",
+        "g++ -DFOO x.cpp -std=gnu++20",
+    )
+
+    _, _, _, cxx_flags = idedata.parse_entry(entry)
+
+    assert cxx_flags == ["-std=gnu++20"]
+
+
+@pytest.mark.parametrize(
+    ("cxx_path", "expected"),
+    [
+        pytest.param("g++", "gcc", id="bare"),
+        pytest.param("/usr/bin/g++", "/usr/bin/gcc", id="path"),
+        pytest.param("xtensa-esp32-elf-g++.exe", "xtensa-esp32-elf-gcc.exe", id="exe"),
+        pytest.param("C:\\tools\\g++.exe", "C:\\tools\\gcc.exe", id="backslash"),
+        pytest.param("/usr/bin/clang++", "/usr/bin/clang++", id="clang"),
+    ],
+)
+def test_cc_path_from_cxx(cxx_path: str, expected: str) -> None:
+    """Only a g++ program name is rewritten to gcc."""
+    assert idedata._cc_path_from_cxx(cxx_path) == expected
+
+
+@pytest.mark.parametrize(
+    ("token", "expected"),
+    [
+        pytest.param("-includefoo.h", True, id="joined"),
+        pytest.param("-include", False, id="separate"),
+        pytest.param("-include-pch", False, id="pch"),
+        pytest.param("-DFOO", False, id="other"),
+    ],
+)
+def test_is_joined_include(token: str, expected: bool) -> None:
+    """Only the joined ``-includefoo.h`` spelling counts."""
+    assert idedata.is_joined_include(token) is expected
+
+
 def test_pick_entry_falls_back_to_any_cxx_tu() -> None:
     """With no ``/src/esphome/`` TU present, the first C++ entry is the fallback."""
     entries = [
@@ -345,6 +407,48 @@ def test_split_command_empty_returns_empty() -> None:
     assert idedata._split_command("   ") == []
 
 
+@pytest.fixture(name="fake_windows_argv")
+def fixture_fake_windows_argv(monkeypatch: pytest.MonkeyPatch) -> list[bool]:
+    """Run ``_split_command`` down its Windows branch on any host.
+
+    ``CommandLineToArgvW`` is replaced by a shlex based stand-in; the returned
+    list records whether ``LocalFree`` released the argv block.
+    """
+    freed: list[bool] = []
+
+    def command_line_to_argv(command: str, argc_ref: object) -> list[str]:
+        tokens = shlex.split(command)
+        argc_ref._obj.value = len(tokens)  # type: ignore[attr-defined]
+        return tokens
+
+    windll = SimpleNamespace(
+        shell32=SimpleNamespace(CommandLineToArgvW=command_line_to_argv),
+        kernel32=SimpleNamespace(LocalFree=lambda _ptr: freed.append(True)),
+    )
+    monkeypatch.setattr(ctypes, "windll", windll, raising=False)
+    monkeypatch.setattr(idedata.os, "name", "nt")
+    return freed
+
+
+def test_split_command_windows_branch(fake_windows_argv: list[bool]) -> None:
+    """On Windows the command is tokenized by the OS parser and freed after."""
+    assert idedata._split_command("g++ -DFOO=1 -c x.cpp") == [
+        "g++",
+        "-DFOO=1",
+        "-c",
+        "x.cpp",
+    ]
+    assert fake_windows_argv == [True]
+
+
+def test_split_command_windows_branch_blank_command(
+    fake_windows_argv: list[bool],
+) -> None:
+    """A blank command yields no tokens without calling the OS parser."""
+    assert idedata._split_command("  ") == []
+    assert fake_windows_argv == []
+
+
 @pytest.mark.skipif(os.name != "nt", reason="Windows argv tokenization")
 def test_parse_entry_normalizes_windows_cxx_path() -> None:
     """A backslash compiler path is emitted forward-slashed; define unescaped."""

tests/unit_tests/core/test_entity_helpers.py

@@ -7,8 +7,10 @@ from typing import Any
 
 import pytest
 
+from esphome import final_validate as fv
 import esphome.codegen as cg
 from esphome.components import button, number, select, switch
+from esphome.config import Config
 from esphome.config_validation import Invalid
 from esphome.const import (
     CONF_DEVICE_CLASS,
@@ -28,6 +30,7 @@ from esphome.core.entity_helpers import (
     entity_duplicate_validator,
     finalize_entity_strings,
     get_base_entity_object_id,
+    inherit_property_from,
     register_device_class,
     register_icon,
     register_unit_of_measurement,
@@ -1607,3 +1610,59 @@ async def test_sub_entities_create_from_whole_config(
     assert str(var) == "rf_sw"
     assert calls == [("rf_sw", "<esphome__rf_sw__fn>")]
     assert setup_test_environment == ["hub->set_rf_switch(rf_sw)"]
+
+
+def test_entity_duplicate_validator_without_name() -> None:
+    """A config with no name is returned untouched and not tracked."""
+    validator = entity_duplicate_validator("sensor")
+    config = {CONF_ID: "no_name"}
+
+    assert validator(config) is config
+    assert not CORE.unique_ids
+
+
+def test_entity_duplicate_validator_testing_mode_allows_duplicates() -> None:
+    validator = entity_duplicate_validator("sensor")
+    config = {CONF_NAME: "Temperature"}
+    validator(dict(config))
+    CORE.testing_mode = True
+    try:
+        assert validator(dict(config)) == config
+    finally:
+        CORE.testing_mode = False
+
+
+def _inherit_config() -> tuple[Config, ID, dict[str, Any]]:
+    parent_id = ID("parent", is_declaration=True)
+    child_id = ID("child", is_declaration=True)
+    parent = {CONF_ID: parent_id, CONF_ICON: "mdi:parent"}
+    child = {CONF_ID: child_id, "source": ID("parent"), "inner": {"x": 1}}
+    config = Config()
+    config["sensor"] = [parent, child]
+    config.declare_ids.append((parent_id, ["sensor", 0, CONF_ID]))
+    config.declare_ids.append((child_id, ["sensor", 1, CONF_ID]))
+    return config, child_id, child
+
+
+def test_inherit_property_from_path_inherits_value() -> None:
+    config, child_id, child = _inherit_config()
+    validator = inherit_property_from(["inner", CONF_ICON], "source")
+    token = fv.full_config.set(config)
+    try:
+        assert validator(child) is child
+    finally:
+        fv.full_config.reset(token)
+
+    assert config["sensor"][1]["inner"][CONF_ICON] == "mdi:parent"
+
+
+def test_inherit_property_from_path_missing_is_ignored() -> None:
+    config, _, child = _inherit_config()
+    validator = inherit_property_from(["absent", CONF_ICON], "source")
+    token = fv.full_config.set(config)
+    try:
+        assert validator(child) is child
+    finally:
+        fv.full_config.reset(token)
+
+    assert "absent" not in config["sensor"][1]

tests/unit_tests/test_bundle.py

@@ -22,6 +22,7 @@ from esphome.bundle import (
     _add_bytes_to_tar,
     _default_target_dir,
     _find_used_secret_keys,
+    _validate_tar_members,
     add_bundle_file,
     add_secret_scan_dir,
     extract_bundle,
@@ -1872,3 +1873,81 @@ def test_bundle_round_trip_with_secrets(tmp_path: Path) -> None:
 
     manifest = read_bundle_manifest(bundle_path)
     assert manifest.has_secrets is True
+
+
+def test_add_secret_scan_dir_relative_path(tmp_path: Path) -> None:
+    """A relative scan directory is taken as relative to the config directory."""
+    config_dir = _setup_config_dir(tmp_path)
+    (config_dir / "secrets.yaml").write_text("pw: hunter2\nunused: nope\n")
+    package_dir = config_dir / "pkg"
+    package_dir.mkdir()
+    (package_dir / "base.yml").write_text("password: !secret pw\n")
+    add_secret_scan_dir(Path("pkg"))
+
+    result = ConfigBundleCreator({}).create_bundle()
+
+    assert result.manifest[ManifestKey.HAS_SECRETS] is True
+    with tarfile.open(fileobj=io.BytesIO(result.data), mode="r:gz") as tar:
+        secrets_data = tar.extractfile("secrets.yaml").read().decode()
+    assert "hunter2" in secrets_data
+    assert "unused" not in secrets_data
+
+
+def test_discover_files_skips_system_includes_c(tmp_path: Path) -> None:
+    """System includes in includes_c are not local files."""
+    _setup_config_dir(tmp_path)
+    config: dict[str, Any] = {"esphome": {"includes_c": ["<system.h>"]}}
+
+    files = ConfigBundleCreator(config).discover_files()
+
+    assert [f.path for f in files] == ["test.yaml"]
+
+
+def test_discover_files_ignores_missing_path_and_scalars(tmp_path: Path) -> None:
+    """Absolute Path values that are not files and non-path values are ignored."""
+    _setup_config_dir(tmp_path)
+    config: dict[str, Any] = {
+        "missing": tmp_path / "does_not_exist.bin",
+        "number": 5,
+        "flag": True,
+    }
+
+    files = ConfigBundleCreator(config).discover_files()
+
+    assert [f.path for f in files] == ["test.yaml"]
+
+
+def test_build_filtered_secrets_skips_files_outside_config_dir(
+    tmp_path: Path,
+) -> None:
+    _setup_config_dir(tmp_path)
+    outside = tmp_path / "outside_secrets.yaml"
+    outside.write_text("k: v\n")
+    creator = ConfigBundleCreator({})
+    creator._secrets_paths.add(outside.resolve())
+
+    assert creator._build_filtered_secrets({"k"}) == {}
+
+
+def test_validate_tar_members_rejects_escape_through_symlinked_dir(
+    tmp_path: Path,
+) -> None:
+    """A member resolving outside the target through an existing symlink is refused."""
+    target = tmp_path / "out"
+    target.mkdir()
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    try:
+        (target / "link").symlink_to(outside, target_is_directory=True)
+    except (OSError, NotImplementedError):
+        pytest.skip("symlinks not available")
+    buf = io.BytesIO()
+    with tarfile.open(fileobj=buf, mode="w") as tar:
+        _add_bytes_to_tar(tar, "link/file.txt", b"data")
+    buf.seek(0)
+
+    with (
+        tarfile.open(fileobj=buf, mode="r") as tar,
+        pytest.raises(EsphomeError, match="outside target"),
+    ):
+        _validate_tar_members(tar, target.resolve())

tests/unit_tests/test_config_helpers.py

@@ -6,6 +6,8 @@ from unittest.mock import patch
 import pytest
 
 from esphome.config_helpers import (
+    Extend,
+    Remove,
     filter_source_files_from_defines,
     filter_source_files_from_platform,
     frameworks_for_platforms,
@@ -172,3 +174,38 @@ def test_filter_source_files_from_defines() -> None:
 
         mock_core.defines = set()
         assert sorted(filter_func()) == ["automation.cpp", "filter.cpp"]
+
+
+def test_extend_str_repr_and_equality() -> None:
+    ext = Extend("my_id")
+
+    assert str(ext) == "!extend my_id"
+    assert repr(ext) == "Extend(my_id)"
+    assert ext == Extend("my_id")
+    assert ext != Extend("other")
+    assert ext != "my_id"
+
+
+def test_remove_str_repr_and_equality() -> None:
+    rem = Remove("my_id")
+
+    assert str(rem) == "!remove my_id"
+    assert repr(rem) == "Remove(my_id)"
+    assert rem == Remove("my_id")
+    assert rem != Remove("other")
+    assert rem != "my_id"
+
+
+def test_filter_source_files_from_platform_unknown_pair() -> None:
+    """A target platform/framework pair with no PlatformFramework yields no exclusions."""
+    mock_core_data = {
+        KEY_CORE: {
+            KEY_TARGET_PLATFORM: "not_a_platform",
+            KEY_TARGET_FRAMEWORK: "arduino",
+        }
+    }
+    with patch("esphome.config_helpers.CORE.data", mock_core_data):
+        func = filter_source_files_from_platform(
+            {"a.cpp": {PlatformFramework.ESP32_ARDUINO}}
+        )
+        assert func() == []

tests/unit_tests/test_coroutine.py

@@ -1,10 +1,18 @@
 """Tests for the coroutine module."""
 
+from collections.abc import AsyncGenerator, Generator
 import contextvars
+from typing import Any
 
 import pytest
 
-from esphome.coroutine import CoroPriority, FakeEventLoop, coroutine_with_priority
+from esphome.coroutine import (
+    CoroPriority,
+    FakeEventLoop,
+    _Task,
+    coroutine,
+    coroutine_with_priority,
+)
 
 
 def test_coro_priority_enum_values() -> None:
@@ -262,3 +270,77 @@ def test_context_inherits_ambient_value_at_schedule_time() -> None:
         my_var.reset(token)
 
     assert seen == {"value": "ambient"}
+
+
+def test_coroutine_does_not_rewrap_esphome_coroutine() -> None:
+    @coroutine
+    def wrapped() -> Generator[None, None, None]:
+        yield
+
+    assert coroutine(wrapped) is wrapped
+
+
+def test_coroutine_rejects_async_generator() -> None:
+    async def agen() -> AsyncGenerator[int, None]:
+        yield 1
+
+    with pytest.raises(ValueError, match="Async generator functions are not allowed"):
+        coroutine(agen)
+
+
+def test_coroutine_flattens_awaitable_and_generator_yields() -> None:
+    results: list[Any] = []
+
+    async def native() -> int:
+        return 3
+
+    def inner() -> Generator[None, None, int]:
+        yield
+        return 4
+
+    @coroutine
+    def outer() -> Generator[Any, Any, None]:
+        results.append((yield native()))
+        results.append((yield inner()))
+
+    loop = FakeEventLoop()
+    loop.add_job(outer)
+    loop.flush_tasks()
+
+    assert results == [3, 4]
+
+
+def test_task_equality_and_ordering() -> None:
+    ctx = contextvars.copy_context()
+    first = _Task(0.0, 0, iter(()), None, ctx)
+    same = first.with_priority(0.0)
+    higher = _Task(1.0, 1, iter(()), None, ctx)
+
+    assert first == same
+    assert (first != same) is False
+    assert first != higher
+    assert higher < first
+
+
+def test_add_job_rejects_coroutine_object() -> None:
+    async def native() -> None:
+        """Never awaited; only used to build a coroutine object."""
+
+    coro_obj = native()
+    try:
+        with pytest.raises(ValueError, match="not coroutine objects"):
+            FakeEventLoop().add_job(coro_obj)
+    finally:
+        coro_obj.close()
+
+
+def test_flush_tasks_detects_deadlock() -> None:
+    def forever() -> Generator[None, None, None]:
+        while True:
+            yield
+
+    loop = FakeEventLoop()
+    loop.add_job(forever)
+
+    with pytest.raises(RuntimeError, match="Circular dependency"):
+        loop.flush_tasks()

tests/unit_tests/test_cpp_helpers.py

@@ -1,5 +1,6 @@
 from collections.abc import Awaitable, Callable
 import importlib.util
+import inspect
 import logging
 from pathlib import Path
 import sys
@@ -362,3 +363,43 @@ def test_add_optional_skips_none(monkeypatch: pytest.MonkeyPatch) -> None:
     assert ch.add_optional(setter, None) is None
     setter.assert_not_called()
     add_mock.assert_not_called()
+
+
+def _register_component_mocks(monkeypatch: pytest.MonkeyPatch) -> tuple[Mock, Mock]:
+    base = Mock()
+    base.__str__ = lambda self: "foo.bar"
+    base.type = Mock()
+    base.type.__str__ = lambda self: "foo::Bar"
+    var = Mock(base=base)
+    app_mock = Mock(register_component_=Mock(return_value=var))
+    monkeypatch.setattr(ch, "App", app_mock)
+    monkeypatch.setattr(ch, "CORE", Mock(component_ids=["foo.bar"], data={}))
+    monkeypatch.setattr(ch, "add", Mock())
+    return var, app_mock
+
+
+@pytest.mark.asyncio
+async def test_register_component_stack_error_warns_and_skips_source(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    var, app_mock = _register_component_mocks(monkeypatch)
+    monkeypatch.setattr(inspect, "stack", Mock(side_effect=IndexError("no stack")))
+
+    with caplog.at_level(logging.WARNING):
+        assert await ch.register_component(var, {}) is var
+
+    assert "Error while finding name of component" in caplog.text
+    app_mock.register_component_.assert_called_once_with(var)
+
+
+@pytest.mark.asyncio
+async def test_register_component_without_module_has_no_source(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    var, app_mock = _register_component_mocks(monkeypatch)
+    monkeypatch.setattr(inspect, "stack", Mock(return_value=[(None,), (None,)]))
+    monkeypatch.setattr(inspect, "getmodule", lambda _: None)
+
+    assert await ch.register_component(var, {}) is var
+
+    app_mock.register_component_.assert_called_once_with(var)

tests/unit_tests/test_espidf_component.py

@@ -367,6 +367,56 @@ def test_generate_cmakelists_txt_references_project_managed_components_variable(
     assert "${ESPHOME_PROJECT_MANAGED_COMPONENTS}" in content
 
 
+def test_generate_cmakelists_txt_uses_declared_src_dir(
+    tmp_component: IDFComponent,
+) -> None:
+    """A library.json srcDir is used as is instead of probing src/Src/."""
+    lib_dir = tmp_component.path / "lib"
+    lib_dir.mkdir()
+    (lib_dir / "main.c").write_text("int main() {}")
+    tmp_component.data = {"build": {"srcDir": "lib"}}
+
+    content = generate_cmakelists_txt(tmp_component)
+
+    assert "SRCS" in content
+    assert "main.c" in content
+    assert "INCLUDE_DIRS" in content
+
+
+def test_generate_cmakelists_txt_without_include_dirs(
+    tmp_component: IDFComponent,
+) -> None:
+    """No INCLUDE_DIRS line when none of the include directories exist."""
+    tmp_component.data = {"build": {"srcDir": "missing"}}
+
+    content = generate_cmakelists_txt(tmp_component)
+
+    assert "INCLUDE_DIRS" not in content
+
+
+def test_generate_idf_component_yml_with_several_dependencies(
+    tmp_component: IDFComponent, tmp_path: Path
+) -> None:
+    """Every dependency lands under one shared dependencies section."""
+    deps = []
+    for name in ("dep_a", "dep_b"):
+        dep = IDFComponent(name, "1.0", source=URLSource("http://dummy.com"))
+        dep.path = tmp_path / name
+        deps.append(dep)
+    tmp_component.dependencies = deps
+    tmp_component.data = {}
+
+    result = generate_idf_component_yml(tmp_component)
+
+    assert result == (
+        "dependencies:\n"
+        "  dep_a:\n"
+        f"    override_path: {deps[0].path}\n"
+        "  dep_b:\n"
+        f"    override_path: {deps[1].path}\n"
+    )
+
+
 def test_generate_idf_component_yml_basic(tmp_component):
     tmp_component.data = {"description": "test", "repository": {"url": "http://aaa"}}
     result = generate_idf_component_yml(tmp_component)

tests/unit_tests/test_espota2.py

@@ -2055,3 +2055,28 @@ def test_perform_ota_tcp_send_network_error_unchanged(mock_socket: Mock) -> None
     ]
     with pytest.raises(espota2.OTANetworkError, match="sending frame"):
         espota2.perform_ota(mock_socket, None, io.BytesIO(b"x" * 100), "test.bin")
+
+
+@pytest.mark.parametrize(
+    ("dashboard", "expect_device_hint"), [(False, True), (True, False)]
+)
+def test_run_ota_impl_resolve_failure_hint(
+    mock_resolve_ip: Mock,
+    firmware_file: Path,
+    caplog: pytest.LogCaptureFixture,
+    monkeypatch: pytest.MonkeyPatch,
+    dashboard: bool,
+    expect_device_hint: bool,
+) -> None:
+    """The --device hint is only useful on the command line, not the dashboard."""
+    from esphome.core import CORE
+
+    monkeypatch.setattr(CORE, "dashboard", dashboard)
+    mock_resolve_ip.side_effect = EsphomeError("no such host")
+
+    with caplog.at_level(logging.ERROR), pytest.raises(espota2.OTAError):
+        espota2.run_ota_impl_("test.local", 3232, None, firmware_file)
+
+    assert "Is it connected to WiFi?" in caplog.text
+    assert ("try --device <IP>" in caplog.text) is expect_device_hint
+    assert "static IP address" in caplog.text

tests/unit_tests/test_final_validate.py

@@ -2,7 +2,7 @@
 
 import pytest
 
-from esphome import final_validate as fv
+from esphome import config_validation as cv, final_validate as fv
 from esphome.config import Config
 from esphome.const import CONF_ID
 from esphome.core import ID
@@ -45,3 +45,30 @@ def test_non_dict_parent_raises_type_error() -> None:
     config.declare_ids.append((list_id, ["listed", 0]))
     with pytest.raises(TypeError):
         fv.get_declaring_config(ID("in_list"), config)
+
+
+@pytest.mark.parametrize("pre_built", [False, True])
+def test_id_declaration_match_schema_validates_declaration(pre_built: bool) -> None:
+    config, hub_id, _ = _config()
+    schema = {cv.Required("mode"): cv.string}
+    if pre_built:
+        schema = cv.Schema(schema, extra=cv.ALLOW_EXTRA)
+    validator = fv.id_declaration_match_schema(schema)
+    token = fv.full_config.set(config)
+    try:
+        result = validator(ID("my_hub"))
+    finally:
+        fv.full_config.reset(token)
+
+    assert result["mode"] == "fast"
+
+
+def test_id_declaration_match_schema_rejects_mismatch() -> None:
+    config, _, _ = _config()
+    validator = fv.id_declaration_match_schema({cv.Required("mode"): cv.one_of("slow")})
+    token = fv.full_config.set(config)
+    try:
+        with pytest.raises(cv.Invalid):
+            validator(ID("my_hub"))
+    finally:
+        fv.full_config.reset(token)

tests/unit_tests/test_git.py

@@ -3233,3 +3233,72 @@ def test_resolve_symlink_stub_returns_none_when_read_bytes_raises(
         result = git.resolve_symlink_stub(repo_dir, stub)
 
     assert result is None
+
+
+def test_git_file_git_url() -> None:
+    gf = git.GitFile("github.com", "esphome", "repo", "file.yaml")
+
+    assert gf.git_url == "https://github.com/esphome/repo.git"
+
+
+@pytest.mark.parametrize(
+    ("domain", "expected"),
+    [
+        (
+            "codeberg.org",
+            "https://codeberg.org/owner/repo/raw/commit/main/dir/file.yaml",
+        ),
+        (
+            "github.com",
+            "https://raw.githubusercontent.com/owner/repo/main/dir/file.yaml",
+        ),
+        ("gitlab.com", "https://gitlab.com/owner/repo/-/raw/main/dir/file.yaml"),
+    ],
+)
+def test_git_file_raw_url(domain: str, expected: str) -> None:
+    gf = git.GitFile(domain, "owner", "repo", "dir/file.yaml", ref="main")
+
+    assert gf.raw_url == expected
+
+
+def test_git_file_raw_url_requires_ref() -> None:
+    gf = git.GitFile("github.com", "owner", "repo", "file.yaml")
+
+    with pytest.raises(ValueError, match="URL has no ref"):
+        _ = gf.raw_url
+
+
+def test_git_file_raw_url_unsupported_domain() -> None:
+    gf = git.GitFile("example.com", "owner", "repo", "file.yaml", ref="main")
+
+    with pytest.raises(NotImplementedError, match="example.com not supported"):
+        _ = gf.raw_url
+
+
+def test_git_file_from_shorthand_full() -> None:
+    gf = git.GitFile.from_shorthand("gitlab://owner/repo/sub/file.yaml@v1.0?query")
+
+    assert gf == git.GitFile(
+        "gitlab.com", "owner", "repo", "sub/file.yaml", ref="v1.0", query="query"
+    )
+
+
+def test_git_file_from_shorthand_without_ref_or_query() -> None:
+    gf = git.GitFile.from_shorthand("codeberg://owner/repo/file.yaml")
+
+    assert gf == git.GitFile("codeberg.org", "owner", "repo", "file.yaml")
+
+
+def test_git_file_from_shorthand_requires_string() -> None:
+    with pytest.raises(ValueError, match="must be a string"):
+        git.GitFile.from_shorthand(123)
+
+
+def test_git_file_from_shorthand_bad_format() -> None:
+    with pytest.raises(ValueError, match="not in expected"):
+        git.GitFile.from_shorthand("not a url")
+
+
+def test_git_file_from_shorthand_unknown_domain() -> None:
+    with pytest.raises(ValueError, match="Unknown git domain bitbucket"):
+        git.GitFile.from_shorthand("bitbucket://owner/repo/file.yaml")

tests/unit_tests/test_log.py

@@ -11,7 +11,7 @@ import time
 import pytest
 
 from esphome.core import CORE
-from esphome.log import AnsiFore, AnsiStyle, color, setup_log
+from esphome.log import AnsiFore, AnsiStyle, ESPHomeLogFormatter, color, setup_log
 
 
 class _FakeTty(io.StringIO):
@@ -339,3 +339,87 @@ def test_setup_log_win32_always_imports_colorama(
     monkeypatch.setattr(sys, "stderr", _FakeTty())
     setup_log()
     assert "colorama" in sys.modules
+
+
+def _record(level: int, message: str) -> logging.LogRecord:
+    return logging.LogRecord("esphome", level, __file__, 1, message, None, None)
+
+
+@pytest.mark.parametrize(
+    ("level", "colour"),
+    [
+        (logging.DEBUG, AnsiFore.CYAN),
+        (logging.INFO, AnsiFore.GREEN),
+        (logging.WARNING, AnsiFore.YELLOW),
+        (logging.ERROR, AnsiFore.RED),
+        (logging.CRITICAL, AnsiFore.RED),
+    ],
+)
+def test_formatter_colours_by_level(
+    monkeypatch: pytest.MonkeyPatch, level: int, colour: AnsiFore
+) -> None:
+    monkeypatch.setattr(CORE, "dashboard", False)
+    formatter = ESPHomeLogFormatter(include_timestamp=False)
+
+    result = formatter.format(_record(level, "hello"))
+
+    name = logging.getLevelName(level)
+    assert result == f"{colour.value}{name} hello{AnsiStyle.RESET_ALL.value}"
+
+
+def test_formatter_unknown_level_has_no_colour_prefix(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(CORE, "dashboard", False)
+    formatter = ESPHomeLogFormatter(include_timestamp=False)
+
+    result = formatter.format(_record(25, "hello"))
+
+    assert result == f"Level 25 hello{AnsiStyle.RESET_ALL.value}"
+
+
+def test_formatter_includes_timestamp_when_requested(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(CORE, "dashboard", False)
+    formatter = ESPHomeLogFormatter(include_timestamp=True)
+    record = _record(logging.INFO, "hello")
+
+    result = formatter.format(record)
+
+    assert formatter.formatTime(record) in result
+
+
+def test_formatter_dashboard_escapes_ansi(monkeypatch: pytest.MonkeyPatch) -> None:
+    """The dashboard prints the escape character as literal text."""
+    monkeypatch.setattr(CORE, "dashboard", True)
+    formatter = ESPHomeLogFormatter(include_timestamp=False)
+
+    result = formatter.format(_record(logging.INFO, "hello"))
+
+    assert "\033" not in result
+    assert result.startswith("\\033[32mINFO hello")
+
+
+def test_setup_log_debug_sets_verbose(
+    monkeypatch: pytest.MonkeyPatch, colorama_probe: None
+) -> None:
+    monkeypatch.setattr(CORE, "dashboard", True)
+    logging.getLogger().handlers.clear()
+
+    setup_log(logging.DEBUG)
+
+    assert CORE.verbose is True
+    assert CORE.quiet is False
+
+
+def test_setup_log_critical_sets_quiet(
+    monkeypatch: pytest.MonkeyPatch, colorama_probe: None
+) -> None:
+    monkeypatch.setattr(CORE, "dashboard", True)
+    logging.getLogger().handlers.clear()
+
+    setup_log(logging.CRITICAL)
+
+    assert CORE.quiet is True
+    assert CORE.verbose is False

tests/unit_tests/test_pins.py

@@ -1,13 +1,26 @@
 """Unit tests for esphome/pins.py."""
 
+from collections.abc import Iterator
+import logging
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
 from esphome import config_validation as cv, pins
 import esphome.codegen as cg
-from esphome.const import CONF_ADDRESS
-from esphome.core import ID
+from esphome.config import Config, path_context
+from esphome.const import (
+    CONF_ADDRESS,
+    CONF_ALLOW_OTHER_USES,
+    CONF_ID,
+    CONF_IGNORE_STRAPPING_WARNING,
+    CONF_INVERTED,
+    CONF_MODE,
+    CONF_NUMBER,
+    KEY_CORE,
+    KEY_TARGET_PLATFORM,
+)
+from esphome.core import CORE, ID
 from esphome.schema_extractors import SCHEMA_EXTRACT
 
 SomeHub = cg.esphome_ns.class_("SomeHub")
@@ -166,3 +179,181 @@ async def test_new_parented_pin(args: tuple[str, ...]) -> None:
     var.set_inverted.assert_called_once_with(True)
     var.set_flags.assert_called_once()
     assert add.call_count == 3
+
+
+@pytest.fixture
+def registry() -> pins.PinRegistry:
+    reg = pins.PinRegistry()
+    reg.register("hub", lambda conf: conf)(lambda conf: conf)
+    return reg
+
+
+@pytest.fixture
+def pin_path() -> Iterator[None]:
+    CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: "esp32"}
+    token = path_context.set(["pin"])
+    try:
+        yield
+    finally:
+        path_context.reset(token)
+
+
+@pytest.mark.usefixtures("pin_path")
+def test_registry_validate_tracks_usage_and_get_count(
+    registry: pins.PinRegistry,
+) -> None:
+    assert registry.get_count("hub", "h1", 4) == 0
+
+    registry.validate({"hub": "h1", CONF_NUMBER: 4}, "hub")
+
+    assert registry.get_count("hub", "h1", 4) == 1
+    assert registry.get_count("hub", "h1", 5) == 0
+
+
+@pytest.mark.usefixtures("pin_path")
+def test_registry_validate_without_number_is_not_tracked(
+    registry: pins.PinRegistry,
+) -> None:
+    assert registry.validate({"hub": "h1"}, "hub") == {"hub": "h1"}
+    assert not registry.pins_used
+
+
+def test_registry_get_to_code(registry: pins.PinRegistry) -> None:
+    to_code = registry["hub"][0]
+
+    assert registry.get_to_code("hub") is to_code
+    assert registry.get_to_code({"hub": "h1"}) is to_code
+
+
+def _final_config(client_id: ID) -> Config:
+    config = Config()
+    config["hub"] = [{CONF_ID: client_id}]
+    config.declare_ids.append((client_id, ["hub", 0, CONF_ID]))
+    return config
+
+
+@pytest.mark.parametrize(
+    ("uses", "allow_others", "testing_mode", "error"),
+    [
+        (2, False, False, "used in multiple places"),
+        (2, True, False, None),
+        (2, False, True, None),
+        (1, False, False, None),
+        (1, True, False, "incorrectly sets allow_other_uses"),
+        (1, True, True, None),
+    ],
+)
+@pytest.mark.usefixtures("pin_path")
+def test_registry_final_validate_reuse_rules(
+    registry: pins.PinRegistry,
+    monkeypatch: pytest.MonkeyPatch,
+    uses: int,
+    allow_others: bool,
+    testing_mode: bool,
+    error: str | None,
+) -> None:
+    client_id = ID("h1", is_declaration=True)
+    extra = {CONF_ALLOW_OTHER_USES: True} if allow_others else {}
+    for _ in range(uses):
+        registry.validate({"hub": client_id, CONF_NUMBER: 4, **extra}, "hub")
+    monkeypatch.setattr(CORE, "testing_mode", testing_mode)
+    config = _final_config(client_id)
+
+    registry.final_validate(config)
+
+    if error is None:
+        assert not config.errors
+    else:
+        assert all(error in str(err) for err in config.errors)
+        assert config.errors
+
+
+@pytest.mark.usefixtures("pin_path")
+def test_registry_final_validate_runs_hook_with_parent_config(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    hook = MagicMock()
+    registry = pins.PinRegistry()
+    registry.register("hub", lambda conf: conf, hook)(lambda conf: conf)
+    client_id = ID("h1", is_declaration=True)
+    pin = registry.validate({"hub": client_id, CONF_NUMBER: 4}, "hub")
+    config = _final_config(client_id)
+
+    registry.final_validate(config)
+
+    hook.assert_called_once_with(pin, config["hub"][0])
+    assert not config.errors
+
+
+@pytest.mark.parametrize(
+    ("mode", "expected"),
+    [
+        ("input", {"input": True}),
+        ("INPUT_PULLUP", {"input": True, "pullup": True}),
+        ("output_open_drain", {"output": True, "open_drain": True}),
+    ],
+)
+def test_set_mode_parses_string_modes(mode: str, expected: dict[str, bool]) -> None:
+    assert pins._set_mode({CONF_NUMBER: 1, CONF_MODE: mode}, {}) == {
+        CONF_NUMBER: 1,
+        CONF_MODE: expected,
+    }
+
+
+def test_set_mode_rejects_unknown_string_mode() -> None:
+    with pytest.raises(cv.Invalid, match="Unknown pin mode bogus"):
+        pins._set_mode({CONF_MODE: "bogus"}, {})
+
+
+def test_set_mode_keeps_dict_mode_and_applies_default() -> None:
+    explicit = {CONF_MODE: {"input": True}}
+    assert pins._set_mode(explicit, {"output": True}) is explicit
+    assert pins._set_mode({}, {"output": True}) == {CONF_MODE: {"output": True}}
+
+
+def test_gpio_flags_expr() -> None:
+    assert str(pins.gpio_flags_expr({"input": False})) == str(cg.gpio_Flags.FLAG_NONE)
+    flags = str(pins.gpio_flags_expr({"input": True, "pullup": True}))
+    assert "FLAG_INPUT" in flags
+    assert "FLAG_PULLUP" in flags
+
+
+def test_gpio_validate_modes() -> None:
+    mode = {"input": True, "output": False}
+    assert pins.gpio_validate_modes(mode) is mode
+    with pytest.raises(cv.Invalid, match="Mode must be input or output"):
+        pins.gpio_validate_modes({"input": False, "output": False})
+
+
+INPUT_MODE = {"input": True}
+
+
+def test_gpio_base_schema_rejects_placeholder_number() -> None:
+    schema = pins.gpio_base_schema(SomeHub, cv.int_)
+
+    with pytest.raises(cv.Invalid, match="placeholder 'GPIOX'"):
+        schema({CONF_NUMBER: "GPIOX", CONF_MODE: INPUT_MODE})
+    assert schema({CONF_NUMBER: 3, CONF_MODE: INPUT_MODE})[CONF_NUMBER] == 3
+
+
+@pytest.mark.parametrize("invertible", [True, False])
+def test_gpio_base_schema_invertible(invertible: bool) -> None:
+    schema = pins.gpio_base_schema(SomeHub, cv.int_, invertible=invertible)
+
+    assert (
+        CONF_INVERTED in schema({CONF_NUMBER: 3, CONF_MODE: INPUT_MODE})
+    ) is invertible
+
+
+def test_check_strapping_pin(caplog: pytest.LogCaptureFixture) -> None:
+    logger = logging.getLogger("test_pins")
+    with caplog.at_level(logging.WARNING, logger="test_pins"):
+        pins.check_strapping_pin({CONF_NUMBER: 2}, {2}, logger)
+        pins.check_strapping_pin(
+            {CONF_NUMBER: 2, CONF_IGNORE_STRAPPING_WARNING: True}, {2}, logger
+        )
+    assert caplog.text.count("GPIO2 is a strapping pin") == 1
+    with pytest.raises(cv.Invalid, match="not a strapping pin"):
+        pins.check_strapping_pin(
+            {CONF_NUMBER: 3, CONF_IGNORE_STRAPPING_WARNING: True}, {2}, logger
+        )

tests/unit_tests/test_platformio_runner.py

@@ -5,10 +5,12 @@ from __future__ import annotations
 from collections.abc import Callable
 import io
 from pathlib import Path
+import runpy
 import sys
-from types import ModuleType
+from types import ModuleType, SimpleNamespace
 from unittest.mock import Mock
 
+import platformio
 from platformio.registry.client import RegistryClient
 import pytest
 
@@ -146,3 +148,73 @@ def test_patch_registry_private_packages_skips_account_probe(
 
     assert RegistryClient.allowed_private_packages() is False
     assert RegistryClient().allowed_private_packages() is False
+
+
+def test_patch_file_downloader_retries_when_session_cleanup_fails(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A failing close of the old session must not stop the retry."""
+    package_exception = type("PackageException", (Exception,), {})
+    attempts: list[int] = []
+
+    def flaky_init(self: object, *args: object, **kwargs: object) -> None:
+        attempts.append(1)
+        if len(attempts) < 2:
+            raise package_exception("502")
+
+    downloader = type("FileDownloader", (), {"__init__": flaky_init})
+    monkeypatch.setitem(
+        sys.modules,
+        "platformio.package.download",
+        SimpleNamespace(FileDownloader=downloader),
+    )
+    monkeypatch.setitem(
+        sys.modules,
+        "platformio.package.exception",
+        SimpleNamespace(PackageException=package_exception),
+    )
+    monkeypatch.setattr(runner.time, "sleep", lambda _delay: None)
+
+    runner.patch_file_downloader()
+
+    instance = object.__new__(downloader)
+    instance._http_response = Mock(close=Mock(side_effect=OSError("boom")))
+    downloader.__init__(instance)
+
+    assert len(attempts) == 2
+    instance._http_response.close.assert_called_once()
+
+
+def test_runner_runs_main_when_executed_as_a_script(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Running the file as ``__main__`` exits with PlatformIO's return code."""
+    monkeypatch.setattr(sys, "argv", ["runner", "run"])
+    monkeypatch.setattr(sys, "stdout", io.StringIO())
+    monkeypatch.setattr(sys, "stderr", io.StringIO())
+    # The script re-imports and applies the PlatformIO patches on its own
+    # module namespace, so give it stand-ins rather than the real internals.
+    monkeypatch.setitem(sys.modules, "platformio.run", Mock())
+    monkeypatch.setitem(
+        sys.modules,
+        "platformio.package.download",
+        SimpleNamespace(FileDownloader=type("FileDownloader", (), {})),
+    )
+    monkeypatch.setitem(
+        sys.modules,
+        "platformio.package.exception",
+        SimpleNamespace(PackageException=Exception),
+    )
+    monkeypatch.setitem(
+        sys.modules,
+        "platformio.registry.client",
+        SimpleNamespace(RegistryClient=type("RegistryClient", (), {})),
+    )
+    pio_main = SimpleNamespace(main=lambda: 3)
+    monkeypatch.setitem(sys.modules, "platformio.__main__", pio_main)
+    monkeypatch.setattr(platformio, "__main__", pio_main, raising=False)
+
+    with pytest.raises(SystemExit) as exc_info:
+        runpy.run_path(str(Path(runner.__file__)), run_name="__main__")
+
+    assert exc_info.value.code == 3

tests/unit_tests/test_platformio_toolchain.py

@@ -112,6 +112,22 @@ def test_idedata_cc_path(setup_core: Path) -> None:
     )
 
 
+@pytest.mark.parametrize(
+    ("raw_data", "expected"),
+    [
+        pytest.param(
+            {"defines": ["USE_ESP32", "FOO=1"]}, ["USE_ESP32", "FOO=1"], id="set"
+        ),
+        pytest.param({}, [], id="missing"),
+    ],
+)
+def test_idedata_defines(raw_data: dict[str, list[str]], expected: list[str]) -> None:
+    """IDEData.defines returns the defines, or an empty list when absent."""
+    idedata = toolchain.IDEData({"prog_path": "/path/to/firmware.elf", **raw_data})
+
+    assert idedata.defines == expected
+
+
 def test_flash_image_dataclass() -> None:
     """Test FlashImage dataclass stores path and offset correctly."""
     image = FlashImage(path=Path("/path/to/image.bin"), offset="0x10000")
@@ -758,6 +774,17 @@ def test_copy_ccache_script(setup_core: Path) -> None:
     assert dest.read_text() == source.read_text()
 
 
+def test_copy_pch_script(setup_core: Path) -> None:
+    """The shared PCH pre-script is copied into the build dir."""
+    CORE.build_path = setup_core / "build" / "test"
+
+    toolchain.copy_pch_script()
+
+    dest = setup_core / "build" / "test" / "pch.py"
+    source = Path(toolchain.__file__).parent / "pch.py.script"
+    assert dest.read_text() == source.read_text()
+
+
 class _FakeSConsEnv(dict):
     """Just enough of a SCons construction environment for ccache.py."""
 
@@ -1015,6 +1042,20 @@ def test_run_platformio_cli_run_builds_command(
     )
 
 
+def test_run_platformio_cli_run_without_verbose(
+    setup_core: Path, mock_run_platformio_cli: Mock
+) -> None:
+    """Without verbose, no -v flag is passed to PlatformIO."""
+    CORE.build_path = str(setup_core / "build" / "test")
+    mock_run_platformio_cli.return_value = 0
+
+    toolchain.run_platformio_cli_run({"name": "test"}, False, "extra")
+
+    mock_run_platformio_cli.assert_called_once_with(
+        "run", "-d", CORE.build_path, "extra"
+    )
+
+
 def test_run_compile(setup_core: Path, mock_run_platformio_cli_run: Mock) -> None:
     """Test run_compile with process limit."""
     from esphome.const import CONF_COMPILE_PROCESS_LIMIT, CONF_ESPHOME

tests/unit_tests/test_resolver.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import importlib
+import logging
 import re
 import socket
 from unittest.mock import ANY, patch
@@ -10,6 +12,7 @@ from aioesphomeapi.core import ResolveAPIError, ResolveTimeoutAPIError
 from aioesphomeapi.host_resolver import AddrInfo, IPv4Sockaddr, IPv6Sockaddr
 import pytest
 
+from esphome import resolver
 from esphome.async_thread import AsyncDispatchTimeout
 from esphome.core import EsphomeError
 from esphome.resolver import RESOLVE_TIMEOUT, AsyncResolver
@@ -162,3 +165,18 @@ def test_async_resolver_mixed_addresses(
         mock_resolve.assert_called_once_with(
             ["test.local", "192.168.1.100", "::1"], 6053, timeout=RESOLVE_TIMEOUT
         )
+
+
+def test_invalid_resolve_timeout_env_falls_back_to_default(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A non-numeric ESPHOME_RESOLVE_TIMEOUT warns and uses the default."""
+    monkeypatch.setenv("ESPHOME_RESOLVE_TIMEOUT", "soon")
+    try:
+        with caplog.at_level(logging.WARNING, logger="esphome.resolver"):
+            importlib.reload(resolver)
+        assert resolver.RESOLVE_TIMEOUT == 20.0
+        assert "ESPHOME_RESOLVE_TIMEOUT='soon' is not a valid number" in caplog.text
+    finally:
+        monkeypatch.delenv("ESPHOME_RESOLVE_TIMEOUT")
+        importlib.reload(resolver)

tests/unit_tests/test_schema_extractors.py

@@ -0,0 +1,90 @@
+"""Tests for esphome.schema_extractors."""
+
+from collections.abc import Callable, Iterator
+from typing import Any
+
+import pytest
+
+from esphome import schema_extractors as se
+
+
+@pytest.fixture
+def extraction_enabled() -> Iterator[None]:
+    """Enable schema extraction and restore the module state afterwards."""
+    stores = (
+        se.extended_schemas,
+        se.list_schemas,
+        se.registry_schemas,
+        se.hidden_schemas,
+        se.typed_schemas,
+    )
+    saved = [dict(store) for store in stores]
+    previous = se.EnableSchemaExtraction
+    se.EnableSchemaExtraction = True
+    try:
+        yield
+    finally:
+        se.EnableSchemaExtraction = previous
+        for store, old in zip(stores, saved, strict=True):
+            store.clear()
+            store.update(old)
+
+
+def _validator(value: Any) -> Any:
+    return value
+
+
+def test_decorators_are_passthrough_when_disabled(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(se, "EnableSchemaExtraction", False)
+    assert se.schema_extractor("name")(_validator) is _validator
+    assert se.schema_extractor_registry(object())(_validator) is _validator
+    assert se.schema_extractor_extended(_validator) is _validator
+    assert se.schema_extractor_list(_validator) is _validator
+    assert se.schema_extractor_typed(_validator) is _validator
+
+
+@pytest.mark.usefixtures("extraction_enabled")
+def test_schema_extractor_records_hidden_schema() -> None:
+    assert se.schema_extractor("my_validator")(_validator) is _validator
+    assert se.hidden_schemas[repr(_validator)] == "my_validator"
+
+
+@pytest.mark.usefixtures("extraction_enabled")
+def test_schema_extractor_registry_records_registry() -> None:
+    registry = object()
+    assert se.schema_extractor_registry(registry)(_validator) is _validator
+    assert se.registry_schemas[repr(_validator)] is registry
+
+
+@pytest.mark.parametrize(
+    ("decorator", "store_name"),
+    [
+        (se.schema_extractor_extended, "extended_schemas"),
+        (se.schema_extractor_list, "list_schemas"),
+    ],
+)
+@pytest.mark.usefixtures("extraction_enabled")
+def test_args_recording_decorators(
+    decorator: Callable[[Callable[..., Any]], Callable[..., Any]], store_name: str
+) -> None:
+    def build(*args: Any) -> list[Any]:
+        return list(args)
+
+    wrapped = decorator(build)
+    result = wrapped(1, 2)
+
+    assert result == [1, 2]
+    assert getattr(se, store_name)[repr(result)] == (1, 2)
+
+
+@pytest.mark.usefixtures("extraction_enabled")
+def test_schema_extractor_typed_records_args_and_kwargs() -> None:
+    def build(*args: Any, **kwargs: Any) -> list[Any]:
+        return [args, kwargs]
+
+    result = se.schema_extractor_typed(build)(1, key="x")
+
+    assert result == [(1,), {"key": "x"}]
+    assert se.typed_schemas[repr(result)] == ((1,), {"key": "x"})

tests/unit_tests/test_size_summary.py

@@ -9,7 +9,11 @@ from unittest.mock import patch
 
 import pytest
 
-from esphome.espidf.size_summary import print_summary
+from esphome.espidf.size_summary import (
+    _find_app_partition_size,
+    _parse_size,
+    print_summary,
+)
 
 
 def _write_size_json(tmp_path: Path, data: dict) -> Path:
@@ -317,3 +321,51 @@ def test_print_summary_skips_flash_on_bad_input(
         for r in caplog.records
     )
     assert warned == with_partitions
+
+
+@pytest.mark.parametrize(
+    ("token", "expected"),
+    [
+        pytest.param("", 0, id="empty"),
+        pytest.param("  ", 0, id="blank"),
+        pytest.param("0x10000", 0x10000, id="hex"),
+        pytest.param("0X10", 0x10, id="hex_upper"),
+        pytest.param("1M", 1024 * 1024, id="mebibyte"),
+        pytest.param("512k", 512 * 1024, id="kibibyte_lower"),
+        pytest.param("4096", 4096, id="decimal"),
+    ],
+)
+def test_parse_size(token: str, expected: int) -> None:
+    """Partition sizes accept hex, K/M suffixes, decimals and blanks."""
+    assert _parse_size(token) == expected
+
+
+def test_find_app_partition_size_skips_unrelated_rows(tmp_path: Path) -> None:
+    """Comments, blank lines, short rows and non-app rows are passed over."""
+    partitions = tmp_path / "partitions.csv"
+    partitions.write_text(
+        "# name, type, subtype, offset, size, flags\n"
+        "\n"
+        "short, app, factory\n"
+        "nvs, data, nvs, 0x9000, 0x6000,\n"
+        "app0, app, ota_0, 0x10000, 1M,\n"
+    )
+    assert _find_app_partition_size(partitions) == 1024 * 1024
+
+
+def test_find_app_partition_size_without_app_partition_raises(tmp_path: Path) -> None:
+    """A table with no app+factory or app+ota_0 row is rejected."""
+    partitions = tmp_path / "partitions.csv"
+    partitions.write_text("nvs, data, nvs, 0x9000, 0x6000,\n")
+    with pytest.raises(ValueError, match="No app\\+factory or app\\+ota_0"):
+        _find_app_partition_size(partitions)
+
+
+def test_print_summary_skips_unparseable_json(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    """A truncated size json prints nothing instead of raising."""
+    size_json = tmp_path / "esp_idf_size.json"
+    size_json.write_text("{not json")
+    print_summary(size_json, _write_partitions(tmp_path), tmp_path / "firmware.elf")
+    assert capsys.readouterr().out == ""

tests/unit_tests/test_util.py

@@ -1299,7 +1299,6 @@ def test_ordered_dict_repr_is_plain_dict() -> None:
 @pytest.mark.parametrize("platform", ["linux", "darwin"])
 def test_get_serial_ports(platform: str, monkeypatch: pytest.MonkeyPatch) -> None:
     """Ports are filtered by VID:PID; by-id platform devices are added on Linux."""
-    monkeypatch.setattr(sys, "platform", platform)
     comports = [
         ("", "empty", "VID:PID=1:2"),
         ("/dev/ttyUSB1", "usb", "VID:PID=1:2"),
@@ -1320,6 +1319,9 @@ def test_get_serial_ports(platform: str, monkeypatch: pytest.MonkeyPatch) -> Non
         patch("serial.tools.list_ports_linux.SysFS", side_effect=make_device),
         patch.object(util, "Path", return_value=fake_path),
     ):
+        # Only after patch() has imported pyserial: it picks its backend from
+        # sys.platform at import time, and the macOS one fails on other hosts
+        monkeypatch.setattr(sys, "platform", platform)
         ports = util.get_serial_ports()
 
     paths = [p.path for p in ports]

tests/unit_tests/test_vscode.py

@@ -7,7 +7,7 @@ import esphome.config_validation as cv
 from esphome.core import EsphomeError
 
 
-def _run_repl_test(input_data):
+def _run_repl_test(input_data, *, ace=False):
     """Reusable test function for different input scenarios."""
     input_data.append(_exit())
     with (
@@ -15,7 +15,8 @@ def _run_repl_test(input_data):
         patch("sys.stdout") as mock_stdout,
     ):
         args = Mock([])
-        args.ace = False
+        args.ace = ace
+        args.configuration = "config_dir"
         args.substitution = None
         vscode.read_config(args)
 
@@ -264,3 +265,23 @@ logger:
     (validation_error,) = error["validation_errors"]
     assert validation_error["range"] is None
     assert "extra keys not allowed" in validation_error["message"]
+
+
+def test_ace_mode_loads_config_relative_to_configuration_dir():
+    output_lines = _run_repl_test(
+        [
+            _validate("x.yaml"),
+            # read_file x.yaml
+            _file_response("""esphome:
+  name: test1
+esp8266:
+  board: d1_mini
+"""),
+        ],
+        ace=True,
+    )
+
+    assert output_lines == [
+        _read_file(str(Path("config_dir", "x.yaml"))),
+        RESULT_NO_ERROR,
+    ]

tests/unit_tests/test_wizard.py

@@ -1,5 +1,6 @@
 """Tests for the wizard.py file."""
 
+import importlib.util
 from pathlib import Path
 from typing import Any
 from unittest.mock import MagicMock, patch
@@ -716,3 +717,74 @@ def test_fallback_psk_uses_secrets_choice(
 
     assert 'password: "XXXXXXXXXXXX"' in config
     assert mock_choice.call_count == 12
+
+
+def test_wizard_file_rpipico_has_no_wifi_or_api() -> None:
+    """The rpipico has no WiFi, so the config stops after the logger."""
+    config = wz.wizard_file(name="pico", platform="RP2", board="rpipico")
+
+    assert "logger:" in config
+    assert "api:" not in config
+    assert "wifi:" not in config
+
+
+def test_wizard_file_secret_ssid_is_not_quoted(
+    default_config: dict[str, Any],
+) -> None:
+    default_config["ssid"] = "!secret wifi_ssid"
+    default_config["psk"] = "!secret wifi_password"
+
+    config = wz.wizard_file(**default_config)
+
+    assert "  ssid: !secret wifi_ssid\n" in config
+    assert "  password: !secret wifi_password\n" in config
+
+
+def test_wizard_file_without_ssid_comments_out_credentials(
+    default_config: dict[str, Any],
+) -> None:
+    del default_config["ssid"]
+    del default_config["psk"]
+
+    config = wz.wizard_file(**default_config)
+
+    assert '  # ssid: "My SSID"' in config
+    assert '  # password: "mypassword"' in config
+
+
+def test_wizard_write_rejects_unknown_board(
+    tmp_path: Path, default_config: dict[str, Any], monkeypatch: MonkeyPatch
+) -> None:
+    del default_config["platform"]
+    default_config["board"] = "not_a_real_board"
+    monkeypatch.setattr(CORE, "config_path", tmp_path.parent)
+
+    result = wz.wizard_write(tmp_path / "test.yaml", **default_config)
+
+    assert result is False
+    assert not (tmp_path / "test.yaml").exists()
+
+
+def test_wizard_returns_error_when_write_fails(
+    tmp_path: Path, monkeypatch: MonkeyPatch, wizard_answers: list[str]
+) -> None:
+    monkeypatch.setattr("builtins.input", MagicMock(side_effect=wizard_answers))
+    monkeypatch.setattr(wz, "safe_print", lambda t=None, end=None: 0)
+    monkeypatch.setattr(wz, "sleep", lambda _: 0)
+    monkeypatch.setattr(wz, "wizard_write", MagicMock(return_value=False))
+
+    assert wz.wizard(tmp_path / "test.yaml") == 1
+
+
+def test_quickwizard_env_replaces_sleep_with_noop(monkeypatch: MonkeyPatch) -> None:
+    """With ESPHOME_QUICKWIZARD set the module defines a no-op sleep."""
+    monkeypatch.setenv("ESPHOME_QUICKWIZARD", "1")
+    # Load a private copy so the real module is left untouched
+    spec = importlib.util.spec_from_file_location("_quick_wizard", wz.__file__)
+    assert spec is not None
+    assert spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+
+    assert module.sleep(10) is None
+    assert module.sleep.__module__ == "_quick_wizard"