diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 20c349ac00..d60bd6edc3 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/requirements_test.txt b/requirements_test.txt index b35025fa04..568d79d676 100644 --- a/requirements_test.txt +++ b/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 diff --git a/script/check_import_time.py b/script/check_import_time.py new file mode 100755 index 0000000000..0d5362c968 --- /dev/null +++ b/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 ` 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()) diff --git a/script/determine-jobs.py b/script/determine-jobs.py index f036447542..6fd7ab297c 100755 --- a/script/determine-jobs.py +++ b/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), diff --git a/script/import_time_budget.json b/script/import_time_budget.json new file mode 100644 index 0000000000..1e656dc977 --- /dev/null +++ b/script/import_time_budget.json @@ -0,0 +1,5 @@ +{ + "target_module": "esphome.__main__", + "margin_pct": 15, + "cumulative_us": 123000 +} diff --git a/tests/script/test_check_import_time.py b/tests/script/test_check_import_time.py new file mode 100644 index 0000000000..223c58002c --- /dev/null +++ b/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) diff --git a/tests/script/test_determine_jobs.py b/tests/script/test_determine_jobs.py index 2c726734fe..44c110b689 100644 --- a/tests/script/test_determine_jobs.py +++ b/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"), [