[ci] Balance integration test buckets by recorded durations (#18895)

.github/workflows/ci.yml

@@ -92,6 +92,7 @@ jobs:
     outputs:
       core-ci: ${{ steps.determine.outputs.core-ci }}
       integration-tests: ${{ steps.determine.outputs.integration-tests }}
+      integration-run-all: ${{ steps.determine.outputs.integration-run-all }}
       integration-test-buckets: ${{ steps.determine.outputs.integration-test-buckets }}
       clang-tidy: ${{ steps.determine.outputs.clang-tidy }}
       clang-tidy-mode: ${{ steps.determine.outputs.clang-tidy-mode }}
@@ -152,6 +153,9 @@ jobs:
           # Extract individual fields
           echo "core-ci=$(echo "$output" | jq -r '.core_ci')" >> $GITHUB_OUTPUT
           echo "integration-tests=$(echo "$output" | jq -r '.integration_tests')" >> $GITHUB_OUTPUT
+          # A missing key must fail here, not silently disable the junit upload
+          run_all=$(echo "$output" | jq -r 'if has("integration_run_all") then .integration_run_all else error("integration_run_all missing") end')
+          echo "integration-run-all=${run_all}" >> $GITHUB_OUTPUT
           echo "integration-test-buckets=$(echo "$output" | jq -c '.integration_test_buckets')" >> $GITHUB_OUTPUT
           echo "clang-tidy=$(echo "$output" | jq -r '.clang_tidy')" >> $GITHUB_OUTPUT
           echo "clang-tidy-mode=$(echo "$output" | jq -r '.clang_tidy_mode')" >> $GITHUB_OUTPUT
@@ -427,8 +431,25 @@ jobs:
         run: |
           . venv/bin/activate
           mapfile -t test_files < <(echo "$BUCKET_TESTS" | jq -r '.[]')
+          if [ "${#test_files[@]}" -eq 0 ]; then
+            echo "::error::Empty integration test bucket; pytest would collect the whole tree"
+            exit 1
+          fi
           echo "Bucket ${{ matrix.bucket.name }}: running ${#test_files[@]} integration tests"
-          pytest -vv --no-cov --tb=native --durations=30 -n auto "${test_files[@]}"
+          pytest -vv --no-cov --tb=native --durations=30 -n auto --dist worksteal \
+            --junitxml=junit-integration.xml "${test_files[@]}"
+      - name: Upload junit timings
+        # Consumed by sync-integration-durations.yml through
+        # script/update_integration_test_durations.py; only full matrix dev
+        # runs produce usable data.
+        if: github.ref == 'refs/heads/dev' && needs.determine-jobs.outputs.integration-run-all == 'true'
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: junit-integration-${{ strategy.job-index }}
+          path: junit-integration.xml
+          if-no-files-found: error
+          # A full cron period of margin for the weekly refresh
+          retention-days: 14
       - name: Print ccache statistics
         # esphome stores the PlatformIO ccache under the machine-global cache
         # dir (see _ccache_env() in esphome/platformio/toolchain.py).

.github/workflows/sync-integration-durations.yml

@@ -0,0 +1,98 @@
+---
+name: Refresh integration test durations
+
+on:
+  workflow_dispatch:
+  schedule:
+    - cron: "45 5 * * 1"
+
+# Repo writes (branch push, PR open) happen via the App token minted below,
+# so the workflow's GITHUB_TOKEN does not need any write scopes.
+permissions:
+  contents: read
+  actions: read  # gh api / gh run download for the CI junit artifacts
+
+jobs:
+  sync:
+    name: Refresh integration test durations
+    runs-on: ubuntu-latest
+    if: github.repository == 'esphome/esphome'
+    steps:
+      - name: Generate a token
+        id: generate-token
+        uses: actions/create-github-app-token@bcd2ba49218906704ab6c1aa796996da409d3eb1 # v3.2.0
+        with:
+          client-id: ${{ vars.ESPHOME_GITHUB_APP_CLIENT_ID }}
+          private-key: ${{ secrets.ESPHOME_GITHUB_APP_PRIVATE_KEY }}
+          permission-contents: write       # push the sync branch
+          permission-pull-requests: write  # open or refresh the sync PR
+
+      - name: Checkout
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+
+      - name: Setup Python
+        uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
+        with:
+          python-version: "3.13"
+
+      - name: Refresh from the newest usable dev run
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: |
+          # Only full matrix dev runs upload junit-integration-* artifacts
+          # (see the integration-tests job); the merge script re-checks
+          # coverage regardless.
+          # Newest-first candidates via their bucket-0 artifact. Fork PRs run
+          # their own ci.yml, so name and branch are spoofable; require
+          # same-repo. Assignment failures trip set -e and fail loudly.
+          candidates=$(
+            gh api "repos/${GITHUB_REPOSITORY}/actions/artifacts?name=junit-integration-0&per_page=100" \
+              --jq '.artifacts[] | select(.expired | not) | .workflow_run
+                    | select(.head_branch == "dev" and .head_repository_id != null
+                             and .head_repository_id == .repository_id)
+                    | .id'
+          )
+          # Green runs first, then the rest newest first; a run missing a
+          # bucket fails the coverage check and the next one is tried
+          green=""
+          rest=""
+          for id in ${candidates}; do
+            conclusion=$(gh api "repos/${GITHUB_REPOSITORY}/actions/runs/${id}" --jq '.conclusion // ""')
+            if [ "${conclusion}" = "success" ]; then
+              green="${green} ${id}"
+            elif [ -n "${conclusion}" ]; then
+              rest="${rest} ${id}"
+            fi
+          done
+          # helpers.py imports colorama; the script needs nothing else
+          pip install colorama
+          for id in ${green} ${rest}; do
+            rm -rf /tmp/junit
+            if ! gh run download "${id}" --repo "${GITHUB_REPOSITORY}" -p "junit-integration-*" -D /tmp/junit; then
+              echo "::warning::Could not download artifacts for run ${id}; trying the next"
+              continue
+            fi
+            status=0
+            python script/update_integration_test_durations.py /tmp/junit || status=$?
+            if [ "${status}" -eq 0 ]; then
+              echo "Refreshed from run ${id}"
+              exit 0
+            fi
+            # Only EXIT_LOW_COVERAGE (3) from the script advances to the next run
+            [ "${status}" -eq 3 ] || exit 1
+            echo "::warning::Run ${id} covers too few test files; trying the next"
+          done
+          echo "::error::No dev CI run with usable junit artifacts in range; the feed is starved"
+          exit 1
+
+      - name: Commit changes
+        uses: peter-evans/create-pull-request@5f6978faf089d4d20b00c7766989d076bb2fc7f1 # v8.1.1
+        with:
+          commit-message: "[ci] Refresh integration test durations"
+          committer: esphome[bot] <115708604+esphome[bot]@users.noreply.github.com>
+          author: esphome[bot] <115708604+esphome[bot]@users.noreply.github.com>
+          branch: sync/integration-durations
+          delete-branch: true
+          title: "[ci] Refresh integration test durations"
+          body-path: .github/PULL_REQUEST_TEMPLATE.md
+          token: ${{ steps.generate-token.outputs.token }}

script/determine-jobs.py

@@ -53,8 +53,10 @@ from collections import Counter
 from enum import StrEnum
 from functools import cache
 import json
+import math
 import os
 from pathlib import Path
+import statistics
 import sys
 from typing import Any
 
@@ -67,7 +69,9 @@ from clang_tidy_hash import (
 from helpers import (
     CPP_FILE_EXTENSIONS,
     ESPHOME_TESTS_COMPONENTS_PATH,
+    INTEGRATION_TESTS_PATH,
     PYTHON_FILE_EXTENSIONS,
+    all_integration_test_files,
     base_python_changed,
     changed_files,
     core_changed,
@@ -83,6 +87,8 @@ from helpers import (
     get_target_branch,
     git_ls_files,
     is_validate_only_file,
+    load_integration_durations,
+    lpt_partition,
     root_path,
 )
 from split_components_for_ci import create_intelligent_batches
@@ -96,10 +102,13 @@ CLANG_TIDY_SPLIT_THRESHOLD = 65
 # Isolated components count as 10x, groupable components count as 1x
 COMPONENT_TEST_BATCH_SIZE = 40
 
-# Integration test bucketing: when more than the threshold tests are scheduled,
-# fan out across this many parallel jobs. Below the threshold, a single job runs.
+# Above the threshold, fan out across up to this many jobs, balanced by the
+# recorded per-file durations. The target is serial junit-time weight per
+# bucket, not wall time (calibrated with the conftest compile cap); it
+# sizes the bucket count for small subsets.
 INTEGRATION_TESTS_SPLIT_THRESHOLD = 10
-INTEGRATION_TESTS_SPLIT_BUCKETS = 3
+INTEGRATION_TESTS_SPLIT_BUCKETS = 5
+INTEGRATION_TESTS_TARGET_BUCKET_WEIGHT = 360.0
 
 # platformio and aioesphomeapi (requirements.txt), the pytest stack
 # (requirements_test.txt) and the fixture every session compiles; a change
@@ -113,27 +122,13 @@ INTEGRATION_TESTS_TRIGGER_FILES = frozenset(
 )
 
 
-def _split_list(items: list[str], n: int) -> list[list[str]]:
-    """Split a list into n roughly-equal contiguous parts (matches script/clang-tidy)."""
-    k, m = divmod(len(items), n)
-    return [items[i * k + min(i, m) : (i + 1) * k + min(i + 1, m)] for i in range(n)]
-
-
-def _all_integration_test_files() -> list[str]:
-    """Return all integration test file paths, sorted, relative to repo root."""
-    return sorted(
-        str(p.relative_to(root_path))
-        for p in (Path(root_path) / "tests" / "integration").glob("test_*.py")
-    )
-
-
 def _compute_integration_test_buckets(
     integration_run_all: bool,
     integration_test_files: list[str],
 ) -> tuple[bool, list[dict[str, Any]]]:
     """Compute (run_integration, buckets) from the determine_integration_tests result.
 
-    Pure function for unit testing — no I/O beyond `_all_integration_test_files`
+    Pure function for unit testing — no I/O beyond `all_integration_test_files`
     when `integration_run_all` is set.
 
     `buckets` is a list of `{name, tests}` dicts where `tests` is a JSON-friendly
@@ -141,7 +136,7 @@ def _compute_integration_test_buckets(
     shell word-splitting / glob hazards.
     """
     if integration_run_all:
-        files = _all_integration_test_files()
+        files = all_integration_test_files()
     else:
         files = sorted(integration_test_files)
 
@@ -152,12 +147,23 @@ def _compute_integration_test_buckets(
         return False, []
 
     if len(files) > INTEGRATION_TESTS_SPLIT_THRESHOLD:
-        parts = [
-            part for part in _split_list(files, INTEGRATION_TESTS_SPLIT_BUCKETS) if part
-        ]
+        durations = load_integration_durations()
+        # Unrecorded files weigh the recording's median; with no recording a
+        # file weighs a whole bucket, which keeps the full fan-out
+        default = (
+            statistics.median(durations.values())
+            if durations
+            else INTEGRATION_TESTS_TARGET_BUCKET_WEIGHT
+        )
+        weights = {f: durations.get(f, default) for f in files}
+        count = min(
+            INTEGRATION_TESTS_SPLIT_BUCKETS,
+            math.ceil(sum(weights.values()) / INTEGRATION_TESTS_TARGET_BUCKET_WEIGHT),
+        )
+        # count <= SPLIT_BUCKETS < threshold < len(files): no group is empty
+        parts = [sorted(part) for part in lpt_partition(files, weights, count)]
         buckets = [
-            {"name": f"{i + 1}/{len(parts)}", "tests": part}
-            for i, part in enumerate(parts)
+            {"name": f"{i + 1}/{count}", "tests": part} for i, part in enumerate(parts)
         ]
     else:
         buckets = [{"name": "1/1", "tests": files}]
@@ -264,9 +270,9 @@ def determine_integration_tests(branch: str | None = None) -> tuple[bool, list[s
     # If infrastructure Python files changed (conftest, utils, etc.), run all tests
     # Excludes test files (test_*.py), fixtures, and non-Python files (README.md)
     if any(
-        f.startswith("tests/integration/")
+        f.startswith(INTEGRATION_TESTS_PATH)
         and f.endswith(".py")
-        and not f.startswith("tests/integration/test_")
+        and not f.startswith(f"{INTEGRATION_TESTS_PATH}test_")
         and "/fixtures/" not in f
         for f in files
     ):
@@ -277,9 +283,9 @@ def determine_integration_tests(branch: str | None = None) -> tuple[bool, list[s
     fixture_to_test_files = get_fixture_to_test_files()
 
     for f in files:
-        if f.startswith("tests/integration/test_") and f.endswith(".py"):
+        if f.startswith(f"{INTEGRATION_TESTS_PATH}test_") and f.endswith(".py"):
             test_files.add(f)
-        elif f.startswith("tests/integration/fixtures/"):
+        elif f.startswith(f"{INTEGRATION_TESTS_PATH}fixtures/"):
             if f.endswith(".yaml"):
                 # Fixture YAML changed - add corresponding test file(s)
                 test_files.update(fixture_to_test_files.get(Path(f).stem, ()))
@@ -1415,6 +1421,7 @@ def main() -> None:
     output: dict[str, Any] = {
         "core_ci": run_core_ci,
         "integration_tests": run_integration,
+        "integration_run_all": integration_run_all,
         "integration_test_buckets": integration_test_buckets,
         "clang_tidy": run_clang_tidy,
         "clang_tidy_mode": clang_tidy_mode,

script/helpers.py

@@ -43,6 +43,53 @@ ESPHOME_TESTS_COMPONENTS_PATH = "tests/components/"
 # Tuple of component and test paths for efficient startswith checks
 COMPONENT_AND_TESTS_PATHS = (ESPHOME_COMPONENTS_PATH, ESPHOME_TESTS_COMPONENTS_PATH)
 
+# Integration tests path prefix
+INTEGRATION_TESTS_PATH = "tests/integration/"
+
+# Per-file integration test durations from CI junit output; shared by the
+# reader (determine-jobs) and writer (update_integration_test_durations)
+INTEGRATION_TEST_DURATIONS_FILE = "tests/integration/integration_test_durations.json"
+
+
+def all_integration_test_files() -> list[str]:
+    """Return all integration test file paths, sorted, relative to repo root."""
+    return sorted(
+        p.relative_to(root_path).as_posix()
+        for p in (Path(root_path) / "tests" / "integration").glob("test_*.py")
+    )
+
+
+def load_integration_durations() -> dict[str, float]:
+    """Return recorded per-file pytest durations in seconds; empty when unavailable."""
+    try:
+        raw = json.loads(
+            (Path(root_path) / INTEGRATION_TEST_DURATIONS_FILE).read_text()
+        )
+        if not isinstance(raw, dict):
+            print(
+                f"integration durations unavailable: expected an object, "
+                f"got {type(raw).__name__}",
+                file=sys.stderr,
+            )
+            return {}
+    except (OSError, ValueError) as err:
+        # The file ships in the repo; degrade to unweighted bucketing, loudly
+        print(f"integration durations unavailable: {err}", file=sys.stderr)
+        return {}
+    durations = {
+        key: seconds
+        for key, value in raw.items()
+        if isinstance(value, (int, float)) and (seconds := float(value)) > 0
+    }
+    if len(durations) != len(raw):
+        # One bad entry must not discard the whole recording
+        print(
+            f"dropped {len(raw) - len(durations)} invalid duration entries",
+            file=sys.stderr,
+        )
+    return durations
+
+
 # Base bus components - these ARE the bus implementations and should not
 # be flagged as needing migration since they are the platform/base components
 BASE_BUS_COMPONENTS = {
@@ -1545,3 +1592,21 @@ def get_cpp_changed_components(files: list[str]) -> list[str]:
         if file.startswith(ESPHOME_COMPONENTS_PATH):
             affected.update(find_children_of_component(components_graph, component))
     return sorted(c for c in affected if has_cpp_unit_tests(c, tests_dir))
+
+
+def lpt_partition(
+    items: list[str], weights: dict[str, float], count: int
+) -> list[list[str]]:
+    """Partition items into `count` weight-balanced groups (LPT greedy).
+
+    Heaviest item first into the lightest group. Ties keep input order, so
+    pass pre-sorted items for deterministic output. script/clang-tidy's
+    split_list is the unweighted contiguous sibling.
+    """
+    groups: list[list[str]] = [[] for _ in range(count)]
+    group_weights = [0.0] * count
+    for item in sorted(items, key=lambda i: -weights[i]):
+        lightest = min(range(count), key=group_weights.__getitem__)
+        groups[lightest].append(item)
+        group_weights[lightest] += weights[item]
+    return groups

script/update_integration_test_durations.py

@@ -0,0 +1,119 @@
+#!/usr/bin/env python3
+"""Merge CI junit output into tests/integration/integration_test_durations.json.
+
+The integration-tests CI job uploads one junit XML artifact per bucket on
+full matrix dev runs. Download a run's artifacts and merge the per file
+durations into the recording used by script/determine-jobs.py:
+
+    gh run download <run-id> --repo esphome/esphome -p "junit-integration-*" -D /tmp/junit
+    script/update_integration_test_durations.py /tmp/junit
+
+Missing files keep their previous recording and deleted files drop out; a
+run covering under 90% of the test files aborts unless --allow-partial.
+"""
+
+from __future__ import annotations
+
+import argparse
+from collections import defaultdict
+import json
+from pathlib import Path
+import sys
+import xml.etree.ElementTree as ET
+
+from helpers import (
+    INTEGRATION_TEST_DURATIONS_FILE,
+    INTEGRATION_TESTS_PATH,
+    all_integration_test_files,
+    load_integration_durations,
+    root_path,
+)
+
+DURATIONS_FILE = Path(root_path) / INTEGRATION_TEST_DURATIONS_FILE
+MIN_COVERAGE = 0.9
+# Exit code for the expected "run covers too few files" refusal, so the
+# refresh workflow can move on to the next candidate run
+EXIT_LOW_COVERAGE = 3
+
+
+def collect_durations(junit_dir: Path, known_files: set[str]) -> dict[str, float]:
+    """Sum junit testcase times per integration test file, in seconds."""
+    durations: defaultdict[str, float] = defaultdict(float)
+    unmatched = 0
+    xml_files = sorted(junit_dir.rglob("*.xml"))
+    if not xml_files:
+        raise SystemExit(f"no junit XML files found under {junit_dir}")
+    for xml_file in xml_files:
+        for testcase in ET.parse(xml_file).getroot().iter("testcase"):
+            # Skipped/errored testcases carry time="0"; recording them would
+            # overwrite a good previous duration
+            if any(
+                testcase.find(tag) is not None
+                for tag in ("skipped", "error", "failure")
+            ):
+                continue
+            # classname is the dotted module plus any test class, e.g.
+            # tests.integration.test_x or tests.integration.test_x.TestFoo
+            parts = testcase.get("classname", "").split(".")
+            if parts[:2] != ["tests", "integration"] or len(parts) < 3:
+                unmatched += 1
+                continue
+            path = f"{INTEGRATION_TESTS_PATH}{parts[2]}.py"
+            if path not in known_files:
+                print(f"skipping unknown test module {path}", file=sys.stderr)
+                continue
+            durations[path] += float(testcase.get("time", "0"))
+    if unmatched:
+        # A junit naming change would otherwise shrink the recording silently
+        raise SystemExit(
+            f"{unmatched} testcases with unexpected classnames; the junit layout changed"
+        )
+    # An all-skipped file totals 0.0; let the merge keep its previous entry
+    return {k: v for k, v in durations.items() if v > 0}
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "junit_dir", type=Path, help="directory containing downloaded junit XML files"
+    )
+    parser.add_argument(
+        "--allow-partial",
+        action="store_true",
+        help="merge a run covering under 90%% of the test files",
+    )
+    args = parser.parse_args()
+
+    on_disk = set(all_integration_test_files())
+    if not on_disk:
+        raise SystemExit("no integration test files found; wrong checkout root?")
+    collected = collect_durations(args.junit_dir, on_disk)
+    coverage = len(collected.keys() & on_disk) / len(on_disk)
+    if coverage < MIN_COVERAGE and not args.allow_partial:
+        print(
+            f"artifacts cover only {coverage:.0%} of {len(on_disk)} test files; "
+            "use a full matrix run or pass --allow-partial to merge anyway",
+            file=sys.stderr,
+        )
+        return EXIT_LOW_COVERAGE
+
+    # Validated load: a bad previous entry cannot survive the round trip, and
+    # an unreadable file aborts rather than being overwritten
+    previous = load_integration_durations()
+    if DURATIONS_FILE.is_file() and not previous:
+        raise SystemExit(f"{DURATIONS_FILE} is unreadable; refusing to overwrite it")
+    # New recordings win, absent files keep theirs, deleted files drop out
+    merged = {
+        path: collected.get(path, previous.get(path))
+        for path in sorted(on_disk)
+        if path in collected or path in previous
+    }
+    DURATIONS_FILE.write_text(
+        json.dumps({k: round(v, 2) for k, v in merged.items()}, indent=2) + "\n"
+    )
+    print(f"wrote {len(merged)} entries to {DURATIONS_FILE} ({coverage:.0%} fresh)")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/integration/conftest.py

@@ -23,6 +23,7 @@ import pytest_asyncio
 
 import esphome.config
 from esphome.core import CORE
+from esphome.helpers import get_usable_cpu_count
 from esphome.platformio.toolchain import get_idedata
 
 from .const import (
@@ -67,6 +68,14 @@ def _get_platformio_env(cache_dir: Path) -> dict[str, str]:
     env["PLATFORMIO_LIBDEPS_DIR"] = str(cache_dir / "libdeps" / worker)
     # Prevent cache cleaning during integration tests
     env["ESPHOME_SKIP_CLEAN_BUILD"] = "1"
+    # Cap each compile's -j so several xdist workers do not each spawn a
+    # full-width compiler fan-out on the same machine. An explicit env wins.
+    if "ESPHOME_DEFAULT_COMPILE_PROCESS_LIMIT" not in os.environ:
+        workers = int(os.environ.get("PYTEST_XDIST_WORKER_COUNT", "1"))
+        # Floor of 2 keeps a lone tail compile from running fully serial
+        env["ESPHOME_DEFAULT_COMPILE_PROCESS_LIMIT"] = str(
+            max(2, get_usable_cpu_count() // workers)
+        )
     # Compile with THIS tree's esphome sources, not wherever the venv's editable
     # install points (which may be a different git worktree or checkout).
     repo_root = str(Path(__file__).resolve().parent.parent.parent)

tests/integration/integration_test_durations.json

@@ -0,0 +1,142 @@
+{
+  "tests/integration/test_action_concurrent_reentry.py": 45.23,
+  "tests/integration/test_addressable_light_transition.py": 74.47,
+  "tests/integration/test_alarm_control_panel_state_transitions.py": 74.1,
+  "tests/integration/test_api_action_metadata.py": 62.1,
+  "tests/integration/test_api_action_responses.py": 71.08,
+  "tests/integration/test_api_action_timeout.py": 21.64,
+  "tests/integration/test_api_conditional_memory.py": 13.72,
+  "tests/integration/test_api_custom_services.py": 24.16,
+  "tests/integration/test_api_get_time_response_timezone.py": 23.48,
+  "tests/integration/test_api_homeassistant.py": 37.87,
+  "tests/integration/test_api_homeassistant_action_no_subscriber.py": 14.38,
+  "tests/integration/test_api_list_entities_backpressure.py": 26.85,
+  "tests/integration/test_api_message_size_batching.py": 33.36,
+  "tests/integration/test_api_reboot_timeout.py": 13.63,
+  "tests/integration/test_api_string_lambda.py": 25.04,
+  "tests/integration/test_api_vv_logging.py": 16.6,
+  "tests/integration/test_api_zero_psk_provisioning.py": 43.14,
+  "tests/integration/test_areas_and_devices.py": 25.98,
+  "tests/integration/test_automation_wait_actions.py": 21.91,
+  "tests/integration/test_automations.py": 42.43,
+  "tests/integration/test_batch_delay_zero_rapid_transitions.py": 16.65,
+  "tests/integration/test_binary_sensor_autorepeat_filter.py": 28.67,
+  "tests/integration/test_binary_sensor_invalidate_state.py": 23.69,
+  "tests/integration/test_blocking_warning_log_time_not_charged_to_next_operation.py": 22.99,
+  "tests/integration/test_build_info.py": 24.96,
+  "tests/integration/test_camera_mock.py": 14.47,
+  "tests/integration/test_climate_control_action.py": 31.07,
+  "tests/integration/test_climate_custom_modes.py": 28.59,
+  "tests/integration/test_continuation_actions.py": 14.96,
+  "tests/integration/test_cover_control_action.py": 26.14,
+  "tests/integration/test_crc8_helper.py": 10.92,
+  "tests/integration/test_device_id_in_state.py": 64.97,
+  "tests/integration/test_duplicate_entities.py": 30.81,
+  "tests/integration/test_entity_icon.py": 32.85,
+  "tests/integration/test_fan_turn_on_action.py": 24.91,
+  "tests/integration/test_fnv1_hash_object_id.py": 12.54,
+  "tests/integration/test_fnv1a_hash.py": 21.8,
+  "tests/integration/test_gpio_expander_cache.py": 5.2,
+  "tests/integration/test_host_logger_thread_safety.py": 21.7,
+  "tests/integration/test_host_mode_basic.py": 13.62,
+  "tests/integration/test_host_mode_batch_delay.py": 14.56,
+  "tests/integration/test_host_mode_climate_basic_state.py": 30.95,
+  "tests/integration/test_host_mode_climate_control.py": 29.06,
+  "tests/integration/test_host_mode_empty_string_options.py": 27.22,
+  "tests/integration/test_host_mode_entity_fields.py": 30.95,
+  "tests/integration/test_host_mode_fan_preset.py": 14.44,
+  "tests/integration/test_host_mode_many_entities.py": 54.13,
+  "tests/integration/test_host_mode_many_entities_multiple_connections.py": 32.17,
+  "tests/integration/test_host_mode_noise_encryption.py": 42.77,
+  "tests/integration/test_host_mode_reconnect.py": 4.06,
+  "tests/integration/test_host_mode_sensor.py": 13.47,
+  "tests/integration/test_host_ota.py": 21.4,
+  "tests/integration/test_host_preferences.py": 25.43,
+  "tests/integration/test_host_preferences_suspend_resume.py": 19.2,
+  "tests/integration/test_improv_serial_uart.py": 31.52,
+  "tests/integration/test_large_message_batching.py": 15.64,
+  "tests/integration/test_legacy_area.py": 22.63,
+  "tests/integration/test_legacy_climate_compat.py": 26.13,
+  "tests/integration/test_legacy_fan_compat.py": 24.05,
+  "tests/integration/test_light_automations.py": 30.86,
+  "tests/integration/test_light_binary_effect_off_phase.py": 23.19,
+  "tests/integration/test_light_calls.py": 32.35,
+  "tests/integration/test_light_constant_brightness.py": 29.89,
+  "tests/integration/test_light_control_action.py": 29.06,
+  "tests/integration/test_light_dim_relative_action.py": 29.61,
+  "tests/integration/test_light_effect_zero_brightness.py": 18.68,
+  "tests/integration/test_light_initial_state.py": 24.49,
+  "tests/integration/test_light_toggle_action.py": 26.46,
+  "tests/integration/test_lock_automations.py": 23.28,
+  "tests/integration/test_logger_buffered_recursion_guard.py": 24.29,
+  "tests/integration/test_loop_disable_enable.py": 45.28,
+  "tests/integration/test_loop_interval_decoupling.py": 28.35,
+  "tests/integration/test_loop_interval_default_not_pulled_forward.py": 21.97,
+  "tests/integration/test_micros_to_millis.py": 20.79,
+  "tests/integration/test_multi_click_trigger.py": 26.2,
+  "tests/integration/test_multi_device_preferences.py": 16.87,
+  "tests/integration/test_noise_encryption_key_protection.py": 77.05,
+  "tests/integration/test_object_id_api_verification.py": 73.51,
+  "tests/integration/test_object_id_friendly_name_no_mac_suffix.py": 62.33,
+  "tests/integration/test_object_id_no_friendly_name.py": 43.47,
+  "tests/integration/test_online_image_auto_detects_image_bmp_mime.py": 32.21,
+  "tests/integration/test_online_image_auto_detects_redirected_image_bmp_mime.py": 56.86,
+  "tests/integration/test_online_image_bmp.py": 50.9,
+  "tests/integration/test_oversized_payloads.py": 53.2,
+  "tests/integration/test_preference_key_stability.py": 26.09,
+  "tests/integration/test_runtime_stats.py": 18.34,
+  "tests/integration/test_safe_mode_loop_runs.py": 10.07,
+  "tests/integration/test_scheduler_blocking_warning.py": 40.91,
+  "tests/integration/test_scheduler_bulk_cleanup.py": 23.14,
+  "tests/integration/test_scheduler_defer_cancel.py": 24.54,
+  "tests/integration/test_scheduler_defer_cancel_regular.py": 13.48,
+  "tests/integration/test_scheduler_defer_fifo_simple.py": 26.86,
+  "tests/integration/test_scheduler_defer_stress.py": 27.23,
+  "tests/integration/test_scheduler_heap_stress.py": 24.02,
+  "tests/integration/test_scheduler_internal_id_no_collision.py": 24.57,
+  "tests/integration/test_scheduler_interval_reschedule.py": 13.12,
+  "tests/integration/test_scheduler_interval_zero_coerced.py": 22.91,
+  "tests/integration/test_scheduler_null_name.py": 23.46,
+  "tests/integration/test_scheduler_numeric_id_test.py": 24.54,
+  "tests/integration/test_scheduler_pool.py": 25.0,
+  "tests/integration/test_scheduler_rapid_cancellation.py": 14.68,
+  "tests/integration/test_scheduler_recursive_timeout.py": 25.35,
+  "tests/integration/test_scheduler_removed_item_race.py": 26.19,
+  "tests/integration/test_scheduler_self_keyed.py": 23.43,
+  "tests/integration/test_scheduler_simultaneous_callbacks.py": 22.16,
+  "tests/integration/test_scheduler_string_test.py": 15.22,
+  "tests/integration/test_script_array_params.py": 14.67,
+  "tests/integration/test_script_delay_params.py": 15.65,
+  "tests/integration/test_script_queued.py": 24.93,
+  "tests/integration/test_script_queued_idle_loop.py": 5.04,
+  "tests/integration/test_script_wait_on_boot.py": 13.08,
+  "tests/integration/test_select_stringref_trigger.py": 29.6,
+  "tests/integration/test_sensor_filters_delta.py": 28.01,
+  "tests/integration/test_sensor_filters_ring_buffer.py": 25.04,
+  "tests/integration/test_sensor_filters_sliding_window.py": 71.5,
+  "tests/integration/test_sensor_filters_value_list.py": 16.94,
+  "tests/integration/test_sensor_timeout_filter.py": 29.48,
+  "tests/integration/test_socket_wake_gate_tcp.py": 20.36,
+  "tests/integration/test_status_flags.py": 37.42,
+  "tests/integration/test_strftime_to.py": 22.61,
+  "tests/integration/test_syslog.py": 16.34,
+  "tests/integration/test_template_alarm_control_panel_many_sensors.py": 29.81,
+  "tests/integration/test_template_text_save.py": 25.43,
+  "tests/integration/test_text_command.py": 23.34,
+  "tests/integration/test_text_sensor_raw_state.py": 69.57,
+  "tests/integration/test_uart_mock_ld2410.py": 37.95,
+  "tests/integration/test_uart_mock_ld2412.py": 93.22,
+  "tests/integration/test_uart_mock_ld2420.py": 43.24,
+  "tests/integration/test_uart_mock_ld2450.py": 31.75,
+  "tests/integration/test_uart_mock_modbus.py": 667.4,
+  "tests/integration/test_udp.py": 9.38,
+  "tests/integration/test_use_address_runtime.py": 37.05,
+  "tests/integration/test_valve_control_action.py": 24.47,
+  "tests/integration/test_varint_five_byte_device_id.py": 25.03,
+  "tests/integration/test_wait_until_mid_loop_timing.py": 23.73,
+  "tests/integration/test_wait_until_on_boot.py": 9.16,
+  "tests/integration/test_wait_until_ordering.py": 13.3,
+  "tests/integration/test_wait_until_reentrant_restart.py": 25.23,
+  "tests/integration/test_wake_loop_forces_phase_b.py": 23.34,
+  "tests/integration/test_water_heater_template.py": 17.67
+}

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 == []
@@ -3146,3 +3149,86 @@ def test_memory_impact_elf_layouts_are_found(tmp_path: Path) -> None:
         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_helpers.py

@@ -2120,3 +2120,30 @@ def test_get_cpp_changed_components_independent_of_cwd(
     assert helpers.get_cpp_changed_components(
         ["tests/components/time/__init__.py"]
     ) == ["time"]
+
+
+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_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,
+    }