Merge branch 'dev' into jesserockz-2026-446

tests/script/test_build_language_schema.py

@@ -3,11 +3,13 @@
 from __future__ import annotations
 
 import ast
+from collections.abc import Callable
 import importlib.util
 import json
 from pathlib import Path
 import subprocess
 import sys
+from typing import Any
 
 import pytest
 
@@ -205,6 +207,47 @@ def test_convert_keys_no_marker_for_non_sensitive_field() -> None:
     assert "sensitive_source" not in entry
 
 
+def _wildcard_validator(value: Any) -> Any:
+    return value
+
+
+def test_convert_keys_marker_wrapped_callable_key_normalizes() -> None:
+    converted: dict = {}
+    _bls.convert_keys(converted, {cv.Optional(_wildcard_validator): cv.string}, "/root")
+
+    config_vars = converted["schema"]["config_vars"]
+    assert set(config_vars) == {"string"}
+    assert config_vars["string"]["key"] == "Optional"
+    assert config_vars["string"]["key_type"] == "_wildcard_validator"
+
+
+def test_convert_keys_marker_wrapped_callable_beside_fixed_keys() -> None:
+    converted: dict = {}
+    _bls.convert_keys(
+        converted,
+        {cv.Required("id"): cv.string, cv.Optional(_wildcard_validator): cv.string},
+        "/root",
+    )
+
+    assert set(converted["schema"]["config_vars"]) == {"id", "string"}
+
+
+def test_convert_keys_bare_callable_dotted_qualname() -> None:
+    def make_validator() -> Callable[[Any], Any]:
+        def validator(value: Any) -> Any:
+            return value
+
+        return validator
+
+    converted: dict = {}
+    _bls.convert_keys(converted, {make_validator(): cv.string}, "/root")
+
+    assert converted["key"] == "String"
+    assert converted["key_type"].endswith("make_validator.<locals>.validator")
+    assert "at 0x" not in converted["key_type"]
+    assert set(converted["schema"]["config_vars"]) == {"string"}
+
+
 # ---------------------------------------------------------------------------
 # Regression tests for the lvgl schema dump.
 #

tests/script/test_ci_custom.py

@@ -0,0 +1,147 @@
+"""Unit tests for the ESP_LOG-needs-braces lint rule in script/ci-custom.py.
+
+The rule flags an if/else/for/while whose only body is an unbraced ESP_LOG*() call (which becomes an
+empty statement -- and a -Wempty-body warning -- once the log level compiles the macro out). These
+tests pin the comment/string/raw-string masker, the accepted control-statement shapes, and the
+NOLINT escape hatch at both placements a contributor would try.
+"""
+
+import importlib.util
+from pathlib import Path
+import sys
+
+SCRIPT_DIR = (Path(__file__).parent / ".." / ".." / "script").resolve()
+sys.path.insert(0, str(SCRIPT_DIR))
+_spec = importlib.util.spec_from_file_location("ci_custom", SCRIPT_DIR / "ci-custom.py")
+ci_custom = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(ci_custom)
+
+mask = ci_custom._mask_cpp_comments_strings
+
+
+def _lint(content: str) -> list:
+    return ci_custom.lint_esp_log_needs_braces("test.cpp", content)
+
+
+# --- masker ---
+
+
+def test_mask_preserves_length_newlines_and_real_parens() -> None:
+    src = 'foo("bar") + baz();\nqux();\n'
+    masked = mask(src)
+    assert len(masked) == len(src)
+    assert masked.count("\n") == src.count("\n")
+    assert masked.count("(") == src.count("(")  # real parens survive for balancing
+
+
+def test_mask_blanks_line_and_block_comments() -> None:
+    assert "ESP_LOGD" not in mask("a; // if (x) ESP_LOGD(t);\n")
+    assert "ESP_LOGD" not in mask("a; /* if (x) ESP_LOGD(t); */ b;\n")
+
+
+def test_mask_blanks_string_literals() -> None:
+    assert "if" not in mask('x = "if (y) ESP_LOGD";\n')
+
+
+def test_mask_handles_raw_string_without_desync() -> None:
+    # A raw string full of quotes/parens must be consumed as one unit; code after it stays intact.
+    src = 's.print(R"(<a href="x">)");\nreturn;\n'
+    masked = mask(src)
+    assert "href" not in masked
+    assert "return;" in masked  # not swallowed by a desynced string scan
+
+
+# --- rule: flags real violations ---
+
+
+def test_flags_unbraced_if_next_line() -> None:
+    assert _lint("if (x)\n  ESP_LOGD(t);\n")
+
+
+def test_flags_unbraced_same_line() -> None:
+    assert _lint("if (x) ESP_LOGW(t);\n")
+
+
+def test_flags_c_style_for() -> None:
+    assert _lint("for (int i = 0; i < n; i++)\n  ESP_LOGD(t, i);\n")
+
+
+def test_flags_range_for_and_else() -> None:
+    assert _lint("for (auto &x : v)\n  ESP_LOGCONFIG(t);\n")
+    assert _lint("else\n  ESP_LOGE(t);\n")
+
+
+def test_flags_for_header_with_nested_call() -> None:
+    assert _lint("for (auto it = v.begin(); it != v.end(); ++it)\n  ESP_LOGD(t);\n")
+
+
+def test_for_header_does_not_reach_into_a_later_statement() -> None:
+    # The 'for' header is bounded to its own statement, so it cannot swallow the loop body and latch
+    # onto a later ')'. Without that, the '#if' line below is reported as an unbraced body even though
+    # the '#' preprocessor check should skip it.
+    assert not _lint(
+        "for (int i = 0; i < n; i++)\n  arr[i] = 0;\n#if defined(USE_X)\n  ESP_LOGD(t);\n#endif\n"
+    )
+
+
+def test_violation_after_a_for_loop_is_reported_at_its_own_line() -> None:
+    errors = _lint(
+        "for (int i = 0; i < n; i++)\n  sum += a[i];\nif (verbose)\n  ESP_LOGD(t, sum);\n"
+    )
+    lines = [line for line, _col, _msg in errors]
+    assert lines == [3]  # the 'if', not the 'for' on line 1
+
+
+def test_flags_lowercase_esph_log_family() -> None:
+    # core/log.h defines esph_log_*() alongside ESP_LOG*(); both expand to nothing below their level.
+    assert _lint('if (x)\n  esph_log_config(t, "m");\n')
+    assert _lint('if (err != ESP_OK)\n  esph_log_e(t, "m");\n')
+
+
+def test_digit_separator_does_not_disable_the_rest_of_the_file() -> None:
+    # A "'" digit separator must not be read as a char-literal opener, which blanked everything after.
+    assert _lint("uint32_t x = 1'000;\nif (y)\n  ESP_LOGD(t);\n")
+
+
+def test_mask_still_blanks_real_char_literals() -> None:
+    assert "ESP_LOGD" not in mask("char c = '\"'; // if (x) ESP_LOGD(t);\n")
+    assert not _lint("char sep = ';';\nif (x) {\n  ESP_LOGD(t);\n}\n")
+
+
+def test_flags_multiline_log_body() -> None:
+    assert _lint('if (x)\n  ESP_LOGD(t, "%d %d",\n           a, b);\n')
+
+
+def test_raw_string_before_violation_still_caught() -> None:
+    # Regression for the masker desyncing on a raw string and disabling the check for the rest.
+    assert _lint('s.print(R"(<a href="x">)");\nif (y)\n  ESP_LOGD(t);\n')
+
+
+# --- rule: ignores non-violations ---
+
+
+def test_ignores_braced_body() -> None:
+    assert not _lint("if (x) {\n  ESP_LOGD(t);\n}\n")
+
+
+def test_ignores_commented_out_code() -> None:
+    assert not _lint("// if (x) ESP_LOGD(t);\n")
+
+
+def test_ignores_preprocessor_else() -> None:
+    assert not _lint("#else\n  ESP_LOGCONFIG(t);\n#endif\n")
+
+
+def test_ignores_non_log_body() -> None:
+    assert not _lint("if (x)\n  return false;\n")
+
+
+# --- NOLINT escape hatch, both placements ---
+
+
+def test_nolint_at_end_of_log_line_suppresses() -> None:
+    assert not _lint("if (x)\n  ESP_LOGD(t);  // NOLINT\n")
+
+
+def test_nolint_on_control_line_suppresses() -> None:
+    assert not _lint("if (x)  // NOLINT\n  ESP_LOGD(t);\n")

tests/script/test_clang_tidy_hash.py

@@ -81,3 +81,40 @@ def test_read_file_bytes(tmp_path: Path) -> None:
     result = clang_tidy_hash.read_file_bytes(test_file)
 
     assert result == test_content
+
+
+def test_calculate_idedata_cache_hash_changes_with_infra_code(tmp_path: Path) -> None:
+    _populate(tmp_path)
+    infra = tmp_path / "esphome" / "espidf" / "clang_tidy.py"
+    infra.parent.mkdir(parents=True)
+    infra.write_text("a")
+    before = clang_tidy_hash.calculate_idedata_cache_hash(repo_root=tmp_path)
+    assert before == clang_tidy_hash.calculate_idedata_cache_hash(repo_root=tmp_path)
+    infra.write_text("b")
+    assert clang_tidy_hash.calculate_idedata_cache_hash(repo_root=tmp_path) != before
+
+
+def test_calculate_idedata_cache_hash_includes_listed_files(tmp_path: Path) -> None:
+    _populate(tmp_path)
+    before = clang_tidy_hash.calculate_idedata_cache_hash(repo_root=tmp_path)
+    listed = tmp_path / "esphome" / "platformio" / "library.py"
+    listed.parent.mkdir(parents=True)
+    listed.write_text("x")
+    assert clang_tidy_hash.calculate_idedata_cache_hash(repo_root=tmp_path) != before
+
+
+def test_idedata_cache_hash_only_widens_for_esp32(tmp_path: Path) -> None:
+    _populate(tmp_path)
+    infra = tmp_path / "esphome" / "espidf" / "clang_tidy.py"
+    infra.parent.mkdir(parents=True)
+    infra.write_text("a")
+    esp32_before = clang_tidy_hash.idedata_cache_hash("esp32-idf-tidy", tmp_path)
+    other_before = clang_tidy_hash.idedata_cache_hash("esp8266-arduino-tidy", tmp_path)
+    infra.write_text("b")
+    assert (
+        clang_tidy_hash.idedata_cache_hash("esp32-idf-tidy", tmp_path) != esp32_before
+    )
+    assert (
+        clang_tidy_hash.idedata_cache_hash("esp8266-arduino-tidy", tmp_path)
+        == other_before
+    )

tests/script/test_determine_jobs.py

@@ -151,9 +151,14 @@ def test_main_all_tests_should_run(
         patch.object(determine_jobs, "_is_clang_tidy_full_scan", return_value=False),
         patch.object(
             determine_jobs,
-            "_all_integration_test_files",
+            "all_integration_test_files",
             return_value=fake_test_files,
         ),
+        patch.object(
+            determine_jobs,
+            "load_integration_durations",
+            return_value=dict.fromkeys(fake_test_files, 200.0),
+        ),
         patch.object(
             determine_jobs,
             "get_changed_components",
@@ -189,24 +194,12 @@ def test_main_all_tests_should_run(
     output = json.loads(captured.out)
 
     assert output["integration_tests"] is True
-    # run_all=True expands to the full glob and pre-buckets into 3 parts.
-    # Each bucket's `tests` is a JSON list of file paths.
+    assert output["integration_run_all"] is True
+    # run_all=True expands to the full glob; balance and naming are pinned
+    # by the unit tests, main() only needs to round-trip the structure
     assert isinstance(output["integration_test_buckets"], list)
-    assert len(output["integration_test_buckets"]) == 3
-    assert [b["name"] for b in output["integration_test_buckets"]] == [
-        "1/3",
-        "2/3",
-        "3/3",
-    ]
-    for bucket in output["integration_test_buckets"]:
-        assert isinstance(bucket["tests"], list)
-        for path in bucket["tests"]:
-            assert isinstance(path, str)
     bucket_files = [f for b in output["integration_test_buckets"] for f in b["tests"]]
-    assert bucket_files == fake_test_files
-    # Bucket sizes are balanced (max-min difference at most 1).
-    sizes = [len(b["tests"]) for b in output["integration_test_buckets"]]
-    assert max(sizes) - min(sizes) <= 1
+    assert sorted(bucket_files) == fake_test_files
     assert output["clang_tidy"] is True
     assert output["clang_tidy_mode"] in ["nosplit", "split"]
     assert output["clang_format"] is True
@@ -509,14 +502,24 @@ def test_compute_integration_test_buckets_at_threshold_stays_single() -> None:
 
 
 def test_compute_integration_test_buckets_just_over_threshold_splits() -> None:
-    """One file over the threshold triggers the 3-bucket fan-out, balanced."""
+    """One file over the threshold fans out fully when the weights demand it."""
     n = determine_jobs.INTEGRATION_TESTS_SPLIT_THRESHOLD + 1
     files = [f"tests/integration/test_{i:02d}.py" for i in range(n)]
-    run, buckets = determine_jobs._compute_integration_test_buckets(False, files)
+    with patch.object(
+        determine_jobs,
+        "load_integration_durations",
+        return_value=dict.fromkeys(files, 200.0),
+    ):
+        run, buckets = determine_jobs._compute_integration_test_buckets(False, files)
     assert run is True
-    assert [b["name"] for b in buckets] == ["1/3", "2/3", "3/3"]
-    union = [path for b in buckets for path in b["tests"]]
+    # threshold+1 files x 200s caps at the maximum bucket count.
+    n_buckets = determine_jobs.INTEGRATION_TESTS_SPLIT_BUCKETS
+    assert [b["name"] for b in buckets] == [
+        f"{i + 1}/{n_buckets}" for i in range(n_buckets)
+    ]
+    union = sorted(path for b in buckets for path in b["tests"])
     assert union == sorted(files)
+    # Equal weights => bucket sizes are balanced (difference at most 1).
     sizes = [len(b["tests"]) for b in buckets]
     assert max(sizes) - min(sizes) <= 1
 
@@ -526,7 +529,7 @@ def test_compute_integration_test_buckets_run_all_with_empty_glob_disables_run()
 ):
     """run_all=True but glob returns no files => run suppressed (otherwise
     pytest would collect tests outside tests/integration/)."""
-    with patch.object(determine_jobs, "_all_integration_test_files", return_value=[]):
+    with patch.object(determine_jobs, "all_integration_test_files", return_value=[]):
         run, buckets = determine_jobs._compute_integration_test_buckets(True, [])
     assert run is False
     assert buckets == []
@@ -552,6 +555,13 @@ def test_determine_integration_tests(
         assert run_all is True
         assert test_files == []
 
+    # Dependency pins and the session init fixture trigger run_all
+    for trigger in sorted(determine_jobs.INTEGRATION_TESTS_TRIGGER_FILES):
+        with patch.object(determine_jobs, "changed_files", return_value=[trigger]):
+            run_all, test_files = determine_jobs.determine_integration_tests()
+            assert run_all is True
+            assert test_files == []
+
     # Python files directly in esphome/ do NOT trigger tests
     with patch.object(
         determine_jobs, "changed_files", return_value=["esphome/config.py"]
@@ -1120,7 +1130,14 @@ def test_should_run_esp32_platformio_with_branch() -> None:
         (["esphome/espidf/runner.py"], True),
         (["esphome/espidf/framework.py"], True),
         (["esphome/build_gen/espidf.py"], True),
-        # PlatformIO build gen and esp32 component are NOT IDF-infra triggers
+        # Shared native-build modules the IDF build imports -> trigger
+        (["esphome/build_helpers/idedata.py"], True),
+        (["esphome/platformio/library.py"], True),
+        (["esphome/framework_helpers.py"], True),
+        (["esphome/platformio/extra_script.py"], True),
+        # PlatformIO build gen, its toolchain, and the esp32 component are
+        # NOT IDF-infra triggers
+        (["esphome/platformio/toolchain.py"], False),
         (["esphome/build_gen/platformio.py"], False),
         (["esphome/components/esp32/__init__.py"], False),
         (["README.md"], False),
@@ -1132,6 +1149,16 @@ def test_esp_idf_infra_changed(changed_files: list[str], expected: bool) -> None
     assert determine_jobs._esp_idf_infra_changed(changed_files) is expected
 
 
+def test_esp_idf_infra_trigger_paths_exist() -> None:
+    """A renamed or moved trigger module must fail here, not silently stop
+    forcing the esp32 IDF compile."""
+    repo_root = Path(__file__).resolve().parents[2]
+    for file in determine_jobs.ESP_IDF_INFRA_TRIGGER_FILES:
+        assert (repo_root / file).is_file(), f"trigger file {file} moved or renamed"
+    for prefix in determine_jobs.ESP_IDF_INFRA_TRIGGER_PATH_PREFIXES:
+        assert (repo_root / prefix).is_dir(), f"trigger dir {prefix} moved or renamed"
+
+
 @pytest.mark.parametrize(
     ("changed_files", "expected_result"),
     [
@@ -1197,6 +1224,56 @@ def test_count_changed_cpp_files_with_branch() -> None:
         mock_changed.assert_called_once_with("release")
 
 
+@pytest.mark.parametrize(
+    ("changed_files", "expected"),
+    [
+        # Core C++ change runs everything
+        (["esphome/core/helpers.cpp"], (True, [])),
+        # Core Python change runs everything too
+        (["esphome/core/config.py"], (True, [])),
+        # Component C++ change: component plus dependents with C++ tests
+        (["esphome/components/time/posix_tz.cpp"], (False, ["sntp", "time"])),
+        # Component Python change shapes the host build (defines, source
+        # filters), so it must trigger the same tests as a C++ change
+        (["esphome/components/time/__init__.py"], (False, ["sntp", "time"])),
+        # Nothing to build when no selected component has C++ tests
+        (["esphome/components/homeassistant/__init__.py"], (False, [])),
+        # Test manifest override changes only that component
+        (["tests/components/time/__init__.py"], (False, ["time"])),
+        # Test source change only that component
+        (["tests/components/time/posix_tz.cpp"], (False, ["time"])),
+        # pytest files and YAML build tests do not affect the test binary
+        (["tests/components/socket/conftest.py"], (False, [])),
+        (["tests/components/time/test.esp32-idf.yaml"], (False, [])),
+        (["README.md", "script/helpers.py"], (False, [])),
+        ([], (False, [])),
+    ],
+)
+def test_determine_cpp_unit_tests(
+    changed_files: list[str],
+    expected: tuple[bool, list[str]],
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Test which C++ unit tests a set of changed files selects."""
+    tests_dir = tmp_path / "tests" / "components"
+    for component in ("time", "sntp"):
+        (tests_dir / component).mkdir(parents=True)
+        (tests_dir / component / f"{component}.cpp").write_text("")
+    (tests_dir / "homeassistant").mkdir()
+    (tests_dir / "socket").mkdir()
+    monkeypatch.setattr(helpers, "root_path", str(tmp_path))
+    with (
+        patch.object(determine_jobs, "changed_files", return_value=changed_files),
+        patch.object(
+            helpers,
+            "create_components_graph",
+            return_value={"time": ["homeassistant", "sntp"]},
+        ),
+    ):
+        assert determine_jobs.determine_cpp_unit_tests() == expected
+
+
 def test_main_filters_components_without_tests(
     mock_determine_integration_tests: Mock,
     mock_should_run_clang_tidy: Mock,
@@ -2475,6 +2552,35 @@ def test_should_run_benchmarks_core_header_change() -> None:
         assert determine_jobs.should_run_benchmarks() is True
 
 
+def test_should_run_benchmarks_top_level_python_change() -> None:
+    """Test benchmarks trigger on top-level esphome Python module changes.
+
+    The Python benchmarks exercise config loading, so changes to modules
+    like config.py and yaml_util.py must run them; a regression in #16718
+    went unnoticed because these files matched no trigger.
+    """
+    for py_file in [
+        "esphome/config.py",
+        "esphome/yaml_util.py",
+        "esphome/__main__.py",
+        "esphome/helpers.py",
+    ]:
+        with patch.object(determine_jobs, "changed_files", return_value=[py_file]):
+            assert determine_jobs.should_run_benchmarks() is True, (
+                f"Expected benchmarks to run for {py_file}"
+            )
+
+
+def test_should_run_benchmarks_nested_python_change() -> None:
+    """Test benchmarks do NOT trigger for nested non-core Python changes."""
+    with patch.object(
+        determine_jobs,
+        "changed_files",
+        return_value=["esphome/dashboard/web_server.py"],
+    ):
+        assert determine_jobs.should_run_benchmarks() is False
+
+
 def test_should_run_benchmarks_host_platform_change() -> None:
     """Test benchmarks trigger on host platform changes.
 
@@ -2993,3 +3099,136 @@ def test_main_force_all_off_uses_detection(
     assert output["component_test_count"] == 0
     mock_determine_integration_tests.assert_called_once()
     mock_should_run_clang_tidy.assert_called_once()
+
+
+# Every platform the memory impact analysis can select must produce an ELF that
+# find_elf_path knows how to locate. The analysis fails the job when it cannot
+# find one, so a platform with an unknown layout would turn a clean build red.
+_MEMORY_IMPACT_ELF_LAYOUTS = {
+    # Native ESP-IDF toolchain (the esp32 default): <build>/build/firmware.elf
+    "esp32-c6-idf": "build/firmware.elf",
+    "esp32-idf": "build/firmware.elf",
+    "esp32-c3-idf": "build/firmware.elf",
+    "esp32-s2-idf": "build/firmware.elf",
+    "esp32-s3-idf": "build/firmware.elf",
+    # PlatformIO: <build>/.pioenvs/<name>/firmware.elf
+    "esp8266-ard": ".pioenvs/{name}/firmware.elf",
+    "rp2040-ard": ".pioenvs/{name}/firmware.elf",
+    "rp2350-ard": ".pioenvs/{name}/firmware.elf",
+    # LibreTiny: <build>/.pioenvs/<name>/raw_firmware.elf
+    "bk72xx-ard": ".pioenvs/{name}/raw_firmware.elf",
+    "rtl87xx-ard": ".pioenvs/{name}/raw_firmware.elf",
+    "ln882x-ard": ".pioenvs/{name}/raw_firmware.elf",
+    # Zephyr: <build>/.pioenvs/<name>/zephyr/[zephyr/]zephyr.elf
+    "nrf52-adafruit": ".pioenvs/{name}/zephyr/zephyr/zephyr.elf",
+}
+
+
+def test_memory_impact_platforms_have_known_elf_layout() -> None:
+    """Every selectable memory impact platform has a documented ELF layout.
+
+    Adding a platform to the preference list without teaching find_elf_path
+    where its ELF lands would fail the memory impact job on a clean build.
+    """
+    selectable = {
+        platform.value for platform in determine_jobs.MEMORY_IMPACT_PLATFORM_PREFERENCE
+    }
+    selectable.add(determine_jobs.MEMORY_IMPACT_FALLBACK_PLATFORM.value)
+
+    assert selectable == set(_MEMORY_IMPACT_ELF_LAYOUTS)
+
+
+def test_memory_impact_elf_layouts_are_found(tmp_path: Path) -> None:
+    """find_elf_path locates the ELF each memory impact platform produces."""
+    from esphome.analyze_memory.toolchain import find_elf_path
+
+    for platform, layout in _MEMORY_IMPACT_ELF_LAYOUTS.items():
+        build_path = tmp_path / platform / ".esphome" / "build" / "mydevice"
+        elf = build_path / layout.format(name=build_path.name)
+        elf.parent.mkdir(parents=True)
+        elf.write_text("")
+
+        assert find_elf_path(build_path) == elf, f"{platform} ELF not found"
+
+
+def test_compute_integration_test_buckets_no_durations_full_fanout() -> None:
+    """Without recorded durations the fan-out stays at the maximum."""
+    files = [f"tests/integration/test_{i:03d}.py" for i in range(15)]
+    with patch.object(determine_jobs, "load_integration_durations", return_value={}):
+        run, buckets = determine_jobs._compute_integration_test_buckets(False, files)
+    assert run is True
+    assert len(buckets) == determine_jobs.INTEGRATION_TESTS_SPLIT_BUCKETS
+    assert sorted(f for b in buckets for f in b["tests"]) == files
+
+
+def test_compute_integration_test_buckets_adaptive_count() -> None:
+    """A small recorded total weight collapses to one bucket above the threshold."""
+    files = [f"tests/integration/test_{i:03d}.py" for i in range(15)]
+    with patch.object(
+        determine_jobs,
+        "load_integration_durations",
+        return_value=dict.fromkeys(files, 10.0),
+    ):
+        run, buckets = determine_jobs._compute_integration_test_buckets(False, files)
+    assert run is True
+    # 15 files x 10s recorded = 150s, under the per-bucket weight target.
+    assert [b["name"] for b in buckets] == ["1/1"]
+    assert buckets[0]["tests"] == files
+
+
+def test_compute_integration_test_buckets_duration_weighted() -> None:
+    """Heavy files spread across buckets instead of clustering by sorted name."""
+    files = [f"tests/integration/test_{i:03d}.py" for i in range(12)]
+    durations = dict.fromkeys(files, 10.0)
+    durations[files[0]] = 600.0
+    durations[files[1]] = 600.0
+    with patch.object(
+        determine_jobs, "load_integration_durations", return_value=durations
+    ):
+        run, buckets = determine_jobs._compute_integration_test_buckets(False, files)
+    assert run is True
+    assert len(buckets) >= 2
+    heavy_buckets = [b for b in buckets if set(files[:2]) & set(b["tests"])]
+    assert len(heavy_buckets) == 2, "heavy files should land in different buckets"
+    assert sorted(f for b in buckets for f in b["tests"]) == files
+
+
+def test_load_integration_durations_missing_or_corrupt(tmp_path: Path) -> None:
+    """Missing or unparsable durations data degrades to an empty mapping."""
+    with patch.object(helpers, "root_path", str(tmp_path)):
+        assert determine_jobs.load_integration_durations() == {}
+        durations_file = tmp_path / helpers.INTEGRATION_TEST_DURATIONS_FILE
+        durations_file.parent.mkdir(parents=True)
+        durations_file.write_text("not json")
+        assert determine_jobs.load_integration_durations() == {}
+        durations_file.write_text('{"tests/integration/test_a.py": 12.5}')
+        assert determine_jobs.load_integration_durations() == {
+            "tests/integration/test_a.py": 12.5
+        }
+        # Non-positive entries are dropped, valid ones survive
+        durations_file.write_text(
+            '{"tests/integration/test_a.py": 12.5, "tests/integration/test_b.py": -1}'
+        )
+        assert determine_jobs.load_integration_durations() == {
+            "tests/integration/test_a.py": 12.5
+        }
+        # One non-numeric entry cannot discard the whole recording
+        durations_file.write_text(
+            '{"tests/integration/test_a.py": 12.5, "tests/integration/test_b.py": null}'
+        )
+        assert determine_jobs.load_integration_durations() == {
+            "tests/integration/test_a.py": 12.5
+        }
+        # A non-dict top level degrades to empty
+        durations_file.write_text("[12.5]")
+        assert determine_jobs.load_integration_durations() == {}
+
+
+def test_committed_integration_durations_are_sane() -> None:
+    """The committed recording itself holds positive bounded floats."""
+    raw = json.loads(
+        (Path(helpers.root_path) / helpers.INTEGRATION_TEST_DURATIONS_FILE).read_text()
+    )
+    assert raw, "committed durations file missing or empty"
+    assert all(isinstance(v, (int, float)) and 0 < v < 86400 for v in raw.values())
+    assert all(k.startswith("tests/integration/test_") for k in raw)

tests/script/test_docker_build.py

@@ -71,10 +71,12 @@ def test_branch_manifest_targets_ghcr_only(
     )
 
     assert commands == [
-        "docker buildx imagetools create "
-        "--tag ghcr.io/esphome/esphome-hassio:my-branch "
-        "ghcr.io/esphome/esphome-hassio-amd64:my-branch "
-        "ghcr.io/esphome/esphome-hassio-aarch64:my-branch"
+        (
+            "docker buildx imagetools create "
+            "--tag ghcr.io/esphome/esphome-hassio:my-branch "
+            "ghcr.io/esphome/esphome-hassio-amd64:my-branch "
+            "ghcr.io/esphome/esphome-hassio-aarch64:my-branch"
+        )
     ]
 
 

tests/script/test_helpers.py

@@ -20,6 +20,7 @@ changed_files = helpers.changed_files
 filter_changed = helpers.filter_changed
 get_changed_components = helpers.get_changed_components
 _get_changed_files_from_command = helpers._get_changed_files_from_command
+run_gh_command = helpers.run_gh_command
 _get_pr_number_from_github_env = helpers._get_pr_number_from_github_env
 _get_changed_files_github_actions = helpers._get_changed_files_github_actions
 _filter_changed_ci = helpers._filter_changed_ci
@@ -79,6 +80,22 @@ def test_get_pr_number_from_github_env_event_file(
     assert result == "5678"
 
 
+def test_get_github_event_data_decodes_utf8_regardless_of_locale(
+    monkeypatch: MonkeyPatch, tmp_path: Path
+) -> None:
+    """The event payload is UTF-8; parsing must not depend on the platform
+    default encoding. On Windows the default is cp1252, which raised
+    UnicodeDecodeError as soon as a commit title carried non ASCII text."""
+    event_file = tmp_path / "event.json"
+    event_data = {"head_commit": {"message": "Answer UNPAIR with Response… é"}}
+    event_file.write_bytes(json.dumps(event_data, ensure_ascii=False).encode("utf-8"))
+    monkeypatch.setenv("GITHUB_EVENT_PATH", str(event_file))
+
+    result = helpers._get_github_event_data()
+
+    assert result == event_data
+
+
 def test_get_pr_number_from_github_env_no_pr(
     monkeypatch: MonkeyPatch, tmp_path: Path
 ) -> None:
@@ -227,6 +244,44 @@ def test_get_changed_files_github_actions_pull_request_large_pr(
         assert result == expected_files
 
 
+def test_get_changed_files_github_actions_pull_request_large_diff(
+    monkeypatch: MonkeyPatch,
+) -> None:
+    """Test _get_changed_files_github_actions fallback for PRs with >20000 diff lines."""
+    monkeypatch.setenv("GITHUB_EVENT_NAME", "pull_request")
+
+    expected_files = ["file1.py", "file2.cpp"]
+
+    with (
+        patch("helpers._get_pr_number_from_github_env", return_value="17909"),
+        patch("helpers._get_changed_files_from_command") as mock_get,
+    ):
+        # First call fails with too many diff lines error, second succeeds with API method
+        mock_get.side_effect = [
+            Exception(
+                "could not find pull request diff: HTTP 406: Sorry, "
+                "the diff exceeded the maximum number of lines (20000)"
+            ),
+            expected_files,
+        ]
+
+        result = _get_changed_files_github_actions()
+
+        assert mock_get.call_count == 2
+        mock_get.assert_any_call(["gh", "pr", "diff", "17909", "--name-only"])
+        mock_get.assert_any_call(
+            [
+                "gh",
+                "api",
+                "repos/esphome/esphome/pulls/17909/files",
+                "--paginate",
+                "--jq",
+                ".[].filename",
+            ]
+        )
+        assert result == expected_files
+
+
 def test_get_changed_files_github_actions_pull_request_other_error(
     monkeypatch: MonkeyPatch,
 ) -> None:
@@ -1835,3 +1890,288 @@ def test_get_component_test_files_component_without_tests(
 )
 def test_is_validate_only_file(filename: str, expected: bool, tmp_path: Path) -> None:
     assert helpers.is_validate_only_file(tmp_path / filename) is expected
+
+
+@pytest.mark.parametrize(
+    ("files", "expected"),
+    [
+        (["esphome/config.py"], True),
+        (["esphome/yaml_util.py"], True),
+        (["esphome/__main__.py"], True),
+        (["esphome/const.pyi"], True),
+        (["README.md", "esphome/helpers.py"], True),
+        (["esphome/core/config.py"], False),
+        (["esphome/components/sensor/__init__.py"], False),
+        (["esphome/dashboard/web_server.py"], False),
+        (["esphome/idf_component.yml"], False),
+        (["tests/unit_tests/test_config.py"], False),
+        ([], False),
+    ],
+)
+def test_base_python_changed(files: list[str], expected: bool) -> None:
+    """Only Python modules directly in esphome/ count as base Python changes."""
+    assert helpers.base_python_changed(files) is expected
+
+
+def _gh_error(stderr: str) -> subprocess.CalledProcessError:
+    return subprocess.CalledProcessError(1, ["gh"], output="", stderr=stderr)
+
+
+def _gh_success(stdout: str = "ok\n") -> subprocess.CompletedProcess:
+    return subprocess.CompletedProcess(["gh"], 0, stdout=stdout, stderr="")
+
+
+def test_run_gh_command_success() -> None:
+    """A successful command returns without retrying."""
+    with patch("helpers.subprocess.run", return_value=_gh_success()) as mock_run:
+        result = run_gh_command(["gh", "pr", "diff", "123", "--name-only"])
+
+    assert result.stdout == "ok\n"
+    mock_run.assert_called_once()
+
+
+@pytest.mark.parametrize(
+    "second_error",
+    [
+        (
+            'Post "https://api.github.com/graphql": tls: failed to verify'
+            " certificate: x509: certificate is not valid for any names,"
+            " but wanted to match api.github.com"
+        ),
+        'Post "https://api.github.com/graphql": EOF',
+        (
+            "error connecting to api.github.com\n"
+            "check your internet connection or https://githubstatus.com"
+        ),
+    ],
+)
+def test_run_gh_command_retries_transient_error(second_error: str) -> None:
+    """Transient server errors are retried with 2s/4s backoff."""
+    with (
+        patch(
+            "helpers.subprocess.run",
+            side_effect=[
+                _gh_error("HTTP 502: 502 Bad Gateway (https://api.github.com/graphql)"),
+                _gh_error(second_error),
+                _gh_success(),
+            ],
+        ) as mock_run,
+        patch("helpers.time.sleep") as mock_sleep,
+    ):
+        result = run_gh_command(["gh", "pr", "diff", "123", "--name-only"])
+
+    assert result.stdout == "ok\n"
+    assert mock_run.call_count == 3
+    assert [call.args[0] for call in mock_sleep.call_args_list] == [2, 4]
+
+
+def test_run_gh_command_gives_up_after_max_attempts() -> None:
+    """A persistent transient error raises after the third attempt."""
+    with (
+        patch(
+            "helpers.subprocess.run",
+            side_effect=_gh_error("HTTP 503: Service Unavailable"),
+        ) as mock_run,
+        patch("helpers.time.sleep") as mock_sleep,
+        pytest.raises(subprocess.CalledProcessError),
+    ):
+        run_gh_command(["gh", "pr", "diff", "123", "--name-only"])
+
+    assert mock_run.call_count == 3
+    assert mock_sleep.call_count == 2
+
+
+@pytest.mark.parametrize(
+    "stderr",
+    [
+        "HTTP 404: Not Found (https://api.github.com/repos/x)",
+        "HTTP 401: Bad credentials",
+        "HTTP 403: API rate limit exceeded for installation ID 123.",
+        "diff exceeded the maximum number of changed files (300)",
+        (
+            "GraphQL: Could not resolve to a PullRequest with the number of 999999."
+            " (repository.pullRequest)"
+        ),
+    ],
+)
+def test_run_gh_command_permanent_error_not_retried(stderr: str) -> None:
+    """Permanent failures raise immediately without any retry."""
+    with (
+        patch("helpers.subprocess.run", side_effect=_gh_error(stderr)) as mock_run,
+        patch("helpers.time.sleep") as mock_sleep,
+        pytest.raises(subprocess.CalledProcessError),
+    ):
+        run_gh_command(["gh", "pr", "diff", "123", "--name-only"])
+
+    mock_run.assert_called_once()
+    mock_sleep.assert_not_called()
+
+
+def test_run_gh_command_no_retry_for_non_idempotent_commands() -> None:
+    """retry=False fails on the first error even when it looks transient."""
+    with (
+        patch(
+            "helpers.subprocess.run",
+            side_effect=_gh_error("HTTP 502: 502 Bad Gateway"),
+        ) as mock_run,
+        patch("helpers.time.sleep") as mock_sleep,
+        pytest.raises(subprocess.CalledProcessError),
+    ):
+        run_gh_command(["gh", "pr", "comment", "123", "--body", "x"], retry=False)
+
+    mock_run.assert_called_once()
+    mock_sleep.assert_not_called()
+
+
+def test_get_changed_files_from_command_gh_failure_keeps_stderr() -> None:
+    """Failures from gh surface stderr so callers can detect the 300-file limit."""
+    stderr = "diff exceeded the maximum number of changed files (300)"
+    with (
+        patch("helpers.subprocess.run", side_effect=_gh_error(stderr)),
+        pytest.raises(Exception, match="maximum number of changed files"),
+    ):
+        _get_changed_files_from_command(["gh", "pr", "diff", "123", "--name-only"])
+
+
+@pytest.mark.parametrize(
+    ("file_path", "expected"),
+    [
+        ("esphome/components/time/posix_tz.cpp", True),
+        ("esphome/components/time/posix_tz.h", True),
+        ("esphome/components/time/__init__.py", True),
+        ("esphome/components/sntp/time.py", True),
+        ("tests/components/time/posix_tz.cpp", True),
+        ("tests/components/time/__init__.py", True),
+        # Platform override: tests/components/<component>/<domain>/__init__.py
+        ("tests/components/template/sensor/__init__.py", True),
+        # pytest-only files do not shape the C++ test binary
+        ("tests/components/socket/conftest.py", False),
+        ("tests/components/socket/test_socket.py", False),
+        ("tests/components/time/test.esp32-idf.yaml", False),
+        ("esphome/core/time.cpp", False),
+        ("esphome/config.py", False),
+        ("script/helpers.py", False),
+        ("README.md", False),
+    ],
+)
+def test_filter_cpp_unit_test_files(file_path: str, expected: bool) -> None:
+    """Test which changed files can affect a component's C++ unit test build."""
+    assert helpers.filter_cpp_unit_test_files(file_path) is expected
+
+
+@pytest.fixture
+def cpp_unit_test_tree(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
+    """Fake repo root where time, sntp and api have C++ unit tests.
+
+    homeassistant depends on time but has no C++ tests, so it must be
+    dropped from the selection; socket has only pytest files.
+    """
+    tests_dir = tmp_path / "tests" / "components"
+    for component in ("time", "sntp", "api"):
+        (tests_dir / component).mkdir(parents=True)
+        (tests_dir / component / f"{component}.cpp").write_text("")
+    (tests_dir / "homeassistant").mkdir()
+    (tests_dir / "homeassistant" / "__init__.py").write_text("")
+    (tests_dir / "socket").mkdir()
+    (tests_dir / "socket" / "conftest.py").write_text("")
+    monkeypatch.setattr(helpers, "root_path", str(tmp_path))
+    monkeypatch.setattr(
+        helpers,
+        "create_components_graph",
+        lambda: {"time": ["homeassistant", "sntp"]},
+    )
+    return tmp_path
+
+
+@pytest.mark.parametrize(
+    ("files", "expected"),
+    [
+        # Component changes expand to dependents with C++ tests
+        (["esphome/components/time/posix_tz.cpp"], ["sntp", "time"]),
+        (["esphome/components/time/__init__.py"], ["sntp", "time"]),
+        # Dependent without C++ tests is dropped
+        (["esphome/components/homeassistant/__init__.py"], []),
+        # Test changes select only that component
+        (["tests/components/time/posix_tz.cpp"], ["time"]),
+        (["tests/components/time/__init__.py"], ["time"]),
+        (["tests/components/homeassistant/__init__.py"], []),
+        (["tests/components/socket/conftest.py"], []),
+        (["tests/components/time/test.esp32-idf.yaml"], []),
+        (
+            ["esphome/components/time/__init__.py", "tests/components/api/api.cpp"],
+            ["api", "sntp", "time"],
+        ),
+        ([], []),
+    ],
+)
+@pytest.mark.usefixtures("cpp_unit_test_tree")
+def test_get_cpp_changed_components(files: list[str], expected: list[str]) -> None:
+    """Test that C++ and Python component changes select the right unit tests."""
+    assert helpers.get_cpp_changed_components(files) == expected
+
+
+def test_get_cpp_changed_components_independent_of_cwd(
+    cpp_unit_test_tree: Path,
+    tmp_path_factory: pytest.TempPathFactory,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Test directories resolve against root_path, not the current directory."""
+    monkeypatch.chdir(tmp_path_factory.mktemp("elsewhere"))
+    assert helpers.get_cpp_changed_components(
+        ["tests/components/time/__init__.py"]
+    ) == ["time"]
+
+
+def test_fixture_map_includes_shared_yaml_markers() -> None:
+    """Fixtures named only by shared_yaml markers must map to their test file."""
+    helpers.get_fixture_to_test_files.cache_clear()
+    mapping = helpers.get_fixture_to_test_files()
+    for fixture in (
+        "uart_mock_modbus_loopback",
+        "uart_mock_modbus_mesh",
+        "uart_mock_modbus_server_injected",
+    ):
+        assert mapping[fixture] == frozenset(
+            {"tests/integration/test_uart_mock_modbus.py"}
+        )
+
+
+def test_no_orphan_integration_fixtures() -> None:
+    """Every fixture must reach CI test selection; an orphan selects nothing."""
+    helpers.get_fixture_to_test_files.cache_clear()
+    mapping = helpers.get_fixture_to_test_files()
+    fixtures_dir = (Path(__file__).parent.parent / "integration" / "fixtures").resolve()
+    fixtures = list(fixtures_dir.glob("*.yaml"))
+    assert fixtures, f"no fixtures found under {fixtures_dir}"
+    # cache_init is covered via INTEGRATION_TESTS_TRIGGER_FILES instead
+    orphans = [
+        f.stem for f in fixtures if f.stem != "cache_init" and f.stem not in mapping
+    ]
+    assert not orphans, f"fixtures invisible to CI test selection: {orphans}"
+
+
+def test_lpt_partition_balances_skewed_weights() -> None:
+    """Heavy items spread across groups instead of clustering."""
+    items = [f"i{n}" for n in range(6)]
+    weights = {"i0": 100.0, "i1": 90.0, "i2": 10.0, "i3": 10.0, "i4": 5.0, "i5": 5.0}
+    groups = helpers.lpt_partition(items, weights, 2)
+    group_weights = sorted(sum(weights[i] for i in g) for g in groups)
+    # Contiguous split would give 200 vs 20; LPT lands at 110 vs 110
+    assert group_weights == [110.0, 110.0]
+    assert sorted(i for g in groups for i in g) == items
+
+
+def test_lpt_partition_more_groups_than_items() -> None:
+    """Surplus groups come back empty; every item still lands somewhere."""
+    items = ["a", "b"]
+    groups = helpers.lpt_partition(items, {"a": 1.0, "b": 1.0}, 4)
+    assert len(groups) == 4
+    assert sorted(i for g in groups for i in g) == items
+    assert sum(not g for g in groups) == 2
+
+
+def test_lpt_partition_tie_determinism() -> None:
+    """Equal weights assign in input order, so output is reproducible."""
+    items = [f"i{n}" for n in range(4)]
+    weights = dict.fromkeys(items, 1.0)
+    assert helpers.lpt_partition(items, weights, 2) == [["i0", "i2"], ["i1", "i3"]]

tests/script/test_platformio_install_deps.py

@@ -0,0 +1,649 @@
+"""Tests for script/platformio_install_deps.py."""
+
+from argparse import Namespace
+import importlib.util
+import inspect
+from pathlib import Path
+import shutil
+import sys
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from platformio import fs
+from platformio.cache import ContentCache
+from platformio.exception import InvalidJSONFile
+from platformio.package.manager._install import PackageManagerInstallMixin
+from platformio.package.manager.base import BasePackageManager
+from platformio.package.manager.library import LibraryPackageManager
+from platformio.package.manager.tool import ToolPackageManager
+from platformio.package.meta import PackageCompatibility, PackageItem, PackageSpec
+import pytest
+from semantic_version import Version
+
+_SCRIPT = Path(__file__).parents[2] / "script" / "platformio_install_deps.py"
+
+
+def _load_script():
+    spec = importlib.util.spec_from_file_location("platformio_install_deps", _SCRIPT)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    # The real ContentCache would create dirs under the user's core dir
+    module.ContentCache = lambda *_: None
+    return module
+
+
+def test_spec_key_collapses_destinations() -> None:
+    """Two specs delivering one package share a directory and one key."""
+    mod = _load_script()
+    assert mod.spec_key("esphome/noise-c @ 1.0") == "noise-c"
+    assert mod.spec_key("esphome/noise-c@1.0") == "noise-c"
+    assert mod.spec_key("ESP32Async/AsyncTCP @ ^3.4.10") == mod.spec_key(
+        "esp32async/asynctcp @ 3.5.0"
+    )
+    url = "https://github.com/pioarduino/platform-espressif32/releases/download/{v}/platform-espressif32.zip"
+    assert mod.spec_key(url.format(v="55.03.311")) == mod.spec_key(
+        url.format(v="54.03.20")
+    )
+
+
+def test_parse_specs_and_cli_args(tmp_path: Path) -> None:
+    """Parsing skips unpinned and interpolated entries; the CLI rebuild
+    keeps the original flag pairing."""
+    ini = tmp_path / "platformio.ini"
+    ini.write_text(
+        "[env:a]\n"
+        "platform = fake/platform@1\n"
+        "lib_deps =\n"
+        "    esphome/noise-c @ 1.0\n"
+        "    ${common.lib_deps}\n"
+        "    internal_lib\n"
+        "[env:b]\n"
+        "lib_deps =\n"
+        "    esphome/noise-c @ 1.0\n"
+    )
+    mod = _load_script()
+    args = Namespace(libraries=True, platforms=True, tools=False)
+    libs, platforms, tools = mod.parse_specs(str(ini), args)
+    # exact-string duplicates collapse; distinct version pins survive
+    assert libs == ["esphome/noise-c @ 1.0"]
+    assert platforms == ["fake/platform@1"]
+    assert tools == []
+    assert mod.build_cli_args(libs, platforms, tools) == [
+        "-l",
+        "esphome/noise-c @ 1.0",
+        "-p",
+        "fake/platform@1",
+    ]
+
+
+class _FakeManager:
+    """Scripted manager_cls: records installs, raises on demand."""
+
+    installed: set = set()
+    fail: set = set()
+    calls: list = []
+    lock_events: list = []
+    base_dir: str = ""  # per-test tmp base; set by _reset_fake
+
+    def __init__(self, package_dir) -> None:
+        assert package_dir is None
+
+    @staticmethod
+    def _key(spec) -> str:
+        return spec if isinstance(spec, str) else str(spec)
+
+    def get_package(self, spec):
+        if self._key(spec) in self.installed:
+            return SimpleNamespace(path="/tmp/fake-pkg", spec=self._key(spec))
+        return None
+
+    def memcache_reset(self) -> None:
+        type(self).resets = getattr(type(self), "resets", 0) + 1
+
+    @property
+    def package_dir(self) -> str:
+        return str(Path(type(self).base_dir) / "packages")
+
+    def get_download_dir(self) -> str:
+        return str(Path(type(self).base_dir) / "downloads")
+
+    def get_tmp_dir(self) -> str:
+        return str(Path(type(self).base_dir) / "tmp")
+
+    def lock(self) -> None:
+        type(self).lock_events.append("lock")
+
+    def unlock(self) -> None:
+        type(self).lock_events.append("unlock")
+
+    def _install(self, spec, skip_dependencies, compatibility=None):
+        assert skip_dependencies is True
+        if self._key(spec) in self.fail:
+            raise RuntimeError("boom")
+        type(self).calls.append(spec)
+        type(self).compat_calls.append((self._key(spec), compatibility))
+        type(self).installed.add(self._key(spec))  # atomic under the GIL
+
+    def get_pkg_dependencies(self, pkg):
+        return getattr(type(self), "deps", {}).get(pkg.spec)
+
+    dependency_to_spec = staticmethod(BasePackageManager.dependency_to_spec)
+
+
+def _reset_fake(base_dir: str = "", **kwargs) -> type:
+    # A fresh subclass per test: nothing leaks between tests through the
+    # class-level scripted state
+    return type(
+        "_ScriptedManager",
+        (_FakeManager,),
+        {
+            "base_dir": base_dir,
+            "installed": kwargs.get("installed", set()),
+            "fail": kwargs.get("fail", set()),
+            "calls": [],
+            "compat_calls": [],
+            "lock_events": [],
+        },
+    )
+
+
+def test_parallel_install_empty_specs_is_a_no_op(tmp_path: Path) -> None:
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path))
+    mod.parallel_install(cls, [])
+    assert cls.calls == [] and cls.lock_events == []
+
+
+def test_parallel_install_behavior(tmp_path: Path) -> None:
+    """Duplicates collapse to one install, installed specs are filtered,
+    URL specs stay out of the wave, and the lock wraps the pool."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), installed={"esphome/already @ 1.0"})
+    mod.parallel_install(
+        cls,
+        [
+            "esphome/noise-c @ 1.0",
+            "esphome/noise-c @ 1.0",
+            "esphome/already @ 1.0",
+            "https://x/framework.tar.xz",
+        ],
+    )
+    assert cls.calls == ["esphome/noise-c @ 1.0"]
+    assert cls.lock_events == ["lock", "unlock"]
+
+
+def test_parallel_install_failure_cleans_torn_destination(
+    tmp_path: Path, capsys
+) -> None:
+    """A failed install resets the memcache, removes what get_package can
+    see, and reports; the others still install."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+
+    removed = []
+
+    torn = str(tmp_path / "packages" / "torn-pkg")  # never created; only rmtree'd
+
+    def get_package(self, spec):
+        if spec == "esphome/bad @ 1.0" and getattr(cls, "resets", 0):
+            return SimpleNamespace(path=torn, spec=spec)
+        return _FakeManager.get_package(self, spec)
+
+    cls.get_package = get_package  # throwaway subclass; nothing to restore
+    with patch.object(mod.fs, "rmtree", side_effect=removed.append):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0", "esphome/good @ 1.0"])
+    assert "esphome/good @ 1.0" in cls.calls
+    assert removed == [torn]
+    out = capsys.readouterr().out
+    assert "Pre-install of esphome/bad @ 1.0 failed" in out
+    assert "Pre-install failed for 1 of 2 package(s)" in out
+
+
+def test_parallel_install_runs_dependency_waves(tmp_path: Path) -> None:
+    """Dependencies of wave-installed packages install in a second wave,
+    deduped by name; name-only platform libs stay with the serial pass."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path))
+    cls.deps = {
+        "esphome/noise-c @ 1.0": [
+            {"owner": "esphome", "name": "libsodium", "version": "^1.0"},
+            {"name": "SPI"},
+        ],
+        "esphome/wg @ 1.0": [
+            {"owner": "esphome", "name": "libsodium", "version": "^1.0"},
+        ],
+    }
+    mod.parallel_install(cls, ["esphome/noise-c @ 1.0", "esphome/wg @ 1.0"])
+    assert len(cls.calls) == 3  # the shared dep installs exactly once
+    assert {mod.spec_key(c) for c in cls.calls} == {"noise-c", "wg", "libsodium"}
+    # Wave-1 strings carry no compatibility; the dependency wave does
+    compats = dict(cls.compat_calls)
+    assert compats["esphome/noise-c @ 1.0"] is None
+    dep_compat = next(v for k, v in cls.compat_calls if "libsodium" in k)
+    assert dep_compat is not None  # mirrors pio's install_dependency
+
+
+def test_dependency_wave_excludes_url_specs(tmp_path: Path) -> None:
+    """A dependency pinned to a URL surfaces as spec.uri; it must stay out
+    of the wave like string URL specs do."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path))
+    cls.deps = {
+        "esphome/noise-c @ 1.0": [
+            {"name": "vendored", "version": "https://github.com/x/y.git"},
+        ],
+    }
+    mod.parallel_install(cls, ["esphome/noise-c @ 1.0"])
+    assert {mod.spec_key(c) for c in cls.calls} == {"noise-c"}
+
+
+def test_failed_cleanup_fails_the_build(tmp_path: Path) -> None:
+    """A torn destination still on disk after rmtree must fail the build:
+    fs.rmtree never raises (its onexc handler prints), so only the
+    destination's absence proves the cleanup worked."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    torn = tmp_path / "packages" / "torn-pkg"
+    torn.mkdir(parents=True)
+
+    def get_package(self, spec):
+        if getattr(cls, "resets", 0):
+            return SimpleNamespace(path=str(torn), spec=spec)
+        return None
+
+    cls.get_package = get_package  # throwaway subclass; nothing to restore
+    with (
+        patch.object(mod.fs, "rmtree", lambda path: None),  # onexc swallowed
+        pytest.raises(mod.CleanupError, match="could not remove"),
+    ):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert cls.lock_events == ["lock", "unlock"]  # still released
+
+
+def test_unverifiable_torn_destination_fails_the_build(tmp_path: Path) -> None:
+    """When the scan fails, the spec's own .piopm decides: an unremovable
+    leftover fails the build."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    dest = Path(cls.base_dir) / "packages" / "bad"
+    dest.mkdir(parents=True)
+    (dest / ".piopm").write_text('{"spec": {"owner": "esphome", "name": "bad"}}')
+
+    def bad_reset(self):
+        raise OSError("scan broken")
+
+    cls.memcache_reset = bad_reset
+    with (
+        patch.object(mod.fs, "rmtree", lambda path: None),  # onexc swallowed
+        pytest.raises(mod.CleanupError, match="could not remove"),
+    ):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+
+
+def test_unverifiable_scan_without_leftover_degrades(tmp_path: Path, capsys) -> None:
+    """A failing scan with no destination on disk is never a build
+    failure blaming this spec."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    resets = {"n": 0}
+
+    def bad_reset(self):
+        # Fail clean_torn's reset; the coordinator's later reset works
+        resets["n"] += 1
+        if resets["n"] <= 1:
+            raise OSError("scan broken")
+
+    cls.memcache_reset = bad_reset
+    mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert "No resolvable destination to clean" in capsys.readouterr().out
+
+
+def test_unresolvable_torn_destination_is_printed(tmp_path: Path, capsys) -> None:
+    """A failed install with no resolvable package prints, so an invisible
+    torn directory is at least traceable."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert "No resolvable destination to clean" in capsys.readouterr().out
+
+
+def test_unparsable_torn_destination_is_removed(tmp_path: Path, capsys) -> None:
+    """A torn dir get_package cannot resolve but whose .piopm names the
+    spec is removed instead of surviving into the serial pass."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    dest = Path(cls.base_dir) / "packages" / "bad"
+    dest.mkdir(parents=True)
+    (dest / ".piopm").write_text('{"spec": {"owner": "esphome", "name": "bad"}}')
+
+    with patch.object(mod.fs, "rmtree", shutil.rmtree):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert not dest.exists()
+    assert "Removed torn destination" in capsys.readouterr().out
+
+
+def test_parse_specs_tools_branch(tmp_path: Path) -> None:
+    """platform_packages parsing keeps owner'd tools and rewrites github
+    URL pins to bare URLs the wave then skips via parsed.uri."""
+    mod = _load_script()
+    ini = tmp_path / "platformio.ini"
+    ini.write_text(
+        "[env:t]\n"
+        "platform_packages =\n"
+        "    ${common.platform_packages}\n"
+        "    platformio/tool-scons@~4.40801.0\n"
+        "    framework-arduinopico@https://github.com/earlephilhower/arduino-pico/releases/download/6.0.0/rp2040-6.0.0.zip\n"
+    )
+    args = Namespace(libraries=False, platforms=False, tools=True)
+    libs, platforms, tools = mod.parse_specs(str(ini), args)
+    assert libs == [] and platforms == []
+    assert tools == [
+        "platformio/tool-scons@~4.40801.0",
+        "https://github.com/earlephilhower/arduino-pico/releases/download/6.0.0/rp2040-6.0.0.zip",
+    ]
+    assert mod.build_cli_args([], [], tools)[:2] == ["-t", tools[0]]
+
+
+def test_warm_store_still_walks_dependencies(tmp_path: Path) -> None:
+    """Already-installed top-level packages still feed the dependency
+    wave; a warm store can be missing a transitive dep."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), installed={"esphome/noise-c @ 1.0"})
+    cls.deps = {
+        "esphome/noise-c @ 1.0": [
+            {"owner": "esphome", "name": "libsodium", "version": "^1.0"},
+        ],
+    }
+    mod.parallel_install(cls, ["esphome/noise-c @ 1.0"])
+    assert [mod.spec_key(c) for c in cls.calls] == ["libsodium"]
+
+
+def test_worker_system_exit_still_cleans(tmp_path: Path, capsys) -> None:
+    """A worker SystemExit runs the torn cleanup before propagating; the
+    serial pass must never trust its leftovers."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path))
+    torn = tmp_path / "packages" / "torn-pkg"
+    torn.mkdir(parents=True)
+
+    def exiting_install(self, spec, skip_dependencies, compatibility=None):
+        raise SystemExit(0)
+
+    def get_package(self, spec):
+        if getattr(cls, "resets", 0):
+            return SimpleNamespace(path=str(torn), spec=spec)
+        return None
+
+    cls._install = exiting_install
+    cls.get_package = get_package
+
+    def real_rmtree(path):
+        Path(path).rmdir()
+
+    with (
+        patch.object(mod.fs, "rmtree", real_rmtree),
+        pytest.raises(SystemExit),
+    ):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert not torn.exists()
+
+
+def test_unlock_failure_is_fatal(tmp_path: Path) -> None:
+    """A failed unlock must fail the build: the serial pass in another
+    process would block on the held flock."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path))
+
+    def bad_unlock(self):
+        raise OSError("flock broke")
+
+    cls.unlock = bad_unlock
+    with pytest.raises(mod.LockReleaseError, match="manager lock"):
+        mod.parallel_install(cls, ["esphome/good @ 1.0"])
+
+
+def test_unlock_failure_keeps_inflight_error_as_context(tmp_path: Path) -> None:
+    """An in-flight CleanupError stays attached when the unlock fault
+    takes over the raise."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    torn = tmp_path / "packages" / "bad"
+    torn.mkdir(parents=True)
+
+    def get_package(self, spec):
+        if getattr(cls, "resets", 0):
+            return SimpleNamespace(path=str(torn), spec=spec)
+        return None
+
+    def bad_unlock(self):
+        raise OSError("flock broke")
+
+    cls.get_package = get_package
+    cls.unlock = bad_unlock
+    with (
+        patch.object(mod.fs, "rmtree", lambda path: None),  # leaves torn
+        pytest.raises(mod.LockReleaseError) as err,
+    ):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert isinstance(err.value.__cause__.__context__, mod.CleanupError)
+
+
+def test_chdir_failure_does_not_fail_the_wave(tmp_path: Path, monkeypatch) -> None:
+    """A lost cwd is suppressed: further waves may misbehave and fall to
+    the serial pass, whose cwd is pinned."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path))
+    monkeypatch.setattr(mod.os, "chdir", MagicMock(side_effect=OSError("gone")))
+    mod.parallel_install(cls, ["esphome/good @ 1.0"])
+    assert cls.calls == ["esphome/good @ 1.0"]
+
+
+def test_piopm_match_removes_manifest_named_torn_dir(tmp_path: Path, capsys) -> None:
+    """A torn dir named by its manifest (not the registry spec) is found
+    through its .piopm and removed."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    torn = tmp_path / "packages" / "ManifestName"
+    torn.mkdir(parents=True)
+    (torn / ".piopm").write_text('{"spec": {"owner": "esphome", "name": "bad"}}')
+    innocent = tmp_path / "packages" / "innocent"
+    innocent.mkdir()
+    (innocent / ".piopm").write_text('{"spec": {"owner": "o", "name": "other"}}')
+    with patch.object(mod.fs, "rmtree", shutil.rmtree):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert not torn.exists()
+    assert innocent.exists()  # another package's valid metadata survives
+    assert "Removed torn destination" in capsys.readouterr().out
+
+
+def test_unscannable_package_dir_fails_the_build(tmp_path: Path) -> None:
+    """A storage dir the cleanup cannot scan is not proof of cleanliness."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    real_iterdir = Path.iterdir
+
+    def broken_iterdir(self):
+        if self.name == "packages":
+            raise PermissionError("denied")
+        return real_iterdir(self)
+
+    with (
+        patch.object(Path, "iterdir", broken_iterdir),
+        pytest.raises(mod.CleanupError, match="cleanup failed"),
+    ):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+
+
+def test_stray_file_in_package_dir_is_ignored(tmp_path: Path) -> None:
+    """A plain file (or a pio-link) beside the packages is skipped by
+    pio's own scan and must never hard-fail the build."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    (tmp_path / "packages").mkdir(parents=True)
+    (tmp_path / "packages" / "stray.pio-link").write_text("x")
+    (tmp_path / "packages" / "no-metadata").mkdir()  # pio overwrites these
+    mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert (tmp_path / "packages" / "stray.pio-link").exists()
+    assert (tmp_path / "packages" / "no-metadata").exists()
+
+
+def test_unreadable_piopm_dir_is_removed(tmp_path: Path) -> None:
+    """A persistently corrupt .piopm under this spec's own name would
+    crash pio's storage scan; the dir is removed rather than left to
+    break the serial pass."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    torn = tmp_path / "packages" / "bad"
+    torn.mkdir(parents=True)
+    (torn / ".piopm").write_text("{not json")
+    mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert not torn.exists()
+
+
+def test_unreadable_piopm_under_other_name_survives(tmp_path: Path) -> None:
+    """A corrupt .piopm in another package's dir may be a worker mid-copy;
+    a failing spec must not remove a directory it does not own."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+    other = tmp_path / "packages" / "innocent"
+    other.mkdir(parents=True)
+    (other / ".piopm").write_text("{not json")
+    mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+    assert other.exists()
+
+
+def test_unexpected_cleanup_class_becomes_cleanup_error(tmp_path: Path) -> None:
+    """Cleanup failures of any class fail the build; nothing may be
+    downgraded to the serial fallback over a torn directory."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
+
+    with (
+        patch.object(
+            mod, "piopm_matches", MagicMock(side_effect=ValueError("bad spec"))
+        ),
+        pytest.raises(mod.CleanupError, match="cleanup failed"),
+    ):
+        mod.parallel_install(cls, ["esphome/bad @ 1.0"])
+
+
+def test_main_cleanup_error_fails_before_generic_fallback(tmp_path: Path) -> None:
+    """A CleanupError must escape main's serial fallback: the clause order
+    decides whether a stuck torn package fails the image build."""
+    mod = _load_script()
+    ini = tmp_path / "platformio.ini"
+    ini.write_text("[env:t]\nlib_deps =\n    esphome/x @ 1.0\n")
+    with (
+        patch.object(
+            mod, "parallel_install", side_effect=mod.CleanupError("stuck torn pkg")
+        ),
+        patch.object(mod.subprocess, "check_call"),
+        patch.object(sys, "argv", ["platformio_install_deps.py", str(ini), "-l"]),
+        pytest.raises(mod.CleanupError),
+    ):
+        mod.main()
+
+
+def test_main_generic_failure_still_runs_serial_pass(tmp_path: Path) -> None:
+    """A non-CleanupError wave failure prints, dumps the traceback, and
+    still reaches the authoritative serial pass with the pinned cwd."""
+    mod = _load_script()
+    ini = tmp_path / "platformio.ini"
+    ini.write_text("[env:t]\nlib_deps =\n    esphome/x @ 1.0\n")
+    with (
+        patch.object(mod, "parallel_install", side_effect=RuntimeError("boom")),
+        patch.object(mod.subprocess, "check_call") as mock_call,
+        patch.object(sys, "argv", ["platformio_install_deps.py", str(ini), "-l"]),
+    ):
+        mod.main()
+    mock_call.assert_called_once()
+    args, kwargs = mock_call.call_args
+    assert args[0][:4] == ["platformio", "pkg", "install", "-g"]
+    assert "esphome/x @ 1.0" in args[0]
+    assert kwargs["cwd"] == Path.cwd()
+
+
+def test_content_cache_creates_its_dir(tmp_path: Path, monkeypatch) -> None:
+    """The cold-cache hardening relies on ContentCache.__init__ creating
+    the namespace dir; pin the side effect, not mere callability."""
+    monkeypatch.setenv("PLATFORMIO_CACHE_DIR", str(tmp_path / "cache"))
+    ContentCache("http")
+    assert (tmp_path / "cache" / "http").is_dir()
+
+
+def test_piopm_matches_without_name_matches_nothing(tmp_path: Path) -> None:
+    """A spec with no derivable name can never match a directory."""
+    mod = _load_script()
+    assert mod.piopm_matches(str(tmp_path), "") == []
+
+
+def test_unresolvable_spec_stays_out_of_the_wave(tmp_path: Path, capsys) -> None:
+    """A spec with no derivable name is left to the serial pass; a raw
+    string key would break the one-per-destination dedupe."""
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path))
+    nameless = PackageSpec(requirements="^1.0")
+    mod.parallel_install(cls, [nameless])
+    assert cls.calls == []
+    assert "Skipping unresolvable spec" in capsys.readouterr().out
+
+
+def test_parallel_install_unlocks_when_pool_fails(tmp_path: Path) -> None:
+    mod = _load_script()
+    cls = _reset_fake(str(tmp_path))
+    with (
+        patch.object(mod, "ThreadPoolExecutor", side_effect=RuntimeError("no")),
+        pytest.raises(RuntimeError),
+    ):
+        mod.parallel_install(cls, ["esphome/a @ 1.0"])
+    assert cls.lock_events == ["lock", "unlock"]
+
+
+def test_parse_specs_unreadable_ini_fails_loudly(tmp_path: Path) -> None:
+    """A bad path must not silently build an image with no dependencies."""
+    mod = _load_script()
+    args = Namespace(libraries=True, platforms=False, tools=False)
+    with pytest.raises(SystemExit):
+        mod.parse_specs(str(tmp_path / "missing.ini"), args)
+
+
+def test_platformio_surface_for_install_deps_script() -> None:
+    """A PlatformIO bump that changes these members must fail here, not
+    silently turn the docker image's parallel preinstall into a no-op."""
+    # The script calls these positionally; pin the positions, not just
+    # membership, so a parameter reorder trips the wire too
+    params = inspect.signature(PackageManagerInstallMixin._install).parameters
+    assert list(params)[1] == "spec"
+    assert "skip_dependencies" in params
+    assert "compatibility" in params
+    for cls in (ToolPackageManager, LibraryPackageManager):
+        assert list(inspect.signature(cls.__init__).parameters)[1] == "package_dir"
+    for name in (
+        "lock",
+        "unlock",
+        "get_package",
+        "memcache_reset",
+        "get_pkg_dependencies",
+        "dependency_to_spec",
+        "get_download_dir",
+        "get_tmp_dir",
+    ):
+        assert callable(getattr(BasePackageManager, name))
+    # Losing any of these turns the wave into main()'s silent serial
+    # fallback: ensure_spec runs in the coordinator, the spec attributes
+    # feed the dedupe, cleanup, and dependency filters
+    assert callable(BasePackageManager.ensure_spec)
+    spec = PackageSpec("owner/name @ ^1.0")
+    assert spec.name == "name"
+    assert spec.owner == "owner"
+    assert spec.uri is None
+    assert spec.external is False
+    assert Version("1.5.0") in spec.requirements
+    # The failure-cleanup path degrades to a single line if these vanish
+    assert callable(fs.rmtree)
+    assert callable(fs.load_json)
+    # piopm_matches only tolerates a corrupt .piopm through this base;
+    # losing it would flip a wave failure from degrade to build failure
+    assert issubclass(InvalidJSONFile, ValueError)
+    assert PackageItem("pkg-dir").path == "pkg-dir"
+    assert callable(PackageCompatibility.from_dependency)

tests/script/test_setup.py

@@ -0,0 +1,562 @@
+"""Tests for script/setup.py."""
+
+import importlib.util
+import os
+from pathlib import Path, PurePosixPath, PureWindowsPath
+import runpy
+import sys
+from types import ModuleType
+from unittest.mock import Mock, call, patch
+
+import pytest
+
+_SCRIPT = Path(__file__).parents[2] / "script" / "setup.py"
+
+
+def _load_module() -> ModuleType:
+    spec = importlib.util.spec_from_file_location("script_setup", _SCRIPT)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+@pytest.fixture
+def script_setup() -> ModuleType:
+    """Fresh import of script/setup.py, isolated from other tests."""
+    return _load_module()
+
+
+# --- bin_dir / venv_python / activate_hint -----------------------------------
+
+
+def test_bin_dir_matches_host_layout(script_setup: ModuleType, tmp_path: Path) -> None:
+    """The venv scheme resolves to Scripts on Windows and bin everywhere else."""
+    expected = "Scripts" if os.name == "nt" else "bin"
+    assert script_setup.bin_dir(tmp_path) == tmp_path / expected
+
+
+# Both flavours are exercised on every host. Pure paths are used because a real
+# Path refuses to change flavour: PosixPath cannot be built on Windows, and
+# WindowsPath cannot be built on Unix.
+
+
+def test_venv_python_posix(script_setup: ModuleType, tmp_path: Path) -> None:
+    with (
+        patch.object(
+            script_setup, "bin_dir", return_value=PurePosixPath("/x/venv/bin")
+        ),
+        patch.object(script_setup.os, "name", "posix"),
+    ):
+        result = script_setup.venv_python(tmp_path)
+    assert result == PurePosixPath("/x/venv/bin/python")
+
+
+def test_venv_python_nt(script_setup: ModuleType, tmp_path: Path) -> None:
+    with (
+        patch.object(
+            script_setup, "bin_dir", return_value=PureWindowsPath(r"C:\x\venv\Scripts")
+        ),
+        patch.object(script_setup.os, "name", "nt"),
+    ):
+        result = script_setup.venv_python(tmp_path)
+    assert result == PureWindowsPath(r"C:\x\venv\Scripts\python.exe")
+
+
+def test_activate_hint_posix(script_setup: ModuleType) -> None:
+    with (
+        patch.object(script_setup, "ROOT", PurePosixPath("/x")),
+        patch.object(
+            script_setup, "bin_dir", return_value=PurePosixPath("/x/venv/bin")
+        ),
+        patch.object(script_setup.os, "name", "posix"),
+    ):
+        hint = script_setup.activate_hint()
+    assert hint == "source venv/bin/activate"
+
+
+def test_activate_hint_nt(script_setup: ModuleType) -> None:
+    with (
+        patch.object(script_setup, "ROOT", PureWindowsPath(r"C:\x")),
+        patch.object(
+            script_setup, "bin_dir", return_value=PureWindowsPath(r"C:\x\venv\Scripts")
+        ),
+        patch.object(script_setup.os, "name", "nt"),
+    ):
+        hint = script_setup.activate_hint()
+    # The nt branch returns str(activate) as-is, skipping the "source " prefix.
+    assert hint == r"venv\Scripts\activate"
+
+
+# --- run -----------------------------------------------------------------
+
+
+def test_run_success(script_setup: ModuleType) -> None:
+    with patch.object(
+        script_setup.subprocess, "run", return_value=Mock(returncode=0)
+    ) as mock_run:
+        script_setup.run(["echo", "hi"])
+    mock_run.assert_called_once_with(
+        ["echo", "hi"], cwd=script_setup.ROOT, env=None, check=False
+    )
+
+
+def test_run_failure_raises_system_exit_with_code(
+    script_setup: ModuleType, capsys: pytest.CaptureFixture[str]
+) -> None:
+    with (
+        patch.object(script_setup.subprocess, "run", return_value=Mock(returncode=7)),
+        pytest.raises(SystemExit) as excinfo,
+    ):
+        script_setup.run(["false"])
+    assert excinfo.value.code == 7
+    assert "Failed with exit code 7: false" in capsys.readouterr().err
+
+
+# --- git_output ------------------------------------------------------------
+
+
+def test_git_output_success_strips_stdout(script_setup: ModuleType) -> None:
+    with patch.object(
+        script_setup.subprocess,
+        "run",
+        return_value=Mock(returncode=0, stdout="  /repo/.git \n"),
+    ) as mock_run:
+        result = script_setup.git_output("rev-parse", "--absolute-git-dir")
+    assert result == "/repo/.git"
+    mock_run.assert_called_once_with(
+        ["git", "rev-parse", "--absolute-git-dir"],
+        cwd=script_setup.ROOT,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+
+def test_git_output_nonzero_returncode_is_empty(script_setup: ModuleType) -> None:
+    with patch.object(
+        script_setup.subprocess,
+        "run",
+        return_value=Mock(returncode=1, stdout="whatever"),
+    ):
+        assert script_setup.git_output("status") == ""
+
+
+def test_git_output_oserror_is_empty(script_setup: ModuleType) -> None:
+    with patch.object(script_setup.subprocess, "run", side_effect=OSError("no git")):
+        assert script_setup.git_output("status") == ""
+
+
+# --- create_venv -----------------------------------------------------------
+
+
+def test_create_venv_uses_uv_when_present(
+    script_setup: ModuleType, tmp_path: Path
+) -> None:
+    venv = tmp_path / "venv"
+    with (
+        patch.object(script_setup.shutil, "which", return_value="/usr/bin/uv"),
+        patch.object(
+            script_setup.subprocess, "run", return_value=Mock(returncode=0)
+        ) as mock_run,
+    ):
+        script_setup.create_venv(venv)
+    mock_run.assert_called_once_with(
+        ["/usr/bin/uv", "venv", "--clear", "--seed", str(venv)],
+        cwd=script_setup.ROOT,
+        env=None,
+        check=False,
+    )
+
+
+def test_create_venv_falls_back_to_venv_module(
+    script_setup: ModuleType, tmp_path: Path
+) -> None:
+    venv = tmp_path / "venv"
+    with (
+        patch.object(script_setup.shutil, "which", return_value=None),
+        patch.object(
+            script_setup.subprocess, "run", return_value=Mock(returncode=0)
+        ) as mock_run,
+    ):
+        script_setup.create_venv(venv)
+    mock_run.assert_called_once_with(
+        [sys.executable, "-m", "venv", "--clear", str(venv)],
+        cwd=script_setup.ROOT,
+        env=None,
+        check=False,
+    )
+
+
+# --- venv_environment --------------------------------------------------------
+
+
+def test_venv_environment_sets_virtual_env_and_prepends_path(
+    script_setup: ModuleType, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    venv = tmp_path / "venv"
+    monkeypatch.setenv("PYTHONHOME", "/somewhere")
+    monkeypatch.setenv("PATH", "/usr/bin:/bin")
+    env = script_setup.venv_environment(venv)
+    assert env["VIRTUAL_ENV"] == str(venv)
+    assert "PYTHONHOME" not in env
+    expected_prefix = str(script_setup.bin_dir(venv)) + os.pathsep
+    assert env["PATH"] == expected_prefix + "/usr/bin:/bin"
+
+
+def test_venv_environment_path_fallback_when_unset(
+    script_setup: ModuleType, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    venv = tmp_path / "venv"
+    monkeypatch.delenv("PATH", raising=False)
+    env = script_setup.venv_environment(venv)
+    # No trailing separator: an empty PATH entry means "search the cwd".
+    assert env["PATH"] == str(script_setup.bin_dir(venv))
+
+
+# --- find_uv -----------------------------------------------------------------
+
+
+def test_find_uv_found_immediately(script_setup: ModuleType, tmp_path: Path) -> None:
+    venv = tmp_path / "venv"
+    env = {"PATH": "/usr/bin"}
+    with (
+        patch.object(script_setup.shutil, "which", return_value="/usr/bin/uv"),
+        patch.object(script_setup.subprocess, "run") as mock_run,
+    ):
+        result = script_setup.find_uv(venv, env)
+    assert result == "/usr/bin/uv"
+    mock_run.assert_not_called()
+
+
+def test_find_uv_installed_then_found(script_setup: ModuleType, tmp_path: Path) -> None:
+    venv = tmp_path / "venv"
+    env = {"PATH": "/usr/bin"}
+    with (
+        patch.object(script_setup.shutil, "which", side_effect=[None, "/usr/bin/uv"]),
+        patch.object(
+            script_setup.subprocess, "run", return_value=Mock(returncode=0)
+        ) as mock_run,
+    ):
+        result = script_setup.find_uv(venv, env)
+    assert result == "/usr/bin/uv"
+    mock_run.assert_called_once_with(
+        [str(script_setup.venv_python(venv)), "-m", "pip", "install", "uv"],
+        cwd=script_setup.ROOT,
+        env=env,
+        check=False,
+    )
+
+
+def test_find_uv_still_missing_raises_system_exit(
+    script_setup: ModuleType, tmp_path: Path
+) -> None:
+    venv = tmp_path / "venv"
+    env = {"PATH": "/usr/bin"}
+    with (
+        patch.object(script_setup.shutil, "which", side_effect=[None, None]),
+        patch.object(script_setup.subprocess, "run", return_value=Mock(returncode=0)),
+        pytest.raises(SystemExit, match="uv could not be installed"),
+    ):
+        script_setup.find_uv(venv, env)
+
+
+# --- install_dependencies -----------------------------------------------------
+
+
+def test_install_dependencies_installs_setuptools_then_project(
+    script_setup: ModuleType, tmp_path: Path
+) -> None:
+    venv = tmp_path / "venv"
+    env = {"PATH": "/usr/bin"}
+    with (
+        patch.object(script_setup.shutil, "which", return_value="/usr/bin/uv"),
+        patch.object(
+            script_setup.subprocess, "run", return_value=Mock(returncode=0)
+        ) as mock_run,
+    ):
+        script_setup.install_dependencies(venv, env)
+    assert mock_run.call_args_list == [
+        call(
+            ["/usr/bin/uv", "pip", "install", "setuptools", "wheel"],
+            cwd=script_setup.ROOT,
+            env=env,
+            check=False,
+        ),
+        call(
+            [
+                "/usr/bin/uv",
+                "pip",
+                "install",
+                "-e",
+                ".[dev,test]",
+                "--config-settings",
+                "editable_mode=compat",
+            ],
+            cwd=script_setup.ROOT,
+            env=env,
+            check=False,
+        ),
+    ]
+
+
+# --- install_git_hooks ---------------------------------------------------------
+
+
+def _fake_git_output(git_dir: str, common_dir: str):
+    def _run(*args: str) -> str:
+        if "--absolute-git-dir" in args:
+            return git_dir
+        return common_dir
+
+    return _run
+
+
+def test_install_git_hooks_returns_early_when_git_dir_empty(
+    script_setup: ModuleType,
+) -> None:
+    env = {"PATH": "/usr/bin"}
+    with (
+        patch.object(
+            script_setup, "git_output", side_effect=_fake_git_output("", "/repo/.git")
+        ),
+        patch.object(script_setup.subprocess, "run") as mock_run,
+    ):
+        script_setup.install_git_hooks(env)
+    mock_run.assert_not_called()
+
+
+def test_install_git_hooks_returns_early_when_common_dir_empty(
+    script_setup: ModuleType,
+) -> None:
+    env = {"PATH": "/usr/bin"}
+    with (
+        patch.object(
+            script_setup, "git_output", side_effect=_fake_git_output("/repo/.git", "")
+        ),
+        patch.object(script_setup.subprocess, "run") as mock_run,
+    ):
+        script_setup.install_git_hooks(env)
+    mock_run.assert_not_called()
+
+
+def test_install_git_hooks_returns_early_for_worktree(
+    script_setup: ModuleType,
+) -> None:
+    """A worktree's git-dir differs from the shared common-dir."""
+    env = {"PATH": "/usr/bin"}
+    with (
+        patch.object(
+            script_setup,
+            "git_output",
+            side_effect=_fake_git_output("/repo/.git/worktrees/wt", "/repo/.git"),
+        ),
+        patch.object(script_setup.subprocess, "run") as mock_run,
+    ):
+        script_setup.install_git_hooks(env)
+    mock_run.assert_not_called()
+
+
+def test_install_git_hooks_missing_prek_raises_system_exit(
+    script_setup: ModuleType,
+) -> None:
+    env = {"PATH": "/usr/bin"}
+    with (
+        patch.object(
+            script_setup,
+            "git_output",
+            side_effect=_fake_git_output("/repo/.git", "/repo/.git"),
+        ),
+        patch.object(script_setup.shutil, "which", return_value=None),
+        patch.object(script_setup.subprocess, "run") as mock_run,
+        pytest.raises(SystemExit, match="prek was not installed"),
+    ):
+        script_setup.install_git_hooks(env)
+    mock_run.assert_not_called()
+
+
+def test_install_git_hooks_happy_path_installs_hook(
+    script_setup: ModuleType, tmp_path: Path
+) -> None:
+    env = {"PATH": "/usr/bin"}
+    common_dir = tmp_path / "repo" / ".git"
+    hooks_dir = common_dir / "hooks"
+    hooks_dir.mkdir(parents=True)
+    source_hook = tmp_path / "post-checkout"
+    source_hook.write_text("#!/bin/sh\necho post-checkout\n")
+
+    with (
+        patch.object(script_setup, "POST_CHECKOUT_HOOK", source_hook),
+        patch.object(
+            script_setup,
+            "git_output",
+            side_effect=_fake_git_output(str(common_dir), str(common_dir)),
+        ),
+        patch.object(script_setup.shutil, "which", return_value="/usr/bin/prek"),
+        patch.object(
+            script_setup.subprocess, "run", return_value=Mock(returncode=0)
+        ) as mock_run,
+    ):
+        script_setup.install_git_hooks(env)
+
+    mock_run.assert_called_once_with(
+        ["/usr/bin/prek", "install", "--overwrite"],
+        cwd=script_setup.ROOT,
+        env=env,
+        check=False,
+    )
+    installed = hooks_dir / "post-checkout"
+    assert installed.read_text() == source_hook.read_text()
+    if os.name != "nt":
+        # Windows has no POSIX permission bits for chmod to set.
+        assert (installed.stat().st_mode & 0o777) == 0o755
+
+
+def test_install_git_hooks_skips_copy_when_hooks_dir_missing(
+    script_setup: ModuleType, tmp_path: Path
+) -> None:
+    """The prek install still runs when the hooks directory does not exist."""
+    env = {"PATH": "/usr/bin"}
+    common_dir = tmp_path / "repo" / ".git"
+    common_dir.mkdir(parents=True)  # no "hooks" subdirectory created
+
+    with (
+        patch.object(
+            script_setup,
+            "git_output",
+            side_effect=_fake_git_output(str(common_dir), str(common_dir)),
+        ),
+        patch.object(script_setup.shutil, "which", return_value="/usr/bin/prek"),
+        patch.object(
+            script_setup.subprocess, "run", return_value=Mock(returncode=0)
+        ) as mock_run,
+    ):
+        script_setup.install_git_hooks(env)
+
+    mock_run.assert_called_once()
+    assert not (common_dir / "hooks").exists()
+
+
+# --- report ------------------------------------------------------------------
+
+
+def test_report_active_state(
+    script_setup: ModuleType, capsys: pytest.CaptureFixture
+) -> None:
+    venv = Path("/opt/esphome-venv")
+    script_setup.report(script_setup.VENV_ACTIVE, venv)
+    out = capsys.readouterr().out
+    assert "Dependencies installed into the active virtual environment:" in out
+    assert str(venv) in out
+    assert "is already active in this shell" in out
+
+
+def test_report_reused_state(
+    script_setup: ModuleType, capsys: pytest.CaptureFixture
+) -> None:
+    script_setup.report(script_setup.VENV_REUSED, script_setup.DEFAULT_VENV)
+    out = capsys.readouterr().out
+    assert "Dependencies updated in the existing ./venv" in out
+
+
+def test_report_created_state(
+    script_setup: ModuleType, capsys: pytest.CaptureFixture
+) -> None:
+    script_setup.report(script_setup.VENV_CREATED, script_setup.DEFAULT_VENV)
+    out = capsys.readouterr().out
+    assert "Virtual environment created at ./venv" in out
+
+
+# --- main --------------------------------------------------------------------
+
+
+def test_main_raises_system_exit_when_python_too_old(
+    script_setup: ModuleType,
+) -> None:
+    with (
+        patch.object(script_setup.sys, "version_info", (3, 11, 5)),
+        pytest.raises(SystemExit, match="ESPHome needs Python 3.12"),
+    ):
+        script_setup.main()
+
+
+def test_main_uses_active_virtual_env(
+    script_setup: ModuleType, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    active_venv = tmp_path / "active-venv"
+    monkeypatch.setenv("VIRTUAL_ENV", str(active_venv))
+    with (
+        patch.object(script_setup, "ROOT", tmp_path),
+        patch.object(script_setup, "create_venv") as mock_create_venv,
+        patch.object(script_setup, "install_dependencies") as mock_install_deps,
+        patch.object(script_setup, "install_git_hooks") as mock_install_hooks,
+        patch.object(script_setup, "report") as mock_report,
+    ):
+        script_setup.main()
+    mock_create_venv.assert_not_called()
+    mock_install_deps.assert_called_once()
+    mock_install_hooks.assert_called_once()
+    mock_report.assert_called_once_with(script_setup.VENV_ACTIVE, active_venv)
+    assert (tmp_path / ".temp").is_dir()
+
+
+def test_main_reuses_existing_venv(
+    script_setup: ModuleType, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.delenv("VIRTUAL_ENV", raising=False)
+    default_venv = tmp_path / "venv"
+    python_path = script_setup.venv_python(default_venv)
+    python_path.parent.mkdir(parents=True)
+    python_path.touch()
+
+    with (
+        patch.object(script_setup, "ROOT", tmp_path),
+        patch.object(script_setup, "DEFAULT_VENV", default_venv),
+        patch.object(script_setup, "create_venv") as mock_create_venv,
+        patch.object(script_setup, "install_dependencies") as mock_install_deps,
+        patch.object(script_setup, "install_git_hooks") as mock_install_hooks,
+        patch.object(script_setup, "report") as mock_report,
+    ):
+        script_setup.main()
+    mock_create_venv.assert_not_called()
+    mock_install_deps.assert_called_once()
+    mock_install_hooks.assert_called_once()
+    mock_report.assert_called_once_with(script_setup.VENV_REUSED, default_venv)
+    assert (tmp_path / ".temp").is_dir()
+
+
+def test_main_creates_new_venv(
+    script_setup: ModuleType, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.delenv("VIRTUAL_ENV", raising=False)
+    default_venv = tmp_path / "venv"  # does not exist yet
+
+    with (
+        patch.object(script_setup, "ROOT", tmp_path),
+        patch.object(script_setup, "DEFAULT_VENV", default_venv),
+        patch.object(script_setup, "create_venv") as mock_create_venv,
+        patch.object(script_setup, "install_dependencies") as mock_install_deps,
+        patch.object(script_setup, "install_git_hooks") as mock_install_hooks,
+        patch.object(script_setup, "report") as mock_report,
+    ):
+        script_setup.main()
+    mock_create_venv.assert_called_once_with(default_venv)
+    mock_install_deps.assert_called_once()
+    mock_install_hooks.assert_called_once()
+    mock_report.assert_called_once_with(script_setup.VENV_CREATED, default_venv)
+    assert (tmp_path / ".temp").is_dir()
+
+
+def test_run_as_script_calls_main(tmp_path: Path) -> None:
+    """The __main__ guard runs the whole flow, with every side effect stubbed."""
+    completed = Mock(returncode=0, stdout="")
+    with (
+        patch("subprocess.run", return_value=completed) as mock_run,
+        patch("shutil.which", return_value="/usr/bin/uv"),
+        patch("pathlib.Path.mkdir") as mock_mkdir,
+        patch.dict(os.environ, {"VIRTUAL_ENV": str(tmp_path / "env")}),
+    ):
+        runpy.run_path(str(_SCRIPT), run_name="__main__")
+
+    # The dependency install ran, and git reported no hooks directory to touch.
+    assert mock_run.called
+    mock_mkdir.assert_called_once_with(exist_ok=True)

tests/script/test_sync_dependency_versions.py

@@ -0,0 +1,219 @@
+"""Unit tests for script/sync_dependency_versions.py."""
+
+from pathlib import Path
+import subprocess
+import sys
+
+import pytest
+import yamlrocks
+
+sys.path.insert(0, str((Path(__file__).parent / ".." / ".." / "script").resolve()))
+
+import sync_dependency_versions as sync_mod  # noqa: E402
+
+PRECOMMIT = """\
+# See https://pre-commit.com for more information
+repos:
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    # Ruff version.
+    rev: v0.1.0
+    hooks:
+      - id: ruff
+  - repo: https://github.com/PyCQA/flake8
+    rev: 7.0.0
+    hooks:
+      - id: flake8
+  - repo: https://github.com/asottile/pyupgrade
+    rev: v3.0.0
+    hooks:
+      - id: pyupgrade
+  - repo: https://github.com/pre-commit/mirrors-clang-format
+    rev: v13.0.1
+    hooks:
+      - id: clang-format
+  - repo: https://github.com/adrienverge/yamllint.git
+    rev: v1.0.0
+    hooks:
+      - id: yamllint
+  - repo: local
+    hooks:
+      - id: pylint
+"""
+
+REQ_TEST = """\
+pylint==4.0.8
+flake8==7.1.0
+ruff==0.2.0  # comment
+pyupgrade==3.0.0
+"""
+
+REQ_DEV = """\
+clang-format==13.0.1
+yamllint==1.0.0
+"""
+
+RUFF_REPO = "https://github.com/astral-sh/ruff-pre-commit"
+DUPLICATE_RUFF_BLOCK = f"  - repo: {RUFF_REPO}\n    rev: v0.3.0\n    hooks: []\n"
+
+EXPECTED_DRIFT = ["ruff: 0.1.0 -> 0.2.0", "flake8: 7.0.0 -> 7.1.0"]
+EXPECTED_PRECOMMIT = PRECOMMIT.replace("rev: v0.1.0", "rev: v0.2.0").replace(
+    "rev: 7.0.0", "rev: 7.1.0"
+)
+
+
+@pytest.fixture
+def root(tmp_path: Path) -> Path:
+    """A fake checkout where ruff (v-prefixed) and flake8 (bare) have drifted."""
+    (tmp_path / ".pre-commit-config.yaml").write_text(PRECOMMIT)
+    (tmp_path / "requirements_test.txt").write_text(REQ_TEST)
+    (tmp_path / "requirements_dev.txt").write_text(REQ_DEV)
+    return tmp_path
+
+
+def _load(text: str) -> object:
+    return yamlrocks.loads(text.encode(), option=yamlrocks.OPT_ROUND_TRIP)
+
+
+@pytest.mark.parametrize(
+    ("requirements", "expected"),
+    [
+        ("prek==0.5.1  # comment\n", "0.5.1"),
+        ("Prek==0.5.1\n", "0.5.1"),
+        ("other==1.0\nprek==0.5.1\n", "0.5.1"),
+        ("prek>=0.5.1\n", None),
+        ("prek-extra==0.5.1\n", None),
+        ("", None),
+    ],
+)
+def test_read_requirement_version(requirements: str, expected: str | None) -> None:
+    assert sync_mod.read_requirement_version(requirements, "prek") == expected
+
+
+def test_find_repo_entry() -> None:
+    entry = sync_mod.find_repo_entry(_load(PRECOMMIT), RUFF_REPO)
+    assert entry["rev"] == "v0.1.0"
+
+
+@pytest.mark.parametrize(
+    ("text", "message"),
+    [
+        ("hooks: []\n", "missing key 'repos'"),
+        ("repos:\n  - rev: 1.0.0\n", "missing key 'repo'"),
+        (PRECOMMIT + DUPLICATE_RUFF_BLOCK, "found 2"),
+        ("repos:\n  - repo: other\n    rev: 1.0.0\n", "found 0"),
+    ],
+)
+def test_find_repo_entry_errors(text: str, message: str) -> None:
+    with pytest.raises(sync_mod.SyncError, match=message):
+        sync_mod.find_repo_entry(_load(text), RUFF_REPO)
+
+
+@pytest.mark.parametrize(
+    ("rev", "expected"),
+    [("v0.1.0", ("v", "0.1.0")), ("7.0.0", ("", "7.0.0")), ("'1.0'", ("", "1.0"))],
+)
+def test_current_rev(rev: str, expected: tuple[str, str]) -> None:
+    doc = _load(f"repos:\n  - repo: {RUFF_REPO}\n    rev: {rev}\n")
+    assert sync_mod.current_rev(doc["repos"][0], RUFF_REPO) == expected
+
+
+@pytest.mark.parametrize(
+    ("block", "message"),
+    [("    hooks: []\n", "has no rev"), ("    rev: 1.0\n", "not a string: 1.0")],
+)
+def test_current_rev_errors(block: str, message: str) -> None:
+    doc = _load(f"repos:\n  - repo: {RUFF_REPO}\n{block}")
+    with pytest.raises(sync_mod.SyncError, match=message):
+        sync_mod.current_rev(doc["repos"][0], RUFF_REPO)
+
+
+def test_sync_reports_without_writing(root: Path) -> None:
+    assert sync_mod.sync(root, write=False) == EXPECTED_DRIFT
+    assert (root / ".pre-commit-config.yaml").read_text() == PRECOMMIT
+
+
+def test_sync_writes_keeps_layout_and_is_idempotent(root: Path) -> None:
+    assert sync_mod.sync(root, write=True) == EXPECTED_DRIFT
+    assert (root / ".pre-commit-config.yaml").read_text() == EXPECTED_PRECOMMIT
+    assert sync_mod.sync(root, write=True) == []
+
+
+def test_sync_does_not_touch_a_config_that_matches(root: Path) -> None:
+    (root / ".pre-commit-config.yaml").write_text(EXPECTED_PRECOMMIT)
+    before = (root / ".pre-commit-config.yaml").stat().st_mtime_ns
+    assert sync_mod.sync(root, write=True) == []
+    assert (root / ".pre-commit-config.yaml").stat().st_mtime_ns == before
+
+
+def test_sync_missing_requirement_pin(root: Path) -> None:
+    (root / "requirements_dev.txt").write_text("")
+    with pytest.raises(sync_mod.SyncError, match="no 'clang-format==' pin"):
+        sync_mod.sync(root, write=True)
+
+
+def test_sync_propagates_config_errors(root: Path) -> None:
+    (root / ".pre-commit-config.yaml").write_text(PRECOMMIT + DUPLICATE_RUFF_BLOCK)
+    with pytest.raises(sync_mod.SyncError, match="found 2"):
+        sync_mod.sync(root, write=True)
+
+
+def test_main_check_reports_drift(
+    root: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    assert sync_mod.main(["--check", "--root", str(root)]) == 1
+    assert capsys.readouterr().out.splitlines() == EXPECTED_DRIFT
+    assert (root / ".pre-commit-config.yaml").read_text() == PRECOMMIT
+
+
+def test_main_writes_then_check_is_clean(
+    root: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    assert sync_mod.main(["--root", str(root)]) == 0
+    assert capsys.readouterr().out.splitlines() == EXPECTED_DRIFT
+    assert sync_mod.main(["--check", "--root", str(root)]) == 0
+    assert capsys.readouterr().out == ""
+
+
+def test_main_reports_sync_error(
+    root: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    (root / "requirements_dev.txt").write_text("")
+    assert sync_mod.main(["--root", str(root)]) == 1
+    assert (
+        "error: requirements_dev.txt: no 'clang-format==' pin"
+        in capsys.readouterr().err
+    )
+
+
+def test_main_defaults_to_repo_root(monkeypatch: pytest.MonkeyPatch) -> None:
+    seen: dict[str, object] = {}
+
+    def fake_sync(root: Path, *, write: bool) -> list[str]:
+        seen["root"] = root
+        seen["write"] = write
+        return []
+
+    monkeypatch.setattr(sync_mod, "sync", fake_sync)
+    assert sync_mod.main([]) == 0
+    assert seen == {"root": sync_mod.REPO_ROOT, "write": True}
+
+
+def test_repository_is_in_sync() -> None:
+    """The real checkout must match; a failure here means a rev has drifted.
+
+    Also proves every SYNC_TARGETS entry still resolves in the real files.
+    """
+    assert sync_mod.sync(sync_mod.REPO_ROOT, write=False) == []
+
+
+def test_cli_entry_point(root: Path) -> None:
+    """Run the script the way the workflow does, as a subprocess."""
+    script = Path(sync_mod.__file__)
+    result = subprocess.run(
+        [sys.executable, str(script), "--check", "--root", str(root)],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert result.returncode == 1
+    assert result.stdout.splitlines() == EXPECTED_DRIFT

tests/script/test_update_integration_test_durations.py

@@ -0,0 +1,130 @@
+"""Unit tests for script/update_integration_test_durations.py."""
+
+import json
+from pathlib import Path
+import sys
+from unittest.mock import patch
+
+import pytest
+
+# Add the script directory to Python path so we can import the module
+script_dir = str((Path(__file__).parent / ".." / ".." / "script").resolve())
+sys.path.insert(0, script_dir)
+
+import helpers  # noqa: E402
+import update_integration_test_durations as uitd  # noqa: E402
+
+JUNIT_TEMPLATE = """<?xml version="1.0" encoding="utf-8"?>
+<testsuites><testsuite>{testcases}</testsuite></testsuites>
+"""
+
+KNOWN = {
+    "tests/integration/test_a.py",
+    "tests/integration/test_b.py",
+}
+
+
+def _write_junit(path: Path, testcases: str) -> None:
+    path.write_text(JUNIT_TEMPLATE.format(testcases=testcases), encoding="utf-8")
+
+
+def test_collect_durations_sums_per_file(tmp_path: Path) -> None:
+    """Testcases from the same module sum."""
+    _write_junit(
+        tmp_path / "a.xml",
+        '<testcase classname="tests.integration.test_a" name="t1" time="1.5"/>'
+        '<testcase classname="tests.integration.test_a" name="t2" time="2.0"/>'
+        '<testcase classname="tests.integration.test_b" name="t1" time="4.0"/>',
+    )
+    assert uitd.collect_durations(tmp_path, KNOWN) == {
+        "tests/integration/test_a.py": 3.5,
+        "tests/integration/test_b.py": 4.0,
+    }
+
+
+def test_collect_durations_class_based_testcase(tmp_path: Path) -> None:
+    """A class-based classname still maps to its module file."""
+    _write_junit(
+        tmp_path / "a.xml",
+        '<testcase classname="tests.integration.test_a.TestFoo" name="t" time="2.5"/>',
+    )
+    assert uitd.collect_durations(tmp_path, KNOWN) == {
+        "tests/integration/test_a.py": 2.5
+    }
+
+
+def test_collect_durations_unknown_module_skipped(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    """A classname that maps to no known file is skipped with a warning."""
+    _write_junit(
+        tmp_path / "a.xml",
+        '<testcase classname="tests.integration.test_gone" name="t" time="2.5"/>',
+    )
+    assert uitd.collect_durations(tmp_path, KNOWN) == {}
+    assert "test_gone" in capsys.readouterr().err
+
+
+def test_collect_durations_skips_skipped_testcases(tmp_path: Path) -> None:
+    """Skipped testcases do not record a bogus zero duration."""
+    _write_junit(
+        tmp_path / "a.xml",
+        '<testcase classname="tests.integration.test_a" name="t" time="0">'
+        "<skipped/></testcase>",
+    )
+    assert uitd.collect_durations(tmp_path, KNOWN) == {}
+
+
+def test_collect_durations_unexpected_classname_aborts(tmp_path: Path) -> None:
+    """A classname outside tests.integration means the junit layout changed."""
+    _write_junit(
+        tmp_path / "a.xml",
+        '<testcase classname="tests.unit_tests.test_x" name="t" time="9.0"/>',
+    )
+    with pytest.raises(SystemExit):
+        uitd.collect_durations(tmp_path, KNOWN)
+
+
+def test_collect_durations_empty_dir_aborts(tmp_path: Path) -> None:
+    """No junit XML at all is a hard error, not an empty recording."""
+    with pytest.raises(SystemExit):
+        uitd.collect_durations(tmp_path, KNOWN)
+
+
+def test_main_merges_partial_run(tmp_path: Path) -> None:
+    """A partial run merges over the previous data instead of truncating it."""
+    tests_dir = tmp_path / "tests" / "integration"
+    tests_dir.mkdir(parents=True)
+    for name in ("test_a", "test_b", "test_c"):
+        (tests_dir / f"{name}.py").write_text("", encoding="utf-8")
+    durations_file = tmp_path / helpers.INTEGRATION_TEST_DURATIONS_FILE
+    durations_file.write_text(
+        json.dumps(
+            {
+                "tests/integration/test_a.py": 5.0,
+                "tests/integration/test_b.py": 7.0,
+                "tests/integration/test_gone.py": 9.0,
+            }
+        ),
+        encoding="utf-8",
+    )
+    junit_dir = tmp_path / "junit"
+    junit_dir.mkdir()
+    _write_junit(
+        junit_dir / "a.xml",
+        '<testcase classname="tests.integration.test_a" name="t" time="6.0"/>',
+    )
+    with (
+        patch.object(helpers, "root_path", str(tmp_path)),
+        patch.object(uitd, "DURATIONS_FILE", durations_file),
+    ):
+        # 1 of 3 files covered: refused without --allow-partial
+        with patch.object(sys, "argv", ["uitd", str(junit_dir)]):
+            assert uitd.main() == uitd.EXIT_LOW_COVERAGE
+        with patch.object(sys, "argv", ["uitd", str(junit_dir), "--allow-partial"]):
+            assert uitd.main() == 0
+    # test_a updated, test_b kept, deleted test_gone dropped
+    assert json.loads(durations_file.read_text()) == {
+        "tests/integration/test_a.py": 6.0,
+        "tests/integration/test_b.py": 7.0,
+    }