mirror of
https://github.com/esphome/esphome.git
synced 2026-10-02 17:30:22 +00:00
[espidf] Run cmake and ninja directly instead of through idf.py (#19871)
This commit is contained in:
J. Nick Koston
@@ -0,0 +1,248 @@
|
||||
"""Tests for script/check_idf_py_equivalence.py."""
|
||||
|
||||
from collections.abc import Callable, Iterator
|
||||
import json
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import sys
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent.parent / "script"))
|
||||
|
||||
import check_idf_py_equivalence as guard # noqa: E402
|
||||
|
||||
from esphome.core import CORE # noqa: E402
|
||||
from esphome.espidf import toolchain # noqa: E402
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_core() -> Iterator[None]:
|
||||
"""check() points the global CORE at the tree it inspects."""
|
||||
yield
|
||||
CORE.reset()
|
||||
|
||||
|
||||
def _make_tree(tmp_path: Path) -> Path:
|
||||
tree = tmp_path / "config" / ".esphome" / "build" / "dev"
|
||||
build = tree / "build"
|
||||
for name in (*guard.watched("dev"), *guard.NINJA_LOGS):
|
||||
(tree / name).parent.mkdir(parents=True, exist_ok=True)
|
||||
(tree / name).write_bytes(b"x")
|
||||
(build / "project_description.json").write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"project_name": "dev",
|
||||
"idf_path": "/idf/frameworks/5.5.5",
|
||||
"target": "esp32",
|
||||
}
|
||||
)
|
||||
)
|
||||
(build / ".ninja_log").write_text(
|
||||
"# ninja log v7\n1\t2\t10\tesp-idf/a.obj\t0\n"
|
||||
"1\t2\t10\tbootloader/bootloader.bin\t0\n"
|
||||
)
|
||||
(build / "bootloader" / ".ninja_log").write_text(
|
||||
"# ninja log v7\n1\t2\t10\tbootloader.elf\t0\n"
|
||||
)
|
||||
(tree / "sdkconfig.dev").write_text("")
|
||||
return tree
|
||||
|
||||
|
||||
def _run_check(
|
||||
tree: Path,
|
||||
side_effect: Callable[[list[str]], None] = lambda cmd: None,
|
||||
rc: int = 0,
|
||||
esphome_rcs: tuple[int, int] = (0, 0),
|
||||
) -> tuple[list[str], list[list[str]]]:
|
||||
"""Run check() with idf.py replaced by ``side_effect``; return problems, calls."""
|
||||
calls: list[list[str]] = []
|
||||
|
||||
def run(cmd: list[str], **kwargs: object) -> subprocess.CompletedProcess:
|
||||
calls.append(cmd)
|
||||
side_effect(cmd)
|
||||
return subprocess.CompletedProcess(cmd, rc, "out\n", "err\n")
|
||||
|
||||
with (
|
||||
patch.object(toolchain, "_get_idf_env", return_value={}),
|
||||
patch.object(toolchain, "_get_idf_tool", return_value="/py"),
|
||||
patch.object(toolchain, "_get_idf_path", return_value=Path("/idf")),
|
||||
patch.object(toolchain, "run_reconfigure", return_value=esphome_rcs[0]),
|
||||
patch.object(toolchain, "_run_ninja", return_value=esphome_rcs[1]),
|
||||
patch.object(guard.subprocess, "run", side_effect=run),
|
||||
):
|
||||
return guard.check(tree), calls
|
||||
|
||||
|
||||
def test_check_passes_when_idf_py_changes_nothing(tmp_path: Path) -> None:
|
||||
tree = _make_tree(tmp_path)
|
||||
problems, calls = _run_check(tree)
|
||||
assert problems == []
|
||||
sdkconfig = f"SDKCONFIG={tree / 'sdkconfig.dev'}"
|
||||
assert calls == [
|
||||
["/py", str(Path("/idf/tools/idf.py")), "-D", sdkconfig, "reconfigure"],
|
||||
["/py", str(Path("/idf/tools/idf.py")), "-D", sdkconfig, "build"],
|
||||
]
|
||||
|
||||
|
||||
def test_check_reports_changed_files_and_rebuilt_outputs(tmp_path: Path) -> None:
|
||||
tree = _make_tree(tmp_path)
|
||||
build = tree / "build"
|
||||
|
||||
def drift(cmd: list[str]) -> None:
|
||||
if cmd[-1] == "reconfigure":
|
||||
(build / "CMakeCache.txt").write_text("changed")
|
||||
return
|
||||
# Compacted log. The re-logged bootloader byproduct and a stamp are
|
||||
# not work; a new object mtime is.
|
||||
(build / ".ninja_log").write_text(
|
||||
"# ninja log v7\n3\t4\t20\tesp-idf/a.obj\t0\n"
|
||||
"5\t6\t30\tbootloader/bootloader.bin\t0\n"
|
||||
"5\t6\t30\tbootloader-stamp\t0\n"
|
||||
)
|
||||
# The bootloader sub-build is judged by its own log.
|
||||
(build / "bootloader" / ".ninja_log").write_text(
|
||||
"# ninja log v7\n1\t2\t40\tbootloader.elf\t0\n"
|
||||
)
|
||||
|
||||
problems, _ = _run_check(tree, drift)
|
||||
assert problems == [
|
||||
"idf.py changed build/CMakeCache.txt",
|
||||
"idf.py rebuilt esp-idf/a.obj",
|
||||
"idf.py rebuilt bootloader.elf",
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("after_build", "problem"),
|
||||
[
|
||||
(lambda log: log.unlink(), "missing build/.ninja_log"),
|
||||
(
|
||||
lambda log: log.write_text("# ninja log v7\n"),
|
||||
"no build entries parsed from build/.ninja_log",
|
||||
),
|
||||
(
|
||||
lambda log: log.write_text("# ninja log v7\n1\t2\t10\tesp-idf/b.obj\t0\n"),
|
||||
"idf.py dropped esp-idf/a.obj from build/.ninja_log",
|
||||
),
|
||||
],
|
||||
ids=["log-removed", "log-emptied", "entry-dropped"],
|
||||
)
|
||||
def test_check_reports_a_log_idf_py_left_unusable(
|
||||
tmp_path: Path, after_build: Callable[[Path], None], problem: str
|
||||
) -> None:
|
||||
"""The comparison side gets the same log checks as the baseline."""
|
||||
tree = _make_tree(tmp_path)
|
||||
log = tree / guard.TOP_NINJA_LOG
|
||||
|
||||
def run(cmd: list[str]) -> None:
|
||||
if cmd[-1] == "build":
|
||||
after_build(log)
|
||||
|
||||
problems, _ = _run_check(tree, run)
|
||||
assert problem in problems
|
||||
|
||||
|
||||
def test_check_stops_when_idf_py_fails(tmp_path: Path) -> None:
|
||||
problems, calls = _run_check(_make_tree(tmp_path), rc=2)
|
||||
assert problems == ["idf.py reconfigure failed:\nout\nerr\n"]
|
||||
assert len(calls) == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"remove", ["build/build.ninja", "build/dev.bin", *guard.NINJA_LOGS]
|
||||
)
|
||||
def test_check_fails_when_an_input_is_missing(tmp_path: Path, remove: str) -> None:
|
||||
"""A moved or renamed output must not compare as unchanged."""
|
||||
tree = _make_tree(tmp_path)
|
||||
(tree / remove).unlink()
|
||||
problems, calls = _run_check(tree)
|
||||
assert problems == [f"missing {remove}"]
|
||||
assert calls == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("esphome_rcs", "problem"),
|
||||
[
|
||||
((3, 0), "ESPHome's CMake configure failed with exit code 3"),
|
||||
((0, 4), "ESPHome's ninja build failed with exit code 4"),
|
||||
],
|
||||
ids=["configure", "build"],
|
||||
)
|
||||
def test_check_stops_when_the_esphome_baseline_fails(
|
||||
tmp_path: Path, esphome_rcs: tuple[int, int], problem: str
|
||||
) -> None:
|
||||
"""The baseline is ESPHome's own reconfigure and build."""
|
||||
problems, calls = _run_check(_make_tree(tmp_path), esphome_rcs=esphome_rcs)
|
||||
assert problems == [problem]
|
||||
assert calls == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize("log", guard.NINJA_LOGS)
|
||||
def test_check_fails_when_a_ninja_log_has_no_entries(tmp_path: Path, log: str) -> None:
|
||||
"""A log format change must not leave the rebuild check with nothing to compare."""
|
||||
tree = _make_tree(tmp_path)
|
||||
(tree / log).write_text("# ninja log v99\n1 2 3\n")
|
||||
problems, calls = _run_check(tree)
|
||||
assert problems == [f"no build entries parsed from {log}"]
|
||||
assert calls == []
|
||||
|
||||
|
||||
def test_main_rejects_a_path_that_is_not_a_tree(
|
||||
tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
||||
) -> None:
|
||||
tree = _make_tree(tmp_path / "a")
|
||||
stale = tmp_path / "stale"
|
||||
with (
|
||||
patch.object(sys, "argv", ["check", str(tree), str(stale)]),
|
||||
patch.object(guard, "check", return_value=[]) as mock_check,
|
||||
):
|
||||
assert guard.main() == 1
|
||||
assert f"{stale}: not a configured native ESP-IDF build tree" in (
|
||||
capsys.readouterr().out
|
||||
)
|
||||
mock_check.assert_not_called()
|
||||
|
||||
|
||||
def test_main_without_build_trees(
|
||||
tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
||||
) -> None:
|
||||
with (
|
||||
patch.object(sys, "argv", ["check"]),
|
||||
patch.object(guard, "REPO_ROOT", tmp_path),
|
||||
):
|
||||
assert guard.main() == 1
|
||||
assert "No native ESP-IDF build tree found" in capsys.readouterr().out
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("problems", "rc"), [([], 0), (["idf.py changed x"], 1)])
|
||||
def test_main_reports_each_tree(
|
||||
tmp_path: Path,
|
||||
capsys: pytest.CaptureFixture[str],
|
||||
problems: list[str],
|
||||
rc: int,
|
||||
) -> None:
|
||||
tree = _make_tree(tmp_path)
|
||||
with (
|
||||
patch.object(sys, "argv", ["check", str(tree)]),
|
||||
patch.object(guard, "check", return_value=problems),
|
||||
):
|
||||
assert guard.main() == rc
|
||||
out = capsys.readouterr().out
|
||||
assert f"{tree}: {'DIFFERS' if problems else 'OK'}" in out
|
||||
assert ("no longer matches idf.py" in out) is bool(problems)
|
||||
|
||||
|
||||
def test_main_checks_only_the_first_found_tree(tmp_path: Path) -> None:
|
||||
"""The contract does not depend on the target; one tree per batch is enough."""
|
||||
first = _make_tree(tmp_path / "a")
|
||||
_make_tree(tmp_path / "b")
|
||||
with (
|
||||
patch.object(sys, "argv", ["check"]),
|
||||
patch.object(guard, "REPO_ROOT", tmp_path),
|
||||
patch.object(guard, "DEFAULT_GLOB", "*/config/.esphome/build/*"),
|
||||
patch.object(guard, "check", return_value=[]) as mock_check,
|
||||
):
|
||||
assert guard.main() == 0
|
||||
mock_check.assert_called_once_with(first)
|
||||
@@ -249,11 +249,61 @@ def test_main_all_tests_should_run(
|
||||
# Should contain at least one component (no empty batches)
|
||||
assert len(batch["components"]) > 0
|
||||
assert isinstance(batch["needs_idf"], bool)
|
||||
assert isinstance(batch["check_idf_py"], bool)
|
||||
assert isinstance(batch["needs_nrf"], bool)
|
||||
assert isinstance(batch["needs_arduino8266"], bool)
|
||||
|
||||
|
||||
def test_main_batch_flags_count_variant_tests(
|
||||
def _batch(components: str, idf: bool, check: bool, a8266: bool) -> dict:
|
||||
return {
|
||||
"components": components,
|
||||
"needs_idf": idf,
|
||||
"check_idf_py": check,
|
||||
"needs_nrf": False,
|
||||
"needs_arduino8266": a8266,
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("batches", "component_platforms", "validate_only", "expected"),
|
||||
[
|
||||
# The compile stage builds test-<variant>.<platform>.yaml too, so a
|
||||
# component tested on esp8266 only by a variant still needs the toolchain.
|
||||
(
|
||||
[["safe_mode"]],
|
||||
{"safe_mode": {"esp8266-ard"}},
|
||||
set(),
|
||||
[_batch("safe_mode", idf=False, check=False, a8266=True)],
|
||||
),
|
||||
# Only the first esp32 batch runs the idf.py equivalence check.
|
||||
(
|
||||
[["a"], ["b"], ["c"]],
|
||||
{"a": {"esp8266-ard"}, "b": {"esp32-idf"}, "c": {"esp32-c3-idf"}},
|
||||
set(),
|
||||
[
|
||||
_batch("a", idf=False, check=False, a8266=True),
|
||||
_batch("b", idf=True, check=True, a8266=False),
|
||||
_batch("c", idf=True, check=False, a8266=False),
|
||||
],
|
||||
),
|
||||
# A validate-only batch never compiles, so the next esp32 batch runs it.
|
||||
(
|
||||
[["b"], ["c"]],
|
||||
{"b": {"esp32-idf"}, "c": {"esp32-c3-idf"}},
|
||||
{"b"},
|
||||
[
|
||||
_batch("b", idf=True, check=False, a8266=False),
|
||||
_batch("c", idf=True, check=True, a8266=False),
|
||||
],
|
||||
),
|
||||
],
|
||||
ids=["variant", "idf-check-once", "idf-check-skips-validate-only"],
|
||||
)
|
||||
def test_main_batch_flags(
|
||||
batches: list[list[str]],
|
||||
component_platforms: dict[str, set[str]],
|
||||
validate_only: set[str],
|
||||
expected: list[dict],
|
||||
mock_determine_integration_tests: Mock,
|
||||
mock_should_run_clang_tidy: Mock,
|
||||
mock_should_run_clang_format: Mock,
|
||||
@@ -267,8 +317,7 @@ def test_main_batch_flags_count_variant_tests(
|
||||
capsys: pytest.CaptureFixture[str],
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""The compile stage builds test-<variant>.<platform>.yaml too, so a
|
||||
component tested on esp8266 only by a variant still needs the toolchain."""
|
||||
"""Each batch carries the toolchain flags its test platforms need."""
|
||||
monkeypatch.delenv("GITHUB_ACTIONS", raising=False)
|
||||
mock_determine_integration_tests.return_value = (False, [])
|
||||
mock_should_run_clang_tidy.return_value = False
|
||||
@@ -282,13 +331,20 @@ def test_main_batch_flags_count_variant_tests(
|
||||
mock_changed_files.return_value = ["esphome/components/safe_mode/__init__.py"]
|
||||
|
||||
def platforms(component: str, *, base_only: bool = True) -> set[str]:
|
||||
return set() if base_only else {"esp8266-ard"}
|
||||
return set() if base_only else component_platforms[component]
|
||||
|
||||
with (
|
||||
patch("sys.argv", ["determine-jobs.py"]),
|
||||
patch.object(determine_jobs, "_is_clang_tidy_full_scan", return_value=False),
|
||||
patch.object(
|
||||
determine_jobs, "get_changed_components", return_value=["safe_mode"]
|
||||
determine_jobs,
|
||||
"get_changed_components",
|
||||
return_value=[c for batch in batches for c in batch],
|
||||
),
|
||||
patch.object(
|
||||
determine_jobs,
|
||||
"_component_change_is_validate_only",
|
||||
side_effect=lambda component, changed: component in validate_only,
|
||||
),
|
||||
patch.object(
|
||||
determine_jobs,
|
||||
@@ -298,7 +354,7 @@ def test_main_batch_flags_count_variant_tests(
|
||||
patch.object(
|
||||
determine_jobs,
|
||||
"get_components_with_dependencies",
|
||||
return_value=["safe_mode"],
|
||||
return_value=[c for batch in batches for c in batch],
|
||||
),
|
||||
patch.object(determine_jobs, "_component_has_tests", return_value=True),
|
||||
patch.object(
|
||||
@@ -307,9 +363,7 @@ def test_main_batch_flags_count_variant_tests(
|
||||
return_value={"should_run": "false"},
|
||||
),
|
||||
patch.object(
|
||||
determine_jobs,
|
||||
"create_intelligent_batches",
|
||||
return_value=([["safe_mode"]], {}),
|
||||
determine_jobs, "create_intelligent_batches", return_value=(batches, {})
|
||||
),
|
||||
patch.object(
|
||||
determine_jobs, "get_component_test_platforms", side_effect=platforms
|
||||
@@ -318,14 +372,7 @@ def test_main_batch_flags_count_variant_tests(
|
||||
determine_jobs.main()
|
||||
|
||||
output = json.loads(capsys.readouterr().out)
|
||||
assert output["component_test_batches"] == [
|
||||
{
|
||||
"components": "safe_mode",
|
||||
"needs_idf": False,
|
||||
"needs_nrf": False,
|
||||
"needs_arduino8266": True,
|
||||
}
|
||||
]
|
||||
assert output["component_test_batches"] == expected
|
||||
|
||||
|
||||
def test_main_no_tests_should_run(
|
||||
@@ -1231,8 +1278,10 @@ def test_should_run_esp32_platformio_with_branch() -> None:
|
||||
@pytest.mark.parametrize(
|
||||
("changed_files", "expected"),
|
||||
[
|
||||
# ESP-IDF runner / framework / build generator -> trigger
|
||||
(["esphome/espidf/runner.py"], True),
|
||||
# ESP-IDF toolchain / framework / build generator -> trigger
|
||||
(["esphome/espidf/toolchain.py"], True),
|
||||
(["esphome/build_helpers/tool_runner.py"], True),
|
||||
(["script/check_idf_py_equivalence.py"], True),
|
||||
(["esphome/espidf/framework.py"], True),
|
||||
(["esphome/build_gen/espidf.py"], True),
|
||||
# Shared native-build modules the IDF build imports -> trigger
|
||||
@@ -1250,7 +1299,7 @@ def test_should_run_esp32_platformio_with_branch() -> None:
|
||||
],
|
||||
)
|
||||
def test_esp_idf_infra_changed(changed_files: list[str], expected: bool) -> None:
|
||||
"""ESP-IDF build/runner infra paths are detected; other paths are not."""
|
||||
"""ESP-IDF build infra paths are detected; other paths are not."""
|
||||
assert determine_jobs._esp_idf_infra_changed(changed_files) is expected
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,338 @@
|
||||
"""Tests for esphome.build_helpers.tool_runner."""
|
||||
|
||||
# pylint: disable=protected-access
|
||||
|
||||
import io
|
||||
import os
|
||||
from pathlib import Path
|
||||
import sys
|
||||
import threading
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from esphome.build_helpers import tool_runner
|
||||
from esphome.build_helpers.tool_runner import (
|
||||
UTF8_CODEPAGE,
|
||||
ToolOutput,
|
||||
Utf8Console,
|
||||
run_build_tool,
|
||||
)
|
||||
from esphome.core import CORE
|
||||
|
||||
FILTER = [r"-- Component paths:", r"\s*$"]
|
||||
|
||||
|
||||
def _child(code: str) -> list[str]:
|
||||
return [sys.executable, "-c", code]
|
||||
|
||||
|
||||
def _run(
|
||||
capsys: pytest.CaptureFixture[str],
|
||||
tmp_path: Path,
|
||||
code: str,
|
||||
**kwargs: object,
|
||||
) -> tuple[int, str]:
|
||||
rc = run_build_tool(_child(code), cwd=tmp_path, env=dict(os.environ), **kwargs)
|
||||
# The child's print() ends lines with \r\n on Windows.
|
||||
return rc, capsys.readouterr().out.replace("\r\n", "\n")
|
||||
|
||||
|
||||
def test_run_build_tool_filters_and_returns_exit_code(
|
||||
capsys: pytest.CaptureFixture[str], tmp_path: Path
|
||||
) -> None:
|
||||
rc, out = _run(
|
||||
capsys,
|
||||
tmp_path,
|
||||
"import sys\n"
|
||||
"print('-- Component paths: /a /b')\n"
|
||||
"print('')\n"
|
||||
"print('Compiling main.cpp')\n"
|
||||
"sys.exit(3)",
|
||||
filter_lines=FILTER,
|
||||
)
|
||||
assert rc == 3
|
||||
assert out == "Compiling main.cpp\n"
|
||||
|
||||
|
||||
def test_run_build_tool_passes_everything_without_filter(
|
||||
capsys: pytest.CaptureFixture[str], tmp_path: Path
|
||||
) -> None:
|
||||
rc, out = _run(capsys, tmp_path, "print('-- Component paths: /a')\nprint('')")
|
||||
assert rc == 0
|
||||
assert out == "-- Component paths: /a\n\n"
|
||||
|
||||
|
||||
def test_run_build_tool_merges_stderr_and_runs_in_cwd(
|
||||
capsys: pytest.CaptureFixture[str], tmp_path: Path
|
||||
) -> None:
|
||||
_, out = _run(
|
||||
capsys,
|
||||
tmp_path,
|
||||
"import os, sys\nprint(os.getcwd(), flush=True)\nprint('oops', file=sys.stderr)",
|
||||
filter_lines=FILTER,
|
||||
)
|
||||
assert out.splitlines() == [os.path.realpath(tmp_path), "oops"]
|
||||
|
||||
|
||||
def test_run_build_tool_drains_a_partial_line(
|
||||
capsys: pytest.CaptureFixture[str], tmp_path: Path
|
||||
) -> None:
|
||||
"""A tool that dies mid line still shows that line, terminated."""
|
||||
rc, out = _run(
|
||||
capsys,
|
||||
tmp_path,
|
||||
"import sys\nsys.stdout.write('ld returned 1 exit status')\nsys.exit(1)",
|
||||
filter_lines=FILTER,
|
||||
)
|
||||
assert rc == 1
|
||||
assert out == "ld returned 1 exit status\n"
|
||||
|
||||
|
||||
def test_run_build_tool_replaces_invalid_utf8(
|
||||
capsys: pytest.CaptureFixture[str], tmp_path: Path
|
||||
) -> None:
|
||||
_, out = _run(
|
||||
capsys,
|
||||
tmp_path,
|
||||
# A multi-byte character split across writes must survive too.
|
||||
"import sys\nb = sys.stdout.buffer\n"
|
||||
"b.write(b'\\xc3'); b.flush(); b.write(b'\\xa9 ok\\n\\xff bad\\n')",
|
||||
filter_lines=FILTER,
|
||||
)
|
||||
assert out == "é ok\n� bad\n"
|
||||
|
||||
|
||||
def test_run_build_tool_streams_before_the_tool_exits(tmp_path: Path) -> None:
|
||||
"""Output must reach the user while the tool runs, not when it ends."""
|
||||
seen = threading.Event()
|
||||
lines: list[str] = []
|
||||
|
||||
class _Out(io.StringIO):
|
||||
def write(self, s: str) -> int:
|
||||
lines.append(s)
|
||||
if "first" in s:
|
||||
seen.set()
|
||||
return len(s)
|
||||
|
||||
release = tmp_path / "release"
|
||||
code = (
|
||||
"import os, time\nprint('first', flush=True)\n"
|
||||
f"while not os.path.exists({str(release)!r}): time.sleep(0.02)\n"
|
||||
)
|
||||
with patch.object(tool_runner.sys, "stdout", _Out()):
|
||||
thread = threading.Thread(
|
||||
target=run_build_tool,
|
||||
args=(_child(code),),
|
||||
kwargs={"cwd": tmp_path, "env": dict(os.environ), "filter_lines": FILTER},
|
||||
)
|
||||
thread.start()
|
||||
try:
|
||||
assert seen.wait(30)
|
||||
finally:
|
||||
release.touch()
|
||||
thread.join(30)
|
||||
assert "first" in "".join(lines)
|
||||
|
||||
|
||||
def _tool_output(progress: bool = True) -> tuple[ToolOutput, io.StringIO]:
|
||||
out = io.StringIO()
|
||||
return ToolOutput(out, FILTER, progress), out
|
||||
|
||||
|
||||
def test_tool_output_collapses_progress_lines() -> None:
|
||||
"""Ninja status lines overwrite each other, as idf.py shows them."""
|
||||
output, out = _tool_output()
|
||||
output.write("[1/2] Building a.o\n[2/2] Linking app\nwarning: x\n")
|
||||
assert out.getvalue() == (
|
||||
"\r[1/2] Building a.o\x1b[K\r[2/2] Linking app\x1b[K"
|
||||
+ os.linesep
|
||||
+ "warning: x\n"
|
||||
)
|
||||
|
||||
|
||||
def test_tool_output_drain_ends_a_progress_line() -> None:
|
||||
output, out = _tool_output()
|
||||
output.write("[1/1] Linking app\n")
|
||||
output.drain()
|
||||
assert out.getvalue() == "\r[1/1] Linking app\x1b[K" + os.linesep
|
||||
# Nothing is pending any more.
|
||||
output.drain()
|
||||
assert out.getvalue().count(os.linesep) == 1
|
||||
|
||||
|
||||
def test_tool_output_drain_survives_a_broken_stream() -> None:
|
||||
output, out = _tool_output()
|
||||
output.write("[1/1] Linking app\n")
|
||||
out.close()
|
||||
output.drain()
|
||||
|
||||
|
||||
def test_tool_output_filters_a_matching_progress_line() -> None:
|
||||
out = io.StringIO()
|
||||
output = ToolOutput(out, [r"\[1/2\]"], True)
|
||||
output.write("[1/2] hidden\n[2/2] shown\n")
|
||||
assert out.getvalue() == "\r[2/2] shown\x1b[K"
|
||||
|
||||
|
||||
def test_tool_output_without_progress_keeps_status_lines() -> None:
|
||||
output, out = _tool_output(progress=False)
|
||||
output.write("[1/2] Building a.o\n")
|
||||
assert out.getvalue() == "[1/2] Building a.o\n"
|
||||
|
||||
|
||||
def test_tool_output_escapes_colors_for_the_dashboard() -> None:
|
||||
CORE.dashboard = True
|
||||
output, out = _tool_output()
|
||||
output.write("[1/1] \x1b[1mLinking\x1b[0m\n")
|
||||
assert "\x1b" not in out.getvalue()
|
||||
assert "\\033[K" in out.getvalue()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("width", "text", "expected"),
|
||||
[
|
||||
(0, "x" * 50, "x" * 50),
|
||||
(3, "abcdef", "..."),
|
||||
(20, "short", "short"),
|
||||
(11, "abcdefghijklmnop", "abcd...mnop"),
|
||||
],
|
||||
ids=["unknown", "tiny", "fits", "elided"],
|
||||
)
|
||||
def test_fit_terminal(width: int, text: str, expected: str) -> None:
|
||||
with patch.object(
|
||||
tool_runner.shutil,
|
||||
"get_terminal_size",
|
||||
return_value=os.terminal_size((width, 24)),
|
||||
):
|
||||
assert tool_runner._fit_terminal(text) == expected
|
||||
|
||||
|
||||
class _FakeKernel32:
|
||||
"""Stand-in for the Windows kernel32 console code page calls."""
|
||||
|
||||
def __init__(self, input_cp: int, output_cp: int) -> None:
|
||||
self.input_cp = input_cp
|
||||
self.output_cp = output_cp
|
||||
self.calls: list[tuple[str, int]] = []
|
||||
|
||||
def GetConsoleCP(self) -> int: # noqa: N802
|
||||
return self.input_cp
|
||||
|
||||
def GetConsoleOutputCP(self) -> int: # noqa: N802
|
||||
return self.output_cp
|
||||
|
||||
def SetConsoleCP(self, codepage: int) -> int: # noqa: N802
|
||||
self.calls.append(("SetConsoleCP", codepage))
|
||||
self.input_cp = codepage
|
||||
return 1
|
||||
|
||||
def SetConsoleOutputCP(self, codepage: int) -> int: # noqa: N802
|
||||
self.calls.append(("SetConsoleOutputCP", codepage))
|
||||
self.output_cp = codepage
|
||||
return 1
|
||||
|
||||
|
||||
def test_run_build_tool_switches_the_console_to_utf8(
|
||||
capsys: pytest.CaptureFixture[str], tmp_path: Path
|
||||
) -> None:
|
||||
"""An attached console runs the tool on UTF-8 and is then put back."""
|
||||
kernel32 = _FakeKernel32(850, 850)
|
||||
with patch.object(tool_runner, "_get_kernel32", return_value=kernel32):
|
||||
_run(capsys, tmp_path, "print('x')")
|
||||
assert kernel32.calls == [
|
||||
("SetConsoleCP", UTF8_CODEPAGE),
|
||||
("SetConsoleOutputCP", UTF8_CODEPAGE),
|
||||
("SetConsoleCP", 850),
|
||||
("SetConsoleOutputCP", 850),
|
||||
]
|
||||
|
||||
|
||||
def test_utf8_console_restores_after_an_error() -> None:
|
||||
kernel32 = _FakeKernel32(437, 437)
|
||||
with pytest.raises(RuntimeError), Utf8Console(kernel32):
|
||||
raise RuntimeError
|
||||
assert (kernel32.input_cp, kernel32.output_cp) == (437, 437)
|
||||
|
||||
|
||||
def test_utf8_console_restores_when_the_switch_fails_part_way() -> None:
|
||||
kernel32 = _FakeKernel32(850, 850)
|
||||
|
||||
def _refuse(codepage: int) -> int:
|
||||
kernel32.calls.append(("SetConsoleOutputCP", codepage))
|
||||
return 0
|
||||
|
||||
kernel32.SetConsoleOutputCP = _refuse # type: ignore[method-assign]
|
||||
with Utf8Console(kernel32):
|
||||
pass
|
||||
assert kernel32.input_cp == 850
|
||||
assert kernel32.calls[-2:] == [("SetConsoleCP", 850), ("SetConsoleOutputCP", 850)]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"codepages", [(0, 0), (UTF8_CODEPAGE, UTF8_CODEPAGE)], ids=["none", "utf8"]
|
||||
)
|
||||
def test_utf8_console_leaves_the_console_alone(codepages: tuple[int, int]) -> None:
|
||||
"""No console, or one already on UTF-8 (an overlapping build), is untouched."""
|
||||
kernel32 = _FakeKernel32(*codepages)
|
||||
with Utf8Console(kernel32):
|
||||
pass
|
||||
assert kernel32.calls == []
|
||||
|
||||
|
||||
def test_utf8_console_without_kernel32() -> None:
|
||||
with Utf8Console(None):
|
||||
pass
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.platform == "win32", reason="kernel32 exists on Windows")
|
||||
def test_get_kernel32_is_none_off_windows() -> None:
|
||||
assert tool_runner._get_kernel32() is None
|
||||
|
||||
|
||||
def test_tool_output_collapses_progress_without_a_filter() -> None:
|
||||
"""Progress mode splits lines on its own; it does not need a filter."""
|
||||
out = io.StringIO()
|
||||
output = ToolOutput(out, None, True)
|
||||
output.write("[1/1] Linking app\ndone\n")
|
||||
assert out.getvalue() == "\r[1/1] Linking app\x1b[K" + os.linesep + "done\n"
|
||||
|
||||
|
||||
def test_run_build_tool_flushes_a_truncated_character(
|
||||
capsys: pytest.CaptureFixture[str], tmp_path: Path
|
||||
) -> None:
|
||||
"""Output that ends inside a multi-byte character still shows up."""
|
||||
_, out = _run(
|
||||
capsys,
|
||||
tmp_path,
|
||||
"import sys\nsys.stdout.buffer.write(b'end \\xc3')",
|
||||
filter_lines=FILTER,
|
||||
)
|
||||
assert out == "end �\n"
|
||||
|
||||
|
||||
def test_get_kernel32_loads_it_on_windows(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
kernel32 = object()
|
||||
fake_ctypes = SimpleNamespace(windll=SimpleNamespace(kernel32=kernel32))
|
||||
monkeypatch.setattr(tool_runner.sys, "platform", "win32")
|
||||
monkeypatch.setitem(sys.modules, "ctypes", fake_ctypes)
|
||||
assert tool_runner._get_kernel32() is kernel32
|
||||
|
||||
|
||||
def test_run_build_tool_logs_the_unfiltered_output(
|
||||
capsys: pytest.CaptureFixture[str], tmp_path: Path
|
||||
) -> None:
|
||||
"""The log gets every line, filtered or not, for idf.py's hint matcher."""
|
||||
log = tmp_path / "log" / "ninja_all_output.log"
|
||||
_, out = _run(
|
||||
capsys,
|
||||
tmp_path,
|
||||
"print('-- Component paths: /a')\nprint('\\x1b[1merror:\\x1b[0m boom')",
|
||||
filter_lines=FILTER,
|
||||
log_path=log,
|
||||
)
|
||||
assert out == "\x1b[1merror:\x1b[0m boom\n"
|
||||
assert log.read_text(encoding="utf-8").replace("\r\n", "\n") == (
|
||||
"-- Component paths: /a\nerror: boom\n"
|
||||
)
|
||||
@@ -1,11 +0,0 @@
|
||||
"""Leave a partial line behind and then close the stream under the runner.
|
||||
|
||||
Run through ``esphome/espidf/runner.py`` by test_espidf_runner.py. Draining
|
||||
cannot work here; the point is that the failure is reported rather than
|
||||
raised out of the runner's cleanup, where it would bury the exit code.
|
||||
"""
|
||||
|
||||
import sys
|
||||
|
||||
sys.stdout.write("partial before close")
|
||||
sys.stdout.close()
|
||||
@@ -1,11 +0,0 @@
|
||||
"""Die part way through a line, the way a build that blows up does.
|
||||
|
||||
Run through ``esphome/espidf/runner.py`` by test_espidf_runner.py. The
|
||||
message has no trailing newline, so the runner's shim is holding it when
|
||||
the process exits; nothing else will ever come to release it.
|
||||
"""
|
||||
|
||||
import sys
|
||||
|
||||
sys.stdout.write("FATAL: ld returned 1 exit status")
|
||||
sys.exit(2)
|
||||
@@ -1,15 +0,0 @@
|
||||
"""Write a mix of noisy and useful build lines, without flushing.
|
||||
|
||||
Run through ``esphome/espidf/runner.py`` by test_espidf_runner.py. The
|
||||
runner's shim owns both the filtering and the flushing, so this script
|
||||
only writes.
|
||||
"""
|
||||
|
||||
import sys
|
||||
|
||||
sys.stdout.write("Project build complete.\n")
|
||||
sys.stdout.write("Compiling main.cpp\n")
|
||||
sys.stdout.write("-- Component paths: /a /b /c\n")
|
||||
sys.stdout.write("[2/9] Building C object\n")
|
||||
# No terminator, so the shim has to hold this one back.
|
||||
sys.stdout.write("still going")
|
||||
@@ -1,12 +0,0 @@
|
||||
"""Write a form feed part way through the output.
|
||||
|
||||
Run through ``esphome/espidf/runner.py`` by test_espidf_runner.py. A form
|
||||
feed is not a line terminator here, so everything written must still come
|
||||
out, including the complete lines that follow it.
|
||||
"""
|
||||
|
||||
import sys
|
||||
|
||||
sys.stdout.write("Compiling main.cpp\n")
|
||||
sys.stdout.write("page one\x0cpage two\n")
|
||||
sys.stdout.write("[2/9] Building C object\n")
|
||||
@@ -1,10 +0,0 @@
|
||||
"""End on an unterminated line that the filter is supposed to drop.
|
||||
|
||||
Run through ``esphome/espidf/runner.py`` by test_espidf_runner.py, to
|
||||
check that releasing a held-back line still applies the filter.
|
||||
"""
|
||||
|
||||
import sys
|
||||
|
||||
sys.stdout.write("Compiling main.cpp\n")
|
||||
sys.stdout.write("Project build complete.")
|
||||
@@ -1,14 +0,0 @@
|
||||
"""Print one line, then stay alive so the caller can prove it streamed.
|
||||
|
||||
Run through ``esphome/espidf/runner.py`` by test_espidf_runner.py. The
|
||||
runner wraps stdout in its filtering shim, so this script deliberately
|
||||
does not flush: the shim has to do it. The long sleep keeps the process
|
||||
running, so anything the caller reads must have arrived while the build
|
||||
was still going rather than at exit.
|
||||
"""
|
||||
|
||||
import sys
|
||||
import time
|
||||
|
||||
sys.stdout.write("Compiling main.cpp\n")
|
||||
time.sleep(60)
|
||||
@@ -1,5 +1,7 @@
|
||||
"""Tests for esphome.espidf.clang_tidy tidy-project generation."""
|
||||
|
||||
# pylint: disable=protected-access
|
||||
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
@@ -99,3 +101,39 @@ def test_idedata_from_tidy_project_missing_tu_raises(tmp_path) -> None:
|
||||
compile_commands.write_text(json.dumps([]))
|
||||
with pytest.raises(RuntimeError, match="tidy.cpp not found"):
|
||||
clang_tidy._idedata_from_tidy_project(compile_commands)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("reconfigure_rcs", "error"),
|
||||
[
|
||||
((1,), "ESP-IDF CMake configure \\(discovery\\) failed"),
|
||||
((0, 1), "ESP-IDF CMake configure failed"),
|
||||
((0, 0), None),
|
||||
],
|
||||
ids=["discovery", "full", "ok"],
|
||||
)
|
||||
def test_generate_compile_commands_configures_twice(
|
||||
tmp_path: Path, reconfigure_rcs: tuple[int, ...], error: str | None
|
||||
) -> None:
|
||||
"""Discovery configure, then a configure requiring what it found."""
|
||||
with (
|
||||
patch.object(clang_tidy, "_setup_core"),
|
||||
patch.object(clang_tidy, "_convert_pio_libs", return_value={}),
|
||||
patch.object(clang_tidy, "_write_tidy_project") as mock_write,
|
||||
patch("esphome.espidf.toolchain.run_reconfigure", side_effect=reconfigure_rcs),
|
||||
patch(
|
||||
"esphome.build_gen.espidf.get_available_components",
|
||||
return_value=["lwip", "esp_timer"],
|
||||
),
|
||||
):
|
||||
if error:
|
||||
with pytest.raises(RuntimeError, match=error):
|
||||
clang_tidy._generate_compile_commands(
|
||||
tmp_path, _settings(), tmp_path / "platformio.ini"
|
||||
)
|
||||
return
|
||||
result = clang_tidy._generate_compile_commands(
|
||||
tmp_path, _settings(), tmp_path / "platformio.ini"
|
||||
)
|
||||
assert result == tmp_path / "build" / "compile_commands.json"
|
||||
assert mock_write.call_args_list[1].args[1] == ["esp_timer", "lwip"]
|
||||
|
||||
@@ -1627,8 +1627,8 @@ def test_ccache_env_opt_out_via_env(tmp_path: Path) -> None:
|
||||
# short-circuits before build_path is needed.
|
||||
p1, p2, p3 = _ccache_patches(tmp_path, "/usr/bin/ccache", None)
|
||||
with patch.dict("os.environ", {"IDF_CCACHE_ENABLE": "0"}, clear=True), p1, p2, p3:
|
||||
# The canonical off spelling is exported: the raw value is inherited
|
||||
# by idf.py, where a spelling like "disable" would read as truthy
|
||||
# The canonical off spelling is exported, so every reader of the
|
||||
# env sees the same answer
|
||||
assert _ccache_env() == {"IDF_CCACHE_ENABLE": "0"}
|
||||
|
||||
|
||||
@@ -1636,7 +1636,7 @@ def test_ccache_env_opt_in_without_binary(
|
||||
tmp_path: Path, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
# Explicit IDF_CCACHE_ENABLE=1 forces it on; without a usable binary
|
||||
# idf.py silently skips ccache, so this branch must say so out loud.
|
||||
# IDF's CMake silently skips ccache, so this branch must say so out loud.
|
||||
p1, p2, p3 = _ccache_patches(tmp_path, None, tmp_path / "build")
|
||||
with patch.dict("os.environ", {"IDF_CCACHE_ENABLE": "1"}, clear=True), p1, p2, p3:
|
||||
env = _ccache_env()
|
||||
@@ -1669,7 +1669,7 @@ def test_ccache_env_opt_in_with_working_binary(
|
||||
def test_ccache_env_opt_in_with_rejected_binary(
|
||||
tmp_path: Path, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
# Forced on with a present-but-rejected binary: idf.py does its own
|
||||
# Forced on with a present-but-rejected binary: IDF's CMake does its own
|
||||
# PATH lookup and uses it anyway; the warning must say so, not claim
|
||||
# the build runs without ccache.
|
||||
# A present but non-executable file: the real probe fails and logs
|
||||
@@ -1686,7 +1686,7 @@ def test_ccache_env_opt_in_with_rejected_binary(
|
||||
):
|
||||
env = _ccache_env()
|
||||
assert env["IDF_CCACHE_ENABLE"] == "1"
|
||||
assert "idf.py will use it anyway" in caplog.text
|
||||
assert "the build will use it anyway" in caplog.text
|
||||
# Exactly one story: the resolver's contradictory "compiling without
|
||||
# ccache" must not precede it
|
||||
assert "compiling without ccache" not in caplog.text
|
||||
@@ -1715,7 +1715,7 @@ def test_ccache_env_idf_knob_unrecognized_warns_and_defers(
|
||||
tmp_path: Path, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""An unparsable IDF_CCACHE_ENABLE warns, defers to the shared resolver,
|
||||
and is not forwarded to idf.py as truthy."""
|
||||
and is not forwarded to the build as truthy."""
|
||||
p1, p2, p3 = _ccache_patches(tmp_path, "/usr/bin/ccache", tmp_path / "build")
|
||||
env_vars = {"IDF_CCACHE_ENABLE": "enabled"}
|
||||
with patch.dict("os.environ", env_vars, clear=True), p1, p2, p3:
|
||||
|
||||
@@ -1,315 +0,0 @@
|
||||
"""Tests for esphome.espidf.runner."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import os
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
|
||||
import pytest
|
||||
|
||||
from esphome.espidf import runner
|
||||
|
||||
# A flushing runner delivers the first line in well under a second; this is
|
||||
# only ever waited out when the shim has gone back to buffering, so keep it
|
||||
# just long enough to cover interpreter startup on a loaded CI machine.
|
||||
FIRST_LINE_TIMEOUT = 10.0
|
||||
|
||||
|
||||
def _prepare_main(
|
||||
monkeypatch: pytest.MonkeyPatch, probe: Path, *args: str
|
||||
) -> tuple[io.BytesIO, io.TextIOWrapper]:
|
||||
"""Point ``runner.main()`` at *probe* with a buffered fake stdout.
|
||||
|
||||
``main`` rewrites ``sys.path``, ``sys.argv``, both std streams and
|
||||
``os.get_terminal_size``; every one of those is monkeypatched so it is
|
||||
put back afterwards. The fake stdout is block buffered like a pipe, so
|
||||
the caller can tell whether the shim flushed. The wrapper comes back with
|
||||
the buffer because dropping it would close the buffer underneath us.
|
||||
"""
|
||||
buf = io.BytesIO()
|
||||
stream = io.TextIOWrapper(buf, encoding="utf-8", newline="\n", line_buffering=False)
|
||||
|
||||
monkeypatch.setattr(sys, "path", list(sys.path))
|
||||
monkeypatch.setattr(sys, "argv", ["runner.py", str(probe), *args])
|
||||
monkeypatch.setattr(sys, "stdout", stream)
|
||||
monkeypatch.setattr(sys, "stderr", stream)
|
||||
monkeypatch.setattr(os, "get_terminal_size", os.get_terminal_size)
|
||||
|
||||
return buf, stream
|
||||
|
||||
|
||||
def _run_main(
|
||||
monkeypatch: pytest.MonkeyPatch, probe: Path, *args: str
|
||||
) -> tuple[io.BytesIO, io.TextIOWrapper]:
|
||||
"""Run ``runner.main()`` against *probe* and expect a clean exit."""
|
||||
buf, stream = _prepare_main(monkeypatch, probe, *args)
|
||||
assert runner.main() == 0
|
||||
return buf, stream
|
||||
|
||||
|
||||
def test_main_filters_noise_and_flushes_each_write(
|
||||
monkeypatch: pytest.MonkeyPatch, fixture_path: Path
|
||||
) -> None:
|
||||
"""Useful lines reach the stream right away; noisy ones are dropped."""
|
||||
buf, _stream = _run_main(
|
||||
monkeypatch, fixture_path / "espidf" / "filtering_probe.py"
|
||||
)
|
||||
|
||||
# Read before any flush of our own: the shim has to have flushed.
|
||||
output = buf.getvalue().decode("utf-8")
|
||||
|
||||
assert "Compiling main.cpp\n" in output
|
||||
assert "[2/9] Building C object\n" in output
|
||||
# Matched by FILTER_IDF_LINES, so they never leave the runner.
|
||||
assert "Project build complete." not in output
|
||||
assert "-- Component paths:" not in output
|
||||
# Held back until the end because no terminator arrived.
|
||||
assert output.endswith("still going\n")
|
||||
|
||||
|
||||
def test_main_keeps_output_after_a_form_feed(
|
||||
monkeypatch: pytest.MonkeyPatch, fixture_path: Path
|
||||
) -> None:
|
||||
"""A form feed is text, not a line break, so nothing after it is lost."""
|
||||
buf, _stream = _run_main(monkeypatch, fixture_path / "espidf" / "formfeed_probe.py")
|
||||
|
||||
assert buf.getvalue().decode("utf-8") == (
|
||||
"Compiling main.cpp\npage one\x0cpage two\n[2/9] Building C object\n"
|
||||
)
|
||||
|
||||
|
||||
def test_main_drains_a_partial_line_when_the_build_dies(
|
||||
monkeypatch: pytest.MonkeyPatch, fixture_path: Path
|
||||
) -> None:
|
||||
"""A build that stops mid line must still show that line.
|
||||
|
||||
This is the whole point of draining: the message explaining why the
|
||||
build failed is exactly the one most likely to arrive without a
|
||||
trailing newline.
|
||||
"""
|
||||
buf, _stream = _prepare_main(
|
||||
monkeypatch, fixture_path / "espidf" / "crashing_probe.py"
|
||||
)
|
||||
|
||||
with pytest.raises(SystemExit) as excinfo:
|
||||
runner.main()
|
||||
|
||||
assert excinfo.value.code == 2
|
||||
assert buf.getvalue().decode("utf-8") == "FATAL: ld returned 1 exit status\n"
|
||||
|
||||
|
||||
def test_main_reports_rather_than_raises_when_draining_fails(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
fixture_path: Path,
|
||||
capfd: pytest.CaptureFixture[str],
|
||||
) -> None:
|
||||
"""A stream that closed under us must not crash the runner's cleanup.
|
||||
|
||||
The drain runs from a ``finally``, so an exception there would replace
|
||||
whatever exit code the build was carrying back.
|
||||
"""
|
||||
_prepare_main(monkeypatch, fixture_path / "espidf" / "closing_probe.py")
|
||||
|
||||
assert runner.main() == 0
|
||||
reported = capfd.readouterr().err
|
||||
assert "Could not write out remaining output" in reported
|
||||
# The held line has to come along; the stream it was meant for is gone.
|
||||
assert "partial before close" in reported
|
||||
|
||||
|
||||
def test_main_survives_a_drain_failure_with_nowhere_to_report_it(
|
||||
monkeypatch: pytest.MonkeyPatch, fixture_path: Path
|
||||
) -> None:
|
||||
"""With no real stderr to report to, cleanup still must not raise.
|
||||
|
||||
``sys.__stderr__`` is None on some interpreters, and ``print(file=None)``
|
||||
falls back to ``sys.stdout``, which here is the shim wrapping the stream
|
||||
that just failed.
|
||||
"""
|
||||
monkeypatch.setattr(sys, "__stderr__", None)
|
||||
_prepare_main(monkeypatch, fixture_path / "espidf" / "closing_probe.py")
|
||||
|
||||
assert runner.main() == 0
|
||||
|
||||
|
||||
def test_main_still_filters_a_drained_partial_line(
|
||||
monkeypatch: pytest.MonkeyPatch, fixture_path: Path
|
||||
) -> None:
|
||||
"""Releasing a held line does not smuggle noise past the filter."""
|
||||
buf, _stream = _run_main(
|
||||
monkeypatch, fixture_path / "espidf" / "partial_noise_probe.py"
|
||||
)
|
||||
|
||||
assert buf.getvalue().decode("utf-8") == "Compiling main.cpp\n"
|
||||
|
||||
|
||||
def test_main_keeps_everything_in_verbose_mode(
|
||||
monkeypatch: pytest.MonkeyPatch, fixture_path: Path
|
||||
) -> None:
|
||||
"""``-v`` turns the filter off so the noisy lines survive."""
|
||||
buf, _stream = _run_main(
|
||||
monkeypatch, fixture_path / "espidf" / "filtering_probe.py", "-v"
|
||||
)
|
||||
|
||||
output = buf.getvalue().decode("utf-8")
|
||||
|
||||
assert "Project build complete.\n" in output
|
||||
assert "-- Component paths: /a /b /c\n" in output
|
||||
# With no filter there is no line buffering, so the partial line goes
|
||||
# straight through as well.
|
||||
assert output.endswith("still going")
|
||||
|
||||
|
||||
def test_runner_streams_output_before_the_build_finishes(
|
||||
fixture_path: Path, probe_env: dict[str, str]
|
||||
) -> None:
|
||||
"""The runner must flush, or a dashboard build looks frozen.
|
||||
|
||||
``toolchain.py`` spawns the runner as a plain script with no ``-u``, and
|
||||
hands it a pipe when esphome itself is running under the dashboard. A
|
||||
pipe is block buffered, so without a flush in the shim's ``write()`` the
|
||||
output sits in the child until 8 KiB piles up or the build ends.
|
||||
"""
|
||||
runner_py = Path(runner.__file__)
|
||||
probe = fixture_path / "espidf" / "streaming_probe.py"
|
||||
|
||||
with subprocess.Popen(
|
||||
[sys.executable, str(runner_py), str(probe)],
|
||||
stdout=subprocess.PIPE,
|
||||
# Keep stderr: if the runner dies on startup, its traceback is the
|
||||
# only clue about why no line showed up.
|
||||
stderr=subprocess.PIPE,
|
||||
env=probe_env,
|
||||
text=True,
|
||||
) as proc:
|
||||
assert proc.stdout is not None
|
||||
assert proc.stderr is not None
|
||||
first_line: list[str] = []
|
||||
reader = threading.Thread(
|
||||
target=lambda: first_line.append(proc.stdout.readline()), daemon=True
|
||||
)
|
||||
try:
|
||||
reader.start()
|
||||
reader.join(FIRST_LINE_TIMEOUT)
|
||||
still_running = proc.poll() is None
|
||||
|
||||
# The probe sleeps for a minute after writing, so reaching us at
|
||||
# all means the line was flushed rather than released at exit.
|
||||
assert first_line == ["Compiling main.cpp\n"], (
|
||||
f"runner stderr: {'' if still_running else proc.stderr.read()}"
|
||||
)
|
||||
assert still_running
|
||||
finally:
|
||||
proc.kill()
|
||||
proc.wait()
|
||||
# Join before leaving the block, so the reader is done rather than
|
||||
# racing ``Popen`` closing the pipe under it.
|
||||
reader.join(1.0)
|
||||
|
||||
|
||||
class _FakeKernel32:
|
||||
"""Stand-in for the Windows kernel32 console code page calls."""
|
||||
|
||||
def __init__(self, input_cp: int, output_cp: int) -> None:
|
||||
self.input_cp = input_cp
|
||||
self.output_cp = output_cp
|
||||
self.calls: list[tuple[str, int]] = []
|
||||
|
||||
def GetConsoleCP(self) -> int: # noqa: N802
|
||||
return self.input_cp
|
||||
|
||||
def GetConsoleOutputCP(self) -> int: # noqa: N802
|
||||
return self.output_cp
|
||||
|
||||
def SetConsoleCP(self, codepage: int) -> int: # noqa: N802
|
||||
self.calls.append(("SetConsoleCP", codepage))
|
||||
self.input_cp = codepage
|
||||
return 1
|
||||
|
||||
def SetConsoleOutputCP(self, codepage: int) -> int: # noqa: N802
|
||||
self.calls.append(("SetConsoleOutputCP", codepage))
|
||||
self.output_cp = codepage
|
||||
return 1
|
||||
|
||||
|
||||
def test_main_runs_the_build_with_a_utf8_console(
|
||||
monkeypatch: pytest.MonkeyPatch, fixture_path: Path
|
||||
) -> None:
|
||||
"""An attached console is switched to UTF-8 and then put back."""
|
||||
kernel32 = _FakeKernel32(850, 850)
|
||||
monkeypatch.setattr(runner, "_get_kernel32", lambda: kernel32)
|
||||
|
||||
_run_main(monkeypatch, fixture_path / "espidf" / "filtering_probe.py")
|
||||
|
||||
assert kernel32.calls == [
|
||||
("SetConsoleCP", runner.UTF8_CODEPAGE),
|
||||
("SetConsoleOutputCP", runner.UTF8_CODEPAGE),
|
||||
("SetConsoleCP", 850),
|
||||
("SetConsoleOutputCP", 850),
|
||||
]
|
||||
|
||||
|
||||
def test_main_restores_the_console_when_the_build_dies(
|
||||
monkeypatch: pytest.MonkeyPatch, fixture_path: Path
|
||||
) -> None:
|
||||
"""A failing build must not leave the user's console on UTF-8."""
|
||||
kernel32 = _FakeKernel32(437, 437)
|
||||
monkeypatch.setattr(runner, "_get_kernel32", lambda: kernel32)
|
||||
_prepare_main(monkeypatch, fixture_path / "espidf" / "crashing_probe.py")
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
runner.main()
|
||||
|
||||
assert (kernel32.input_cp, kernel32.output_cp) == (437, 437)
|
||||
|
||||
|
||||
def test_main_restores_the_console_when_the_switch_fails_part_way(
|
||||
monkeypatch: pytest.MonkeyPatch, fixture_path: Path
|
||||
) -> None:
|
||||
"""A failed output page switch must not strand the changed input page."""
|
||||
kernel32 = _FakeKernel32(850, 850)
|
||||
|
||||
def _refuse(codepage: int) -> int:
|
||||
kernel32.calls.append(("SetConsoleOutputCP", codepage))
|
||||
return 0
|
||||
|
||||
kernel32.SetConsoleOutputCP = _refuse # type: ignore[method-assign]
|
||||
monkeypatch.setattr(runner, "_get_kernel32", lambda: kernel32)
|
||||
|
||||
_run_main(monkeypatch, fixture_path / "espidf" / "filtering_probe.py")
|
||||
|
||||
assert kernel32.input_cp == 850
|
||||
assert kernel32.calls[-2:] == [("SetConsoleCP", 850), ("SetConsoleOutputCP", 850)]
|
||||
|
||||
|
||||
def test_main_leaves_the_console_alone_when_there_is_none(
|
||||
monkeypatch: pytest.MonkeyPatch, fixture_path: Path
|
||||
) -> None:
|
||||
"""Without a console the code page calls return 0 and nothing is set."""
|
||||
kernel32 = _FakeKernel32(0, 0)
|
||||
monkeypatch.setattr(runner, "_get_kernel32", lambda: kernel32)
|
||||
|
||||
_run_main(monkeypatch, fixture_path / "espidf" / "filtering_probe.py")
|
||||
|
||||
assert kernel32.calls == []
|
||||
|
||||
|
||||
def test_main_leaves_a_console_already_on_utf8_alone(
|
||||
monkeypatch: pytest.MonkeyPatch, fixture_path: Path
|
||||
) -> None:
|
||||
"""An overlapping build must not save UTF-8 as the page to restore."""
|
||||
kernel32 = _FakeKernel32(runner.UTF8_CODEPAGE, runner.UTF8_CODEPAGE)
|
||||
monkeypatch.setattr(runner, "_get_kernel32", lambda: kernel32)
|
||||
|
||||
_run_main(monkeypatch, fixture_path / "espidf" / "filtering_probe.py")
|
||||
|
||||
assert kernel32.calls == []
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.platform == "win32", reason="kernel32 exists on Windows")
|
||||
def test_get_kernel32_is_none_off_windows() -> None:
|
||||
assert runner._get_kernel32() is None
|
||||
@@ -8,11 +8,12 @@ import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import call, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from esphome.components.esp32.const import KEY_ESP32, KEY_VARIANT
|
||||
from esphome.components.esp32.const import KEY_ESP32, KEY_IDF_VERSION, KEY_VARIANT
|
||||
import esphome.config_validation as cv
|
||||
from esphome.const import (
|
||||
CONF_COMPILE_PROCESS_LIMIT,
|
||||
CONF_ESPHOME,
|
||||
@@ -104,11 +105,27 @@ def _setup_build(setup_core: Path) -> tuple[Path, Path]:
|
||||
"""Point CORE at a build dir; return (compile_commands, idedata cache) paths."""
|
||||
CORE.name = "test"
|
||||
CORE.build_path = setup_core / "build" / "test"
|
||||
CORE.data.setdefault(KEY_ESP32, {})[KEY_IDF_VERSION] = cv.Version(5, 5, 5)
|
||||
compile_commands = CORE.relative_build_path("build", "compile_commands.json")
|
||||
cache = CORE.relative_internal_path("idedata", "test.json")
|
||||
return compile_commands, cache
|
||||
|
||||
|
||||
@contextmanager
|
||||
def _up_to_date_compile(ninja_side_effect=None) -> Iterator[tuple]:
|
||||
"""Patch run_compile's staleness checks to "up to date"; yield the ninja
|
||||
and print_summary mocks."""
|
||||
with (
|
||||
patch.object(toolchain, "need_reconfigure", return_value=False),
|
||||
patch.object(toolchain, "_cache_entries_changed", return_value=False),
|
||||
patch.object(
|
||||
toolchain, "_run_ninja", return_value=0, side_effect=ninja_side_effect
|
||||
) as mock_ninja,
|
||||
patch.object(toolchain, "print_summary") as mock_summary,
|
||||
):
|
||||
yield mock_ninja, mock_summary
|
||||
|
||||
|
||||
def test_has_outdated_files_detects_exclusion_change(setup_core: Path) -> None:
|
||||
"""A newer exclude_components.esphomeinternal stamp forces a reconfigure
|
||||
so components that leave the exclusion set get rediscovered."""
|
||||
@@ -279,28 +296,6 @@ def test_get_cmake_output_missing_build_does_not_resolve_idf_env(
|
||||
mock_run.assert_not_called()
|
||||
|
||||
|
||||
def test_run_idf_py_jobs_sets_build_jobs_env(setup_core: Path) -> None:
|
||||
"""The jobs argument is exported to idf.py as IDF_PY_BUILD_JOBS."""
|
||||
_setup_build(setup_core)
|
||||
|
||||
with (
|
||||
patch.object(toolchain, "_get_idf_path", return_value=Path("/idf")),
|
||||
patch.object(toolchain, "_get_idf_env", return_value={"PATH": "/bin"}),
|
||||
patch.object(toolchain, "_get_idf_tool", return_value="python"),
|
||||
patch.object(toolchain.subprocess, "run") as mock_run,
|
||||
):
|
||||
mock_run.return_value.returncode = 0
|
||||
|
||||
toolchain.run_idf_py("build", jobs=2)
|
||||
env = mock_run.call_args.kwargs["env"]
|
||||
assert env["IDF_PY_BUILD_JOBS"] == "2"
|
||||
assert env["PATH"] == "/bin"
|
||||
|
||||
toolchain.run_idf_py("build")
|
||||
env = mock_run.call_args.kwargs["env"]
|
||||
assert "IDF_PY_BUILD_JOBS" not in env
|
||||
|
||||
|
||||
def test_run_compile_restamps_cmakecache_after_discovery(setup_core: Path) -> None:
|
||||
"""After a successful discovery reconfigure the reference CMakeCache.txt
|
||||
is restamped; cmake does not rewrite it when only properties or plain
|
||||
@@ -325,7 +320,7 @@ def test_run_compile_restamps_cmakecache_after_discovery(setup_core: Path) -> No
|
||||
),
|
||||
patch("esphome.build_gen.espidf.write_project"),
|
||||
patch.object(toolchain, "run_reconfigure", return_value=0),
|
||||
patch.object(toolchain, "run_idf_py", return_value=0),
|
||||
patch.object(toolchain, "_run_ninja", return_value=0),
|
||||
patch.object(toolchain, "print_summary"),
|
||||
):
|
||||
assert toolchain.run_compile(config, verbose=False) == 0
|
||||
@@ -350,7 +345,7 @@ def test_run_compile_discovery_without_cmakecache(setup_core: Path) -> None:
|
||||
),
|
||||
patch("esphome.build_gen.espidf.write_project"),
|
||||
patch.object(toolchain, "run_reconfigure", return_value=0),
|
||||
patch.object(toolchain, "run_idf_py", return_value=0),
|
||||
patch.object(toolchain, "_run_ninja", return_value=0),
|
||||
patch.object(toolchain, "print_summary"),
|
||||
):
|
||||
assert toolchain.run_compile(config, verbose=False) == 0
|
||||
@@ -376,7 +371,7 @@ def test_run_compile_reconfigures_after_full_write_outside_testing_mode(
|
||||
def record_write(minimal: bool = False, builtin_components=None) -> None:
|
||||
calls.append(("write_project", minimal))
|
||||
|
||||
def record_reconfigure() -> int:
|
||||
def record_reconfigure(verbose: bool = False) -> int:
|
||||
nonlocal reconfigures
|
||||
reconfigures += 1
|
||||
calls.append(("run_reconfigure",))
|
||||
@@ -391,7 +386,7 @@ def test_run_compile_reconfigures_after_full_write_outside_testing_mode(
|
||||
),
|
||||
patch("esphome.build_gen.espidf.write_project", side_effect=record_write),
|
||||
patch.object(toolchain, "run_reconfigure", side_effect=record_reconfigure),
|
||||
patch.object(toolchain, "run_idf_py", return_value=0) as mock_build,
|
||||
patch.object(toolchain, "_run_ninja", return_value=0) as mock_build,
|
||||
patch.object(toolchain, "print_summary"),
|
||||
):
|
||||
assert not CORE.testing_mode
|
||||
@@ -421,7 +416,7 @@ def _record_compile_calls(
|
||||
calls: list[tuple] = []
|
||||
rcs = iter(reconfigure_rcs)
|
||||
|
||||
def record_reconfigure() -> int:
|
||||
def record_reconfigure(verbose: bool = False) -> int:
|
||||
calls.append(("run_reconfigure",))
|
||||
return next(rcs, 0)
|
||||
|
||||
@@ -431,6 +426,11 @@ def _record_compile_calls(
|
||||
def record_save(components: list[str]) -> None:
|
||||
calls.append(("save", components))
|
||||
|
||||
def record_ninja(target: str, **kwargs: object) -> int:
|
||||
if target == "all":
|
||||
calls.append(("build",))
|
||||
return 0
|
||||
|
||||
with (
|
||||
patch.object(toolchain, "need_reconfigure", return_value=True),
|
||||
patch.object(toolchain, "load_cached_builtin_components", return_value=cached),
|
||||
@@ -443,11 +443,7 @@ def _record_compile_calls(
|
||||
patch.object(
|
||||
toolchain, "_builtin_component_cache_path", return_value=cache_file
|
||||
),
|
||||
patch.object(
|
||||
toolchain,
|
||||
"run_idf_py",
|
||||
side_effect=lambda *a, **kw: calls.append(("build",)) or 0,
|
||||
),
|
||||
patch.object(toolchain, "_run_ninja", side_effect=record_ninja),
|
||||
patch.object(toolchain, "print_summary"),
|
||||
):
|
||||
rc = toolchain.run_compile({CONF_ESPHOME: {}}, verbose=False)
|
||||
@@ -630,19 +626,21 @@ def test_component_cache_ignores_corrupt_file(setup_core: Path, tmp_path: Path)
|
||||
assert toolchain.load_cached_builtin_components() is None
|
||||
|
||||
|
||||
def test_run_compile_passes_compile_process_limit(setup_core: Path) -> None:
|
||||
"""compile_process_limit is forwarded to run_idf_py as the job limit."""
|
||||
@pytest.mark.parametrize("limit", [1, None])
|
||||
def test_run_compile_passes_compile_process_limit(
|
||||
setup_core: Path, limit: int | None
|
||||
) -> None:
|
||||
"""compile_process_limit is the job limit for both ninja runs."""
|
||||
_setup_build(setup_core)
|
||||
config = {CONF_ESPHOME: {CONF_COMPILE_PROCESS_LIMIT: 1}}
|
||||
esphome = {} if limit is None else {CONF_COMPILE_PROCESS_LIMIT: limit}
|
||||
|
||||
with (
|
||||
patch.object(toolchain, "need_reconfigure", return_value=False),
|
||||
patch.object(toolchain, "run_idf_py", return_value=0) as mock_run,
|
||||
patch.object(toolchain, "print_summary"),
|
||||
):
|
||||
assert toolchain.run_compile(config, verbose=False) == 0
|
||||
with _up_to_date_compile() as (mock_run, _):
|
||||
assert toolchain.run_compile({CONF_ESPHOME: esphome}, verbose=False) == 0
|
||||
|
||||
mock_run.assert_called_once_with("build", "size", jobs=1)
|
||||
assert mock_run.call_args_list == [
|
||||
call("all", verbose=False, jobs=limit, progress=True),
|
||||
call("size", verbose=False, jobs=limit, extra_env=toolchain._size_env()),
|
||||
]
|
||||
|
||||
|
||||
def test_run_compile_passes_size_summary_paths(setup_core: Path) -> None:
|
||||
@@ -650,14 +648,8 @@ def test_run_compile_passes_size_summary_paths(setup_core: Path) -> None:
|
||||
ELF from get_built_elf_path, which must stay in lockstep with the
|
||||
project() name in the generated CMakeLists."""
|
||||
_setup_build(setup_core)
|
||||
config = {CONF_ESPHOME: {}}
|
||||
|
||||
with (
|
||||
patch.object(toolchain, "need_reconfigure", return_value=False),
|
||||
patch.object(toolchain, "run_idf_py", return_value=0),
|
||||
patch.object(toolchain, "print_summary") as mock_summary,
|
||||
):
|
||||
assert toolchain.run_compile(config, verbose=False) == 0
|
||||
with _up_to_date_compile() as (_, mock_summary):
|
||||
assert toolchain.run_compile({CONF_ESPHOME: {}}, verbose=False) == 0
|
||||
|
||||
mock_summary.assert_called_once_with(
|
||||
CORE.relative_build_path("build", "esp_idf_size.json"),
|
||||
@@ -682,21 +674,6 @@ def test_create_elf_copy_missing_source(setup_core: Path) -> None:
|
||||
assert toolchain.create_elf_copy() is False
|
||||
|
||||
|
||||
def test_run_compile_without_compile_process_limit(setup_core: Path) -> None:
|
||||
"""When no compile_process_limit is set, no job limit is passed to idf.py."""
|
||||
_setup_build(setup_core)
|
||||
config = {CONF_ESPHOME: {}}
|
||||
|
||||
with (
|
||||
patch.object(toolchain, "need_reconfigure", return_value=False),
|
||||
patch.object(toolchain, "run_idf_py", return_value=0) as mock_run,
|
||||
patch.object(toolchain, "print_summary"),
|
||||
):
|
||||
assert toolchain.run_compile(config, verbose=False) == 0
|
||||
|
||||
mock_run.assert_called_once_with("build", "size", jobs=None)
|
||||
|
||||
|
||||
def test_run_compile_writes_the_pch_checksum_before_the_build(
|
||||
setup_core: Path,
|
||||
) -> None:
|
||||
@@ -704,11 +681,7 @@ def test_run_compile_writes_the_pch_checksum_before_the_build(
|
||||
order: list[str] = []
|
||||
|
||||
with (
|
||||
patch.object(toolchain, "need_reconfigure", return_value=False),
|
||||
patch.object(
|
||||
toolchain, "run_idf_py", side_effect=lambda *a, **k: order.append("build")
|
||||
),
|
||||
patch.object(toolchain, "print_summary"),
|
||||
_up_to_date_compile(lambda *a, **k: order.append("build") or 0),
|
||||
patch(
|
||||
"esphome.build_gen.espidf.write_pch_checksum",
|
||||
side_effect=lambda: order.append("checksum"),
|
||||
@@ -727,6 +700,404 @@ def test_get_core_framework_version_from_core_data():
|
||||
assert toolchain._get_core_framework_version() == "5.5.4"
|
||||
|
||||
|
||||
@contextmanager
|
||||
def _fake_tools(env: dict[str, str] | None = None) -> Iterator:
|
||||
"""Stub the IDF env and tool lookup; yield the run_build_tool mock."""
|
||||
with (
|
||||
patch.object(
|
||||
toolchain,
|
||||
"_get_idf_env",
|
||||
return_value={"PATH": "/bin", "IDF_CCACHE_ENABLE": "0", **(env or {})},
|
||||
),
|
||||
patch.object(toolchain, "_get_idf_tool", side_effect=lambda n: f"/tools/{n}"),
|
||||
patch.object(toolchain, "run_build_tool", return_value=0) as mock_run,
|
||||
):
|
||||
yield mock_run
|
||||
|
||||
|
||||
def test_run_reconfigure_cmake_argv_matches_idf_py(setup_core: Path) -> None:
|
||||
"""The configure is the command idf.py runs.
|
||||
|
||||
IDF 5.5.5 tools/idf_py_actions/tools.py ensure_build_directory: cmake
|
||||
-G Ninja -DPYTHON_DEPS_CHECKED=1 -DPYTHON=<python> -DESP_PLATFORM=1,
|
||||
then the -D entries (SDKCONFIG, CCACHE_ENABLE), then the real project
|
||||
path, in the build dir. IDF 6.1 adds -B <build dir>, the same as the cwd.
|
||||
"""
|
||||
_setup_build(setup_core)
|
||||
sdkconfig = CORE.relative_build_path("sdkconfig.test")
|
||||
sdkconfig.parent.mkdir(parents=True)
|
||||
sdkconfig.write_text("")
|
||||
project = os.path.realpath(CORE.build_path)
|
||||
|
||||
with _fake_tools() as mock_run:
|
||||
assert toolchain.run_reconfigure() == 0
|
||||
|
||||
assert mock_run.call_args.args[0] == [
|
||||
"/tools/cmake",
|
||||
"-G",
|
||||
"Ninja",
|
||||
"-DPYTHON_DEPS_CHECKED=1",
|
||||
"-DPYTHON=/tools/python",
|
||||
"-DESP_PLATFORM=1",
|
||||
f"-DSDKCONFIG={sdkconfig}",
|
||||
"-DCCACHE_ENABLE=0",
|
||||
project,
|
||||
]
|
||||
kwargs = mock_run.call_args.kwargs
|
||||
assert kwargs["cwd"] == Path(project) / "build"
|
||||
assert kwargs["cwd"].is_dir()
|
||||
assert kwargs["env"]["CLICOLOR_FORCE"] == "1"
|
||||
assert kwargs["filter_lines"] is toolchain.FILTER_IDF_LINES
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("ccache", "expected"), [("1", "True"), ("0", "False")])
|
||||
def test_run_reconfigure_cmake_argv_matches_idf6_py(
|
||||
setup_core: Path, ccache: str, expected: str
|
||||
) -> None:
|
||||
"""IDF 6 idf.py adds -B <build dir> and formats CCACHE_ENABLE as True/False.
|
||||
|
||||
IDF 6.1 tools/idf_py_actions/tools.py ensure_build_directory.
|
||||
"""
|
||||
_setup_build(setup_core)
|
||||
CORE.data[KEY_ESP32][KEY_IDF_VERSION] = cv.Version(6, 1, 0)
|
||||
build_dir = Path(os.path.realpath(CORE.build_path)) / "build"
|
||||
|
||||
with _fake_tools({"IDF_CCACHE_ENABLE": ccache}) as mock_run:
|
||||
assert toolchain.run_reconfigure() == 0
|
||||
|
||||
assert mock_run.call_args.args[0] == [
|
||||
"/tools/cmake",
|
||||
"-G",
|
||||
"Ninja",
|
||||
"-B",
|
||||
str(build_dir),
|
||||
"-DPYTHON_DEPS_CHECKED=1",
|
||||
"-DPYTHON=/tools/python",
|
||||
"-DESP_PLATFORM=1",
|
||||
f"-DCCACHE_ENABLE={expected}",
|
||||
str(build_dir.parent),
|
||||
]
|
||||
|
||||
|
||||
def test_run_reconfigure_without_sdkconfig_or_filter(setup_core: Path) -> None:
|
||||
"""No sdkconfig file means no SDKCONFIG entry; -v shows every line."""
|
||||
_setup_build(setup_core)
|
||||
with _fake_tools() as mock_run:
|
||||
assert toolchain.run_reconfigure(verbose=True) == 0
|
||||
cmd = mock_run.call_args.args[0]
|
||||
assert not any(arg.startswith("-DSDKCONFIG=") for arg in cmd)
|
||||
assert cmd[-2] == "-DCCACHE_ENABLE=0"
|
||||
assert mock_run.call_args.kwargs["filter_lines"] is None
|
||||
|
||||
|
||||
def test_run_reconfigure_failure_removes_cmakecache(
|
||||
setup_core: Path, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""Like idf.py, a failed configure must not leave a cache that looks valid."""
|
||||
_setup_build(setup_core)
|
||||
cache = Path(os.path.realpath(CORE.build_path)) / "build" / "CMakeCache.txt"
|
||||
cache.parent.mkdir(parents=True)
|
||||
cache.write_text("")
|
||||
with (
|
||||
_fake_tools() as mock_run,
|
||||
patch.object(toolchain, "_print_hints") as mock_hints,
|
||||
):
|
||||
mock_run.return_value = 4
|
||||
assert toolchain.run_reconfigure() == 4
|
||||
assert not cache.exists()
|
||||
assert "CMake configure failed with exit code 4" in caplog.text
|
||||
log_path = mock_run.call_args.kwargs["log_path"]
|
||||
assert log_path == cache.parent / "log" / "cmake_output.log"
|
||||
mock_hints.assert_called_once_with(log_path)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("value", "expected"),
|
||||
[
|
||||
("1", "1"),
|
||||
("on", "1"),
|
||||
(" Yes ", "1"),
|
||||
("0", "0"),
|
||||
("", "0"),
|
||||
("disable", "0"),
|
||||
],
|
||||
)
|
||||
def test_cache_entries_ccache_follows_click_booleans(
|
||||
setup_core: Path, value: str, expected: str
|
||||
) -> None:
|
||||
"""IDF_CCACHE_ENABLE maps to CCACHE_ENABLE the way idf.py's click flag does."""
|
||||
_setup_build(setup_core)
|
||||
with _fake_tools({"IDF_CCACHE_ENABLE": value}):
|
||||
assert toolchain._cache_entries() == {"CCACHE_ENABLE": expected}
|
||||
|
||||
|
||||
def test_parse_cmakecache(tmp_path: Path) -> None:
|
||||
cache = tmp_path / "CMakeCache.txt"
|
||||
cache.write_text(
|
||||
"# comment\n"
|
||||
"// help text\n"
|
||||
"SDKCONFIG:UNINITIALIZED=/a/sdkconfig.test\n"
|
||||
"CCACHE_ENABLE:UNINITIALIZED=1\n"
|
||||
"FLAGS:STRING=-DX=1\n"
|
||||
"\n"
|
||||
)
|
||||
assert toolchain._parse_cmakecache(cache) == {
|
||||
"SDKCONFIG": "/a/sdkconfig.test",
|
||||
"CCACHE_ENABLE": "1",
|
||||
"FLAGS": "-DX=1",
|
||||
}
|
||||
|
||||
|
||||
# The constant -D entries every configure writes (see _configure_defines).
|
||||
_CONFIGURED = (
|
||||
"PYTHON_DEPS_CHECKED:UNINITIALIZED=1\n"
|
||||
"PYTHON:UNINITIALIZED=/tools/python\n"
|
||||
"ESP_PLATFORM:UNINITIALIZED=1\n"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("cache_text", "expected"),
|
||||
[
|
||||
(None, True),
|
||||
(_CONFIGURED + "CCACHE_ENABLE:UNINITIALIZED=0\n", False),
|
||||
(_CONFIGURED + "CCACHE_ENABLE:UNINITIALIZED=1\n", True),
|
||||
(
|
||||
_CONFIGURED.replace("/tools/python", "/old/python")
|
||||
+ "CCACHE_ENABLE:UNINITIALIZED=0\n",
|
||||
True,
|
||||
),
|
||||
(_CONFIGURED, True),
|
||||
],
|
||||
ids=["no_cache", "same", "changed", "python_moved", "missing"],
|
||||
)
|
||||
def test_cache_entries_changed(
|
||||
setup_core: Path, cache_text: str | None, expected: bool
|
||||
) -> None:
|
||||
"""Mirrors idf.py's _new_cmakecache_entries, plus the cached PYTHON."""
|
||||
_setup_build(setup_core)
|
||||
if cache_text is not None:
|
||||
cache = Path(os.path.realpath(CORE.build_path)) / "build" / "CMakeCache.txt"
|
||||
cache.parent.mkdir(parents=True)
|
||||
cache.write_text(cache_text)
|
||||
with _fake_tools():
|
||||
assert toolchain._cache_entries_changed() is expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("version", "size_ng"), [(cv.Version(5, 5, 5), True), (cv.Version(6, 1, 0), False)]
|
||||
)
|
||||
def test_size_env(setup_core: Path, version: cv.Version, size_ng: bool) -> None:
|
||||
"""IDF 5.x idf.py sets ESP_IDF_SIZE_NG; 6.x dropped it (core_ext.py size_target)."""
|
||||
CORE.data[KEY_ESP32] = {KEY_IDF_VERSION: version}
|
||||
env = toolchain._size_env()
|
||||
assert env["ESP_IDF_SIZE_FORCE_TERMINAL"] == "1"
|
||||
assert env["SIZE_OUTPUT_FORMAT"] == "default"
|
||||
assert ("ESP_IDF_SIZE_NG" in env) is size_ng
|
||||
|
||||
|
||||
def test_run_ninja_matches_idf_py_run_target(setup_core: Path) -> None:
|
||||
"""The command is ninja [-j N] [-v] <target> in the build dir."""
|
||||
_setup_build(setup_core)
|
||||
with _fake_tools() as mock_run:
|
||||
assert (
|
||||
toolchain._run_ninja(
|
||||
"size", verbose=True, jobs=2, progress=True, extra_env={"A": "b"}
|
||||
)
|
||||
== 0
|
||||
)
|
||||
assert mock_run.call_args.args[0] == ["/tools/ninja", "-j", "2", "-v", "size"]
|
||||
kwargs = mock_run.call_args.kwargs
|
||||
assert kwargs["cwd"] == Path(os.path.realpath(CORE.build_path)) / "build"
|
||||
assert kwargs["env"]["CLICOLOR_FORCE"] == "1"
|
||||
assert kwargs["env"]["A"] == "b"
|
||||
# -v shows every line as it comes, as idf.py does
|
||||
assert kwargs["filter_lines"] is None
|
||||
assert kwargs["progress"] is False
|
||||
|
||||
|
||||
def test_run_ninja_filters_and_reports_failure(
|
||||
setup_core: Path, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
_setup_build(setup_core)
|
||||
with (
|
||||
_fake_tools() as mock_run,
|
||||
patch.object(toolchain, "_print_hints") as mock_hints,
|
||||
):
|
||||
mock_run.return_value = 1
|
||||
assert toolchain._run_ninja("all", verbose=False, jobs=None, progress=True) == 1
|
||||
log_path = mock_run.call_args.kwargs["log_path"]
|
||||
assert log_path.name == "ninja_all_output.log"
|
||||
mock_hints.assert_called_once_with(log_path)
|
||||
assert mock_run.call_args.args[0] == ["/tools/ninja", "all"]
|
||||
assert mock_run.call_args.kwargs["filter_lines"] is toolchain.FILTER_IDF_LINES
|
||||
assert mock_run.call_args.kwargs["progress"] is True
|
||||
assert "ninja all failed with exit code 1" in caplog.text
|
||||
|
||||
|
||||
@pytest.mark.parametrize("reconfigure_rc", [0, 5])
|
||||
def test_run_compile_reconfigures_when_cache_entries_change(
|
||||
setup_core: Path, reconfigure_rc: int
|
||||
) -> None:
|
||||
"""A changed -D entry (ccache toggled) reconfigures, as idf.py build did."""
|
||||
_setup_build(setup_core)
|
||||
with (
|
||||
patch.object(toolchain, "need_reconfigure", return_value=False),
|
||||
patch.object(toolchain, "_cache_entries_changed", return_value=True),
|
||||
patch.object(
|
||||
toolchain, "run_reconfigure", return_value=reconfigure_rc
|
||||
) as mock_reconfigure,
|
||||
patch.object(toolchain, "_run_ninja", return_value=0) as mock_ninja,
|
||||
patch.object(toolchain, "print_summary"),
|
||||
):
|
||||
assert toolchain.run_compile({CONF_ESPHOME: {}}, verbose=True) == reconfigure_rc
|
||||
mock_reconfigure.assert_called_once_with(True)
|
||||
assert mock_ninja.called is (reconfigure_rc == 0)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("failing", ["all", "size"])
|
||||
def test_run_compile_stops_on_ninja_failure(setup_core: Path, failing: str) -> None:
|
||||
"""A failed build skips size; either failure skips the summary."""
|
||||
_setup_build(setup_core)
|
||||
with _up_to_date_compile(lambda target, **kw: 7 if target == failing else 0) as (
|
||||
mock_ninja,
|
||||
mock_summary,
|
||||
):
|
||||
assert toolchain.run_compile({CONF_ESPHOME: {}}, verbose=False) == 7
|
||||
targets = [c.args[0] for c in mock_ninja.call_args_list]
|
||||
assert targets == (["all"] if failing == "all" else ["all", "size"])
|
||||
mock_summary.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("memory_ld_rc", [0, 3])
|
||||
def test_run_compile_testing_mode_builds_memory_ld_first(
|
||||
setup_core: Path, memory_ld_rc: int
|
||||
) -> None:
|
||||
"""Testing mode builds and patches memory.ld before the main build."""
|
||||
_setup_build(setup_core)
|
||||
CORE.testing_mode = True
|
||||
targets: list[str] = []
|
||||
|
||||
def record(target: str, **kwargs: object) -> int:
|
||||
targets.append(target)
|
||||
return memory_ld_rc if target.endswith("memory.ld") else 0
|
||||
|
||||
with (
|
||||
_up_to_date_compile(record),
|
||||
patch.object(toolchain, "_patch_memory_segments") as mock_patch,
|
||||
):
|
||||
assert toolchain.run_compile({CONF_ESPHOME: {}}, verbose=False) == memory_ld_rc
|
||||
memory_ld = str(Path("esp-idf", "esp_system", "ld", "memory.ld"))
|
||||
if memory_ld_rc:
|
||||
assert targets == [memory_ld]
|
||||
mock_patch.assert_not_called()
|
||||
else:
|
||||
assert targets == [memory_ld, "all", "size"]
|
||||
mock_patch.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("version", [cv.Version(5, 5, 5), cv.Version(6, 1, 0)])
|
||||
@pytest.mark.parametrize(
|
||||
("env", "expected"),
|
||||
[
|
||||
({}, {"CLICOLOR_FORCE": "1", "FORCE_COLOR": "1"}),
|
||||
({"CLICOLOR_FORCE": "0"}, {"CLICOLOR_FORCE": "0", "FORCE_COLOR": "1"}),
|
||||
({"NO_COLOR": "1"}, {}),
|
||||
],
|
||||
ids=["default", "user-value", "no-color"],
|
||||
)
|
||||
def test_tool_env_colors(
|
||||
setup_core: Path, version: cv.Version, env: dict[str, str], expected: dict[str, str]
|
||||
) -> None:
|
||||
"""IDF 6 RunTool.__call__ color handling, used for every IDF version."""
|
||||
_setup_build(setup_core)
|
||||
CORE.data[KEY_ESP32][KEY_IDF_VERSION] = version
|
||||
with _fake_tools(env):
|
||||
result = toolchain._tool_env()
|
||||
colors = {k: v for k, v in result.items() if k in ("CLICOLOR_FORCE", "FORCE_COLOR")}
|
||||
assert colors == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("limit", "env_value", "expected"),
|
||||
[
|
||||
(2, "8", 2),
|
||||
(None, "8", 8),
|
||||
(None, "", None),
|
||||
(None, None, None),
|
||||
],
|
||||
ids=["limit-wins", "env", "empty-env", "unset"],
|
||||
)
|
||||
def test_build_jobs_honors_idf_py_build_jobs(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
limit: int | None,
|
||||
env_value: str | None,
|
||||
expected: int | None,
|
||||
) -> None:
|
||||
"""compile_process_limit first, else IDF_PY_BUILD_JOBS as idf.py read it."""
|
||||
if env_value is None:
|
||||
monkeypatch.delenv("IDF_PY_BUILD_JOBS", raising=False)
|
||||
else:
|
||||
monkeypatch.setenv("IDF_PY_BUILD_JOBS", env_value)
|
||||
esphome = {} if limit is None else {CONF_COMPILE_PROCESS_LIMIT: limit}
|
||||
assert toolchain._build_jobs({CONF_ESPHOME: esphome}) == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize("value", ["0", "-1", "many"])
|
||||
def test_build_jobs_rejects_invalid_idf_py_build_jobs(
|
||||
monkeypatch: pytest.MonkeyPatch, value: str
|
||||
) -> None:
|
||||
"""Like idf.py, a value that is not a positive integer is an error."""
|
||||
monkeypatch.setenv("IDF_PY_BUILD_JOBS", value)
|
||||
with pytest.raises(EsphomeError, match="positive integer"):
|
||||
toolchain._build_jobs({CONF_ESPHOME: {}})
|
||||
|
||||
|
||||
@contextmanager
|
||||
def _hint_env(tmp_path: Path, **run_kwargs: object) -> Iterator:
|
||||
with (
|
||||
patch.object(toolchain, "_get_idf_tool", return_value="/py"),
|
||||
patch.object(toolchain, "_get_idf_path", return_value=tmp_path / "idf"),
|
||||
patch.object(toolchain, "_get_idf_env", return_value={}),
|
||||
patch.object(toolchain.subprocess, "run", **run_kwargs) as mock_run,
|
||||
):
|
||||
yield mock_run
|
||||
|
||||
|
||||
def test_print_hints_shows_idf_advice(
|
||||
tmp_path: Path, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""The hint step runs IDF's own generate_hints on the failed tool's log."""
|
||||
log = tmp_path / "ninja_all_output.log"
|
||||
done = subprocess.CompletedProcess([], 0, "HINT: the binary is too big\n", "")
|
||||
with _hint_env(tmp_path, return_value=done) as mock_run:
|
||||
toolchain._print_hints(log)
|
||||
cmd = mock_run.call_args.args[0]
|
||||
assert cmd[0] == "/py"
|
||||
assert "generate_hints" in cmd[2]
|
||||
assert cmd[3:] == [str(tmp_path / "idf" / "tools"), str(log)]
|
||||
assert "HINT: the binary is too big" in caplog.text
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"run_kwargs",
|
||||
[
|
||||
{"return_value": subprocess.CompletedProcess([], 1, "", "ImportError")},
|
||||
{"return_value": subprocess.CompletedProcess([], 0, " \n", "")},
|
||||
{"side_effect": subprocess.TimeoutExpired("py", 60)},
|
||||
{"side_effect": OSError("gone")},
|
||||
],
|
||||
ids=["script-fails", "no-hints", "timeout", "oserror"],
|
||||
)
|
||||
def test_print_hints_never_fails_the_build(
|
||||
tmp_path: Path, caplog: pytest.LogCaptureFixture, run_kwargs: dict
|
||||
) -> None:
|
||||
"""No hints is fine; the hint step must not raise or add warnings."""
|
||||
with caplog.at_level("WARNING"), _hint_env(tmp_path, **run_kwargs):
|
||||
toolchain._print_hints(tmp_path / "log")
|
||||
assert caplog.records == []
|
||||
|
||||
|
||||
def test_get_cmake_cache_value_reads_the_configured_cache(setup_core: Path) -> None:
|
||||
_setup_build(setup_core)
|
||||
output = "CMAKE_OBJDUMP:FILEPATH=/tools/objdump\nESPHOME_PCH:BOOL=OFF\n"
|
||||
|
||||
Reference in New Issue
Block a user