[ci] Add import-time regression check for esphome.__main__ (#15954)

.github/workflows/ci.yml

@@ -108,6 +108,34 @@ jobs:
           script/generate-esp32-boards.py --check
           script/generate-rp2040-boards.py --check
 
+  import-time:
+    name: Check import esphome.__main__ time
+    runs-on: ubuntu-24.04
+    needs:
+      - common
+      - determine-jobs
+    if: needs.determine-jobs.outputs.import-time == 'true'
+    steps:
+      - name: Check out code from GitHub
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+      - name: Restore Python
+        uses: ./.github/actions/restore-python
+        with:
+          python-version: ${{ env.DEFAULT_PYTHON }}
+          cache-key: ${{ needs.common.outputs.cache-key }}
+      - name: Check import time against budget and write waterfall HAR
+        run: |
+          . venv/bin/activate
+          script/check_import_time.py --check --har importtime.har
+      - name: Upload waterfall HAR
+        if: always()
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: import-time-waterfall
+          path: importtime.har
+          if-no-files-found: ignore
+          retention-days: 14
+
   pytest:
     name: Run pytest
     strategy:
@@ -176,6 +204,7 @@ jobs:
       clang-tidy: ${{ steps.determine.outputs.clang-tidy }}
       clang-tidy-mode: ${{ steps.determine.outputs.clang-tidy-mode }}
       python-linters: ${{ steps.determine.outputs.python-linters }}
+      import-time: ${{ steps.determine.outputs.import-time }}
       changed-components: ${{ steps.determine.outputs.changed-components }}
       changed-components-with-tests: ${{ steps.determine.outputs.changed-components-with-tests }}
       directly-changed-components-with-tests: ${{ steps.determine.outputs.directly-changed-components-with-tests }}
@@ -219,6 +248,7 @@ jobs:
           echo "clang-tidy=$(echo "$output" | jq -r '.clang_tidy')" >> $GITHUB_OUTPUT
           echo "clang-tidy-mode=$(echo "$output" | jq -r '.clang_tidy_mode')" >> $GITHUB_OUTPUT
           echo "python-linters=$(echo "$output" | jq -r '.python_linters')" >> $GITHUB_OUTPUT
+          echo "import-time=$(echo "$output" | jq -r '.import_time')" >> $GITHUB_OUTPUT
           echo "changed-components=$(echo "$output" | jq -c '.changed_components')" >> $GITHUB_OUTPUT
           echo "changed-components-with-tests=$(echo "$output" | jq -c '.changed_components_with_tests')" >> $GITHUB_OUTPUT
           echo "directly-changed-components-with-tests=$(echo "$output" | jq -c '.directly_changed_components_with_tests')" >> $GITHUB_OUTPUT

requirements_test.txt

@@ -12,3 +12,6 @@ pytest-asyncio==1.3.0
 pytest-xdist==3.8.0
 asyncmock==0.4.2
 hypothesis==6.92.1
+
+# Used by the import-time regression check (.github/workflows/ci.yml → import-time job)
+importtime-waterfall==1.0.0

script/check_import_time.py

@@ -0,0 +1,241 @@
+#!/usr/bin/env python3
+"""Regression check for `import esphome.__main__` cost.
+
+Runs `python -m importtime_waterfall --har esphome.__main__` (which invokes
+`-X importtime` in fresh subprocesses, best-of-N) and compares the root
+cumulative import time against a checked-in budget
+(`script/import_time_budget.json`).
+
+The CLI pays this cost on every invocation before the requested command even
+runs, so a regression here hurts every user.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+import subprocess
+import sys
+from typing import Any, TextIO
+
+SCRIPT_DIR = Path(__file__).parent
+BUDGET_PATH = SCRIPT_DIR / "import_time_budget.json"
+
+TARGET_MODULE = "esphome.__main__"
+DEFAULT_MARGIN_PCT = 15
+OFFENDERS_TOP_N = 15
+
+
+def run_waterfall(module: str) -> str:
+    """Run `importtime_waterfall --har <module>` and return the HAR JSON text.
+
+    `importtime_waterfall` itself runs the target in 6 fresh subprocesses
+    under `-X importtime` and emits the HAR of the fastest run.
+    """
+    result = subprocess.run(
+        [sys.executable, "-m", "importtime_waterfall", "--har", module],
+        check=True,
+        stdout=subprocess.PIPE,
+        text=True,
+    )
+    return result.stdout
+
+
+def measure(module: str, har_path: Path | None = None) -> dict[str, Any]:
+    """Return the parsed HAR for importing `module`.
+
+    When `har_path` is given, also write the raw HAR JSON to that path so
+    callers can combine `--check` with `--har` without measuring twice.
+    """
+    har_text = run_waterfall(module)
+    if har_path is not None:
+        har_path.write_text(har_text)
+    return json.loads(har_text)
+
+
+def _entries(har: dict[str, Any]) -> list[dict[str, Any]]:
+    return har["log"]["entries"]
+
+
+def root_cumulative_us(har: dict[str, Any], module: str) -> int:
+    """Return the cumulative import time (µs) of `module` from a HAR.
+
+    The HAR `time` field is authored by importtime_waterfall using µs values
+    fed through `timedelta(milliseconds=...)`, so the number read back is the
+    original self/cumulative time in microseconds (labelled "ms" in HAR).
+    """
+    for entry in _entries(har):
+        if entry["request"]["url"] == module:
+            return entry["time"]
+    raise RuntimeError(
+        f"No HAR entry for {module!r}. Is it importable with "
+        f"`python -c 'import {module}'`?"
+    )
+
+
+def top_offenders(har: dict[str, Any], n: int) -> list[tuple[str, int, int]]:
+    """Return up to `n` (name, self_us, cumulative_us), ranked by self_us desc.
+
+    A module imported from multiple places is counted once (first entry wins,
+    matching importtime's own de-duplication).
+    """
+    seen: dict[str, tuple[int, int]] = {}
+    for entry in _entries(har):
+        name = entry["request"]["url"]
+        if name in seen:
+            continue
+        self_us = entry["timings"]["receive"]
+        cumulative_us = entry["time"]
+        seen[name] = (self_us, cumulative_us)
+    ranked = sorted(
+        ((name, s, c) for name, (s, c) in seen.items()),
+        key=lambda row: row[1],
+        reverse=True,
+    )
+    return ranked[:n]
+
+
+def read_budget() -> dict[str, Any]:
+    if not BUDGET_PATH.exists():
+        return {}
+    with BUDGET_PATH.open() as f:
+        return json.load(f)
+
+
+def write_budget(cumulative_us: int, margin_pct: int) -> None:
+    payload = {
+        "target_module": TARGET_MODULE,
+        "margin_pct": margin_pct,
+        "cumulative_us": cumulative_us,
+    }
+    with BUDGET_PATH.open("w") as f:
+        json.dump(payload, f, indent=2)
+        f.write("\n")
+
+
+def _format_us(us: int) -> str:
+    if us >= 1000:
+        return f"{us / 1000:.1f}ms"
+    return f"{us}us"
+
+
+def _print_offenders_table(
+    offenders: list[tuple[str, int, int]], stream: TextIO
+) -> None:
+    name_w = max(len(name) for name, _, _ in offenders)
+    print(f"\n{'module':<{name_w}}  {'self':>10}  {'cumulative':>12}", file=stream)
+    print(f"{'-' * name_w}  {'-' * 10}  {'-' * 12}", file=stream)
+    for name, self_us, cum_us in offenders:
+        print(
+            f"{name:<{name_w}}  {_format_us(self_us):>10}  {_format_us(cum_us):>12}",
+            file=stream,
+        )
+
+
+def cmd_check(args: argparse.Namespace) -> int:
+    budget = read_budget()
+    if not budget:
+        print(
+            f"ERROR: {BUDGET_PATH.name} missing. Run with --update first.",
+            file=sys.stderr,
+        )
+        return 2
+
+    har = measure(TARGET_MODULE, har_path=Path(args.har) if args.har else None)
+    measured = root_cumulative_us(har, TARGET_MODULE)
+
+    baseline = budget["cumulative_us"]
+    margin_pct = budget.get("margin_pct", DEFAULT_MARGIN_PCT)
+    ceiling = int(baseline * (1 + margin_pct / 100))
+
+    summary = (
+        f"measured   {TARGET_MODULE}: {_format_us(measured)} "
+        f"(budget {_format_us(baseline)} + {margin_pct}% = {_format_us(ceiling)})"
+    )
+    passed = measured <= ceiling
+    stream = sys.stdout if passed else sys.stderr
+
+    if passed:
+        print(summary)
+    else:
+        print(
+            f"REGRESSION: `import {TARGET_MODULE}` took {_format_us(measured)}, "
+            f"exceeding the budget of {_format_us(baseline)} + {margin_pct}% "
+            f"({_format_us(ceiling)}).",
+            file=stream,
+        )
+
+    print("\nTop import-time offenders (by self time):", file=stream)
+    _print_offenders_table(top_offenders(har, OFFENDERS_TOP_N), stream)
+
+    if not passed:
+        print(
+            "\nIf this regression is intentional, regenerate the budget with:\n"
+            "  script/check_import_time.py --update\n"
+            "Otherwise, consider making the new import lazy "
+            "(import inside the function that uses it).",
+            file=stream,
+        )
+        return 1
+    return 0
+
+
+def cmd_update(args: argparse.Namespace) -> int:
+    har = measure(TARGET_MODULE, har_path=Path(args.har) if args.har else None)
+    measured = root_cumulative_us(har, TARGET_MODULE)
+    write_budget(measured, args.margin_pct)
+    print(
+        f"Wrote {BUDGET_PATH.name}: "
+        f"{TARGET_MODULE}={_format_us(measured)} "
+        f"(margin {args.margin_pct}%)"
+    )
+    return 0
+
+
+def cmd_har_only(args: argparse.Namespace) -> int:
+    Path(args.har).write_text(run_waterfall(TARGET_MODULE))
+    print(f"Wrote waterfall HAR to {args.har}")
+    return 0
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--margin-pct",
+        type=int,
+        default=DEFAULT_MARGIN_PCT,
+        help=(f"Margin over baseline for --update (default: {DEFAULT_MARGIN_PCT}%%)."),
+    )
+    parser.add_argument(
+        "--har",
+        metavar="PATH",
+        help=(
+            "Write a waterfall HAR file at PATH. Can be combined with "
+            "--check or --update to reuse that run's measurement (avoids "
+            "measuring twice)."
+        ),
+    )
+    mode = parser.add_mutually_exclusive_group()
+    mode.add_argument(
+        "--check", action="store_true", help="Fail if measured time exceeds budget."
+    )
+    mode.add_argument(
+        "--update",
+        action="store_true",
+        help="Rewrite the budget from a fresh measurement.",
+    )
+    args = parser.parse_args()
+
+    if args.check:
+        return cmd_check(args)
+    if args.update:
+        return cmd_update(args)
+    if args.har:
+        return cmd_har_only(args)
+    parser.error("Specify at least one of --check, --update, or --har PATH.")
+    return 2  # unreachable; parser.error exits. Here to satisfy ruff RET503.
+
+
+if __name__ == "__main__":
+    sys.exit(main())

script/determine-jobs.py

@@ -349,6 +349,42 @@ def should_run_python_linters(branch: str | None = None) -> bool:
     return _any_changed_file_endswith(branch, PYTHON_FILE_EXTENSIONS)
 
 
+# Files outside esphome/**/*.py whose changes can affect `import esphome.__main__`
+# cost. requirements.txt / pyproject.toml change the dependency graph pulled in
+# by top-level imports; check_import_time.py itself changes the check's behavior.
+IMPORT_TIME_TRIGGER_FILES = frozenset(
+    {
+        "requirements.txt",
+        "requirements_dev.txt",
+        "requirements_test.txt",
+        "pyproject.toml",
+        "script/check_import_time.py",
+        "script/import_time_budget.json",
+    }
+)
+
+
+def should_run_import_time(branch: str | None = None) -> bool:
+    """Determine if the `import esphome.__main__` time regression check should run.
+
+    Runs when any Python file under `esphome/` changes (those modules are
+    loaded transitively from `esphome.__main__`), when dependency
+    declarations change, or when the check script/budget itself changes.
+
+    Args:
+        branch: Branch to compare against. If None, uses default.
+
+    Returns:
+        True if the import-time check should run, False otherwise.
+    """
+    for file in changed_files(branch):
+        if file.startswith("esphome/") and file.endswith(PYTHON_FILE_EXTENSIONS):
+            return True
+        if file in IMPORT_TIME_TRIGGER_FILES:
+            return True
+    return False
+
+
 def determine_cpp_unit_tests(
     branch: str | None = None,
 ) -> tuple[bool, list[str]]:
@@ -780,6 +816,7 @@ def main() -> None:
     run_clang_tidy = should_run_clang_tidy(args.branch)
     run_clang_format = should_run_clang_format(args.branch)
     run_python_linters = should_run_python_linters(args.branch)
+    run_import_time = should_run_import_time(args.branch)
     changed_cpp_file_count = count_changed_cpp_files(args.branch)
 
     # Get changed components
@@ -913,6 +950,7 @@ def main() -> None:
         "clang_tidy_mode": clang_tidy_mode,
         "clang_format": run_clang_format,
         "python_linters": run_python_linters,
+        "import_time": run_import_time,
         "changed_components": changed_components,
         "changed_components_with_tests": changed_components_with_tests,
         "directly_changed_components_with_tests": list(directly_changed_with_tests),

script/import_time_budget.json

@@ -0,0 +1,5 @@
+{
+  "target_module": "esphome.__main__",
+  "margin_pct": 15,
+  "cumulative_us": 123000
+}

tests/script/test_check_import_time.py

@@ -0,0 +1,191 @@
+"""Unit tests for script/check_import_time.py."""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import os
+from pathlib import Path
+import sys
+from unittest.mock import patch
+
+import pytest
+
+# Load the script-under-test as `check_import_time` (it's a hyphenated path
+# inside `script/` that mirrors the existing `determine_jobs` pattern).
+script_dir = os.path.abspath(
+    os.path.join(os.path.dirname(__file__), "..", "..", "script")
+)
+sys.path.insert(0, script_dir)
+spec = importlib.util.spec_from_file_location(
+    "check_import_time", os.path.join(script_dir, "check_import_time.py")
+)
+check_import_time = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(check_import_time)
+
+
+def _entry(name: str, self_us: int, cumulative_us: int) -> dict:
+    """Build a minimal HAR entry matching `importtime_waterfall --har`."""
+    return {
+        "request": {"url": name},
+        "time": cumulative_us,
+        "timings": {"receive": self_us, "wait": cumulative_us - self_us},
+    }
+
+
+def _har(*entries: dict) -> dict:
+    return {"log": {"entries": list(entries)}}
+
+
+def test_root_cumulative_us_returns_time_for_root_module() -> None:
+    har = _har(
+        _entry("dep_a", 500, 500),
+        _entry("dep_b", 300, 300),
+        _entry("esphome.__main__", 100, 1000),
+    )
+    assert check_import_time.root_cumulative_us(har, "esphome.__main__") == 1000
+
+
+def test_root_cumulative_us_missing_module_raises() -> None:
+    har = _har(_entry("something.else", 100, 100))
+    with pytest.raises(RuntimeError, match="No HAR entry for 'esphome.__main__'"):
+        check_import_time.root_cumulative_us(har, "esphome.__main__")
+
+
+def test_top_offenders_ranks_by_self_time_descending() -> None:
+    har = _har(
+        _entry("small", 100, 100),
+        _entry("big", 5000, 5000),
+        _entry("medium", 2000, 2500),
+    )
+    result = check_import_time.top_offenders(har, n=10)
+    assert [name for name, _, _ in result] == ["big", "medium", "small"]
+    assert result[0] == ("big", 5000, 5000)
+
+
+def test_top_offenders_respects_n_limit() -> None:
+    har = _har(*[_entry(f"m{i}", i * 100, i * 100) for i in range(1, 20)])
+    assert len(check_import_time.top_offenders(har, n=5)) == 5
+
+
+def test_top_offenders_dedupes_repeat_names_keeping_first() -> None:
+    har = _har(
+        _entry("pkg", 5000, 5000),
+        _entry("pkg", 100, 100),  # reimport later in trace
+        _entry("other", 1000, 1000),
+    )
+    result = check_import_time.top_offenders(har, n=10)
+    assert [name for name, _, _ in result] == ["pkg", "other"]
+    # First occurrence wins
+    assert ("pkg", 5000, 5000) in result
+
+
+def test_format_us_switches_to_ms_at_threshold() -> None:
+    assert check_import_time._format_us(500) == "500us"
+    assert check_import_time._format_us(999) == "999us"
+    assert check_import_time._format_us(1000) == "1.0ms"
+    assert check_import_time._format_us(12345) == "12.3ms"
+
+
+def test_read_write_budget_roundtrip(tmp_path: Path) -> None:
+    budget_path = tmp_path / "budget.json"
+    with patch.object(check_import_time, "BUDGET_PATH", budget_path):
+        assert check_import_time.read_budget() == {}
+        check_import_time.write_budget(cumulative_us=12345, margin_pct=20)
+        loaded = check_import_time.read_budget()
+    assert loaded["cumulative_us"] == 12345
+    assert loaded["margin_pct"] == 20
+    assert loaded["target_module"] == check_import_time.TARGET_MODULE
+
+
+def test_cmd_check_passes_when_measured_within_ceiling(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    budget_path = tmp_path / "budget.json"
+    budget_path.write_text(
+        json.dumps(
+            {
+                "target_module": check_import_time.TARGET_MODULE,
+                "margin_pct": 15,
+                "cumulative_us": 100000,  # 100ms
+            }
+        )
+    )
+    # Measured 90ms: inside 100ms + 15% = 115ms ceiling
+    har = _har(_entry(check_import_time.TARGET_MODULE, 1000, 90000))
+    args = type("A", (), {"har": None})()
+    with (
+        patch.object(check_import_time, "BUDGET_PATH", budget_path),
+        patch.object(check_import_time, "measure", return_value=har),
+    ):
+        rc = check_import_time.cmd_check(args)
+    assert rc == 0
+    out = capsys.readouterr().out
+    assert "measured   esphome.__main__:" in out
+    assert "budget 100.0ms" in out
+
+
+def test_cmd_check_fails_when_measured_exceeds_ceiling(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    budget_path = tmp_path / "budget.json"
+    budget_path.write_text(
+        json.dumps(
+            {
+                "target_module": check_import_time.TARGET_MODULE,
+                "margin_pct": 15,
+                "cumulative_us": 100000,
+            }
+        )
+    )
+    # Measured 120ms: over 100ms + 15% = 115ms ceiling
+    har = _har(
+        _entry("offender_a", 10000, 10000),
+        _entry(check_import_time.TARGET_MODULE, 1000, 120000),
+    )
+    args = type("A", (), {"har": None})()
+    with (
+        patch.object(check_import_time, "BUDGET_PATH", budget_path),
+        patch.object(check_import_time, "measure", return_value=har),
+    ):
+        rc = check_import_time.cmd_check(args)
+    assert rc == 1
+    err = capsys.readouterr().err
+    assert "REGRESSION" in err
+    assert "120.0ms" in err
+    assert "offender_a" in err  # top offender table
+
+
+def test_cmd_check_returns_2_when_budget_missing(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    budget_path = tmp_path / "nonexistent.json"
+    args = type("A", (), {"har": None})()
+    with patch.object(check_import_time, "BUDGET_PATH", budget_path):
+        rc = check_import_time.cmd_check(args)
+    assert rc == 2
+    assert "missing" in capsys.readouterr().err
+
+
+def test_cmd_check_writes_har_when_path_given(tmp_path: Path) -> None:
+    budget_path = tmp_path / "budget.json"
+    budget_path.write_text(
+        json.dumps(
+            {
+                "target_module": check_import_time.TARGET_MODULE,
+                "margin_pct": 15,
+                "cumulative_us": 100000,
+            }
+        )
+    )
+    har_path = tmp_path / "out.har"
+    har_text = json.dumps(_har(_entry(check_import_time.TARGET_MODULE, 1000, 80000)))
+    args = type("A", (), {"har": str(har_path)})()
+    with (
+        patch.object(check_import_time, "BUDGET_PATH", budget_path),
+        patch.object(check_import_time, "run_waterfall", return_value=har_text),
+    ):
+        rc = check_import_time.cmd_check(args)
+    assert rc == 0
+    assert har_path.exists()
+    assert json.loads(har_path.read_text()) == json.loads(har_text)

tests/script/test_determine_jobs.py

@@ -56,6 +56,13 @@ def mock_should_run_python_linters() -> Generator[Mock, None, None]:
         yield mock
 
 
+@pytest.fixture
+def mock_should_run_import_time() -> Generator[Mock, None, None]:
+    """Mock should_run_import_time from determine_jobs."""
+    with patch.object(determine_jobs, "should_run_import_time") as mock:
+        yield mock
+
+
 @pytest.fixture
 def mock_determine_cpp_unit_tests() -> Generator[Mock, None, None]:
     """Mock determine_cpp_unit_tests from helpers."""
@@ -91,6 +98,7 @@ def test_main_all_tests_should_run(
     mock_should_run_clang_tidy: Mock,
     mock_should_run_clang_format: Mock,
     mock_should_run_python_linters: Mock,
+    mock_should_run_import_time: Mock,
     mock_changed_files: Mock,
     mock_determine_cpp_unit_tests: Mock,
     capsys: pytest.CaptureFixture[str],
@@ -104,6 +112,7 @@ def test_main_all_tests_should_run(
     mock_should_run_clang_tidy.return_value = True
     mock_should_run_clang_format.return_value = True
     mock_should_run_python_linters.return_value = True
+    mock_should_run_import_time.return_value = True
     mock_determine_cpp_unit_tests.return_value = (False, ["wifi", "api", "sensor"])
 
     # Mock changed_files to return non-component files (to avoid memory impact)
@@ -158,6 +167,7 @@ def test_main_all_tests_should_run(
     assert output["clang_tidy_mode"] in ["nosplit", "split"]
     assert output["clang_format"] is True
     assert output["python_linters"] is True
+    assert output["import_time"] is True
     assert output["changed_components"] == ["wifi", "api", "sensor"]
     # changed_components_with_tests will only include components that actually have test files
     assert "changed_components_with_tests" in output
@@ -189,6 +199,7 @@ def test_main_no_tests_should_run(
     mock_should_run_clang_tidy: Mock,
     mock_should_run_clang_format: Mock,
     mock_should_run_python_linters: Mock,
+    mock_should_run_import_time: Mock,
     mock_changed_files: Mock,
     mock_determine_cpp_unit_tests: Mock,
     capsys: pytest.CaptureFixture[str],
@@ -202,6 +213,7 @@ def test_main_no_tests_should_run(
     mock_should_run_clang_tidy.return_value = False
     mock_should_run_clang_format.return_value = False
     mock_should_run_python_linters.return_value = False
+    mock_should_run_import_time.return_value = False
     mock_determine_cpp_unit_tests.return_value = (False, [])
 
     # Mock changed_files to return no component files
@@ -241,6 +253,7 @@ def test_main_no_tests_should_run(
     assert output["clang_tidy_mode"] == "disabled"
     assert output["clang_format"] is False
     assert output["python_linters"] is False
+    assert output["import_time"] is False
     assert output["changed_components"] == []
     assert output["changed_components_with_tests"] == []
     assert output["component_test_count"] == 0
@@ -261,6 +274,7 @@ def test_main_with_branch_argument(
     mock_should_run_clang_tidy: Mock,
     mock_should_run_clang_format: Mock,
     mock_should_run_python_linters: Mock,
+    mock_should_run_import_time: Mock,
     mock_changed_files: Mock,
     mock_determine_cpp_unit_tests: Mock,
     capsys: pytest.CaptureFixture[str],
@@ -274,6 +288,7 @@ def test_main_with_branch_argument(
     mock_should_run_clang_tidy.return_value = True
     mock_should_run_clang_format.return_value = False
     mock_should_run_python_linters.return_value = True
+    mock_should_run_import_time.return_value = True
     mock_determine_cpp_unit_tests.return_value = (False, ["mqtt"])
 
     # Mock changed_files to return non-component files (to avoid memory impact)
@@ -310,6 +325,7 @@ def test_main_with_branch_argument(
     mock_should_run_clang_tidy.assert_called_once_with("main")
     mock_should_run_clang_format.assert_called_once_with("main")
     mock_should_run_python_linters.assert_called_once_with("main")
+    mock_should_run_import_time.assert_called_once_with("main")
 
     # Check output
     captured = capsys.readouterr()
@@ -322,6 +338,7 @@ def test_main_with_branch_argument(
     assert output["clang_tidy_mode"] in ["nosplit", "split"]
     assert output["clang_format"] is False
     assert output["python_linters"] is True
+    assert output["import_time"] is True
     assert output["changed_components"] == ["mqtt"]
     # changed_components_with_tests will only include components that actually have test files
     assert "changed_components_with_tests" in output
@@ -597,6 +614,50 @@ def test_should_run_python_linters_with_branch() -> None:
         mock_changed.assert_called_once_with("release")
 
 
+@pytest.mark.parametrize(
+    ("changed_files", "expected_result"),
+    [
+        # esphome Python files trigger the check
+        (["esphome/__main__.py"], True),
+        (["esphome/components/wifi/__init__.py"], True),
+        (["esphome/core/config.py"], True),
+        (["esphome/types.pyi"], True),
+        # Dependency declarations and the check's own files trigger
+        (["requirements.txt"], True),
+        (["requirements_dev.txt"], True),
+        (["requirements_test.txt"], True),
+        (["pyproject.toml"], True),
+        (["script/check_import_time.py"], True),
+        (["script/import_time_budget.json"], True),
+        # Mixed: any triggering file is enough
+        (["docs/README.md", "esphome/config.py"], True),
+        # Python files outside esphome/ don't trigger
+        (["script/some_other_script.py"], False),
+        (["tests/script/test_determine_jobs.py"], False),
+        # Non-Python changes don't trigger
+        (["esphome/core/component.cpp"], False),
+        (["tests/components/wifi/test.esp32-idf.yaml"], False),
+        (["README.md"], False),
+        ([], False),
+    ],
+)
+def test_should_run_import_time(
+    changed_files: list[str], expected_result: bool
+) -> None:
+    """Test should_run_import_time function."""
+    with patch.object(determine_jobs, "changed_files", return_value=changed_files):
+        result = determine_jobs.should_run_import_time()
+        assert result == expected_result
+
+
+def test_should_run_import_time_with_branch() -> None:
+    """Test should_run_import_time with branch argument."""
+    with patch.object(determine_jobs, "changed_files") as mock_changed:
+        mock_changed.return_value = []
+        determine_jobs.should_run_import_time("release")
+        mock_changed.assert_called_once_with("release")
+
+
 @pytest.mark.parametrize(
     ("changed_files", "expected_result"),
     [