github/esphome
2
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-10-01 08:50:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'dev' into jesserockz-2026-446

Browse Source
This commit is contained in:
J. Nick Koston
2026-09-24 14:21:53 +01:00
committed by GitHub
parent 63e1adfd5f b0c9a7d5bd
commit 86073304dc
706 changed files with 14509 additions and 8831 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/script/test_ci_custom.py
+202
View File
@@ -4,12 +4,16 @@ The rule flags an if/else/for/while whose only body is an unbraced ESP_LOG*() ca
empty statement -- and a -Wempty-body warning -- once the log level compiles the macro out). These
tests pin the comment/string/raw-string masker, the accepted control-statement shapes, and the
NOLINT escape hatch at both placements a contributor would try.
Also covers the ESP_LOG call scanner (_iter_log_calls) and the bare-literal-ternary lint.
"""
import importlib.util
from pathlib import Path
import sys
import pytest
SCRIPT_DIR = (Path(__file__).parent / ".." / ".." / "script").resolve()
sys.path.insert(0, str(SCRIPT_DIR))
_spec = importlib.util.spec_from_file_location("ci_custom", SCRIPT_DIR / "ci-custom.py")
@@ -145,3 +149,201 @@ def test_nolint_at_end_of_log_line_suppresses() -> None:
def test_nolint_on_control_line_suppresses() -> None:
assert not _lint("if (x) // NOLINT\n ESP_LOGD(t);\n")
# --- rule: UNIT_ constants must not be redefined (mirror of the CONF_ check) ---
# Real UNIT_ constants that live in each canonical home.
UNIT_IN_CONST_PY = ci_custom.UNIT_CONSTANTS[0]
UNIT_IN_COMPONENT_CONST = ci_custom.COMPONENT_UNIT_CONSTANTS[0]
def _unit_def(fname: str, content: str) -> list:
return ci_custom.lint_unit_from_const_py(fname, content)
def test_unit_already_in_const_py_is_flagged() -> None:
errs = _unit_def("esphome/components/x/sensor.py", f'{UNIT_IN_CONST_PY} = "x"\n')
assert errs
assert "const.py" in errs[0][2]
def test_unit_already_in_component_const_is_flagged() -> None:
errs = _unit_def(
"esphome/components/x/sensor.py", f'{UNIT_IN_COMPONENT_CONST} = "x"\n'
)
assert errs
assert "esphome.components.const" in errs[0][2]
def test_unit_not_in_const_py_is_tracked_not_flagged() -> None:
ci_custom.UNIT_CONSTANTS_USES.clear()
assert _unit_def("a.py", 'UNIT_FOO_BAR = "fb"\n') == []
assert ci_custom.UNIT_CONSTANTS_USES["UNIT_FOO_BAR"] == ["a.py"]
def test_unit_defined_in_three_files_is_flagged() -> None:
ci_custom.UNIT_CONSTANTS_USES.clear()
for fname in ("a.py", "b.py", "c.py"):
_unit_def(fname, 'UNIT_FOO_BAR = "fb"\n')
errs = ci_custom.lint_unit_constants_usage()
assert any("UNIT_FOO_BAR" in e and "3 files" in e for e in errs)
def test_unit_defined_in_two_files_is_not_flagged() -> None:
ci_custom.UNIT_CONSTANTS_USES.clear()
for fname in ("a.py", "b.py"):
_unit_def(fname, 'UNIT_FOO_BAR = "fb"\n')
assert ci_custom.lint_unit_constants_usage() == []
# --- same rule for CONF_, now also recognising the components/const home ---
CONF_IN_CONST_PY = ci_custom.CONSTANTS[0]
CONF_IN_COMPONENT_CONST = ci_custom.COMPONENT_CONSTANTS[0]
def _conf_def(fname: str, content: str) -> list:
return ci_custom.lint_conf_from_const_py(fname, content)
def test_conf_already_in_const_py_is_flagged() -> None:
errs = _conf_def("esphome/components/x/sensor.py", f'{CONF_IN_CONST_PY} = "x"\n')
assert errs
assert "const.py" in errs[0][2]
def test_conf_already_in_component_const_is_flagged() -> None:
errs = _conf_def(
"esphome/components/x/sensor.py", f'{CONF_IN_COMPONENT_CONST} = "x"\n'
)
assert errs
assert "esphome.components.const" in errs[0][2]
def test_conf_not_in_a_const_home_is_tracked_not_flagged() -> None:
ci_custom.CONSTANTS_USES.pop("CONF_FOO_BAR", None)
assert _conf_def("a.py", 'CONF_FOO_BAR = "foo_bar"\n') == []
assert ci_custom.CONSTANTS_USES["CONF_FOO_BAR"] == ["a.py"]
# --- ESP_LOG call scanner and bare-literal-ternary lint ---
def _calls(content: str) -> list[str | None]:
return [text for _, text in ci_custom._iter_log_calls(content)]
def _ternary_errors(content: str) -> list[tuple[int, int]]:
errs = ci_custom.lint_log_no_bare_literal_ternary(Path("x.cpp"), content)
return [(line, col) for line, col, _ in errs]
@pytest.mark.parametrize(
"content",
[
'ESP_LOGD(TAG, "a ) b ( c; d")',
'ESP_LOGD(TAG, "quote \\" inside")',
"ESP_LOGD(TAG, \"%s\", format_hex_pretty(x, '-', false).c_str())",
"ESP_LOGD(TAG, \"%c%c\", '(', ')')",
"ESP_LOGD(TAG, \"%d\", 1'000'000)",
'ESP_LOGD(TAG, // it\'s a comment with ) and (\n "x")',
'ESP_LOGD(TAG, /* :) */ "x")',
'ESP_LOGD(TAG, "%s", R"(say "hi" :) )")',
'ESP_LOGD(TAG, "%s", R"x(a)"b)x")',
],
)
def test_iter_log_calls_spans_whole_call(content: str) -> None:
calls = _calls(content + ";\nint other = (1);")
assert calls == [content]
def test_iter_log_calls_reports_unbalanced_call_once() -> None:
content = 'ESP_LOGD(TAG, "x";\nvoid f();'
assert _calls(content) == [None]
errs = ci_custom.lint_log_multiline_continuation(Path("x.cpp"), content)
assert len(errs) == 1
assert errs[0][:2] == (1, 1)
assert "no matching closing parenthesis" in errs[0][2]
assert _ternary_errors(content) == []
@pytest.mark.parametrize(
("content", "expected"),
[
# A ; inside the format string no longer cuts the call short
('ESP_LOGD(TAG, "a; b\\nc %s", x);', [(1, 20)]),
# A \n%s continuation is exempt since %s may expand to leading whitespace
('ESP_LOGD(TAG, "a\\n%s", x);', []),
('ESP_LOGD(TAG, "a\\n b");', []),
],
)
def test_multiline_continuation_detection(
content: str, expected: list[tuple[int, int]]
) -> None:
errs = ci_custom.lint_log_multiline_continuation(Path("x.cpp"), content)
assert [(line, col) for line, col, _ in errs] == expected
def test_exclusion_list_only_names_components_without_esp8266_tests() -> None:
root = Path(__file__).parent / ".." / ".."
for pattern in ci_custom.LOG_LITERAL_LINT_EXCLUDE:
if not pattern.startswith("esphome/components/"):
continue
prefix = pattern.removeprefix("esphome/components/").split("/")[0]
comps = list((root / "esphome" / "components").glob(prefix))
assert comps, f"{pattern!r} matches no component"
for comp in comps:
test = root / "tests" / "components" / comp.name / "test.esp8266-ard.yaml"
assert not test.exists(), (
f"{comp.name} builds for ESP8266, drop {pattern!r}"
)
def test_unbalanced_calls_are_reported_by_a_check_that_sees_every_file() -> None:
# lint_log_no_bare_literal_ternary skips unbalanced calls and relies on this
checks = {c["func"].__name__: c for c in ci_custom.LINT_CONTENT_CHECKS}
continuation = checks["lint_log_multiline_continuation"]
ternary = checks["lint_log_no_bare_literal_ternary"]
assert continuation["exclude"] == []
assert continuation["include"] == ternary["include"]
@pytest.mark.parametrize(
("content", "expected"),
[
('ESP_LOGD(TAG, "%s", x ? "on" : "off");', [(1, 25), (1, 32)]),
(
'ESP_LOGD(TAG, "%s", x ? LOG_STR_LITERAL("on") : LOG_STR_LITERAL("off"));',
[],
),
('ESP_LOGD(TAG, "%s", x ? LOG_STR_LITERAL("on") : "off");', [(1, 49)]),
('ESP_LOGD(TAG, "%s", x ? "on" : "");', [(1, 25)]),
(
'ESP_LOGD(TAG, "%s",\n x ? "yes"\n : "no");',
[(2, 14), (3, 14)],
),
("ESP_LOGD(TAG, \"%c\", x ? '1' : '0');", []),
('ESP_LOGD(TAG, "a ? b : c %s", x ? "on" : "off");', [(1, 35), (1, 42)]),
('ESP_LOGD(TAG, "x:" "y %s", p);', []),
('ESP_LOGD(TAG, "%s", x ? "on" : "off"); // NOLINT', []),
('ESP_LOGD(TAG, "%s",\n x ? "yes"\n : "no"); // NOLINT', []),
('ESP_LOGD(TAG, "%s", x ? /* c */ "on" : "off");', [(1, 33), (1, 40)]),
(
'ESP_LOGD(TAG, "%s",\n x ? "on" // NOLINT(some-clang-check)\n : "off");',
[(2, 14), (3, 14)],
),
],
)
def test_ternary_literal_detection(
content: str, expected: list[tuple[int, int]]
) -> None:
assert _ternary_errors(content) == expected
def test_ternary_error_message_names_the_literal() -> None:
errs = ci_custom.lint_log_no_bare_literal_ternary(
Path("x.cpp"), 'ESP_LOGD(TAG, "%s", x ? "enabled" : LOG_STR_LITERAL("off"));'
)
assert len(errs) == 1
assert 'LOG_STR_LITERAL("enabled")' in errs[0][2]
tests/script/test_platformio_install_deps.py
-649
View File
@@ -1,649 +0,0 @@
"""Tests for script/platformio_install_deps.py."""
from argparse import Namespace
import importlib.util
import inspect
from pathlib import Path
import shutil
import sys
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
from platformio import fs
from platformio.cache import ContentCache
from platformio.exception import InvalidJSONFile
from platformio.package.manager._install import PackageManagerInstallMixin
from platformio.package.manager.base import BasePackageManager
from platformio.package.manager.library import LibraryPackageManager
from platformio.package.manager.tool import ToolPackageManager
from platformio.package.meta import PackageCompatibility, PackageItem, PackageSpec
import pytest
from semantic_version import Version
_SCRIPT = Path(__file__).parents[2] / "script" / "platformio_install_deps.py"
def _load_script():
spec = importlib.util.spec_from_file_location("platformio_install_deps", _SCRIPT)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
# The real ContentCache would create dirs under the user's core dir
module.ContentCache = lambda *_: None
return module
def test_spec_key_collapses_destinations() -> None:
"""Two specs delivering one package share a directory and one key."""
mod = _load_script()
assert mod.spec_key("esphome/noise-c @ 1.0") == "noise-c"
assert mod.spec_key("esphome/noise-c@1.0") == "noise-c"
assert mod.spec_key("ESP32Async/AsyncTCP @ ^3.4.10") == mod.spec_key(
"esp32async/asynctcp @ 3.5.0"
)
url = "https://github.com/pioarduino/platform-espressif32/releases/download/{v}/platform-espressif32.zip"
assert mod.spec_key(url.format(v="55.03.311")) == mod.spec_key(
url.format(v="54.03.20")
)
def test_parse_specs_and_cli_args(tmp_path: Path) -> None:
"""Parsing skips unpinned and interpolated entries; the CLI rebuild
keeps the original flag pairing."""
ini = tmp_path / "platformio.ini"
ini.write_text(
"[env:a]\n"
"platform = fake/platform@1\n"
"lib_deps =\n"
" esphome/noise-c @ 1.0\n"
" ${common.lib_deps}\n"
" internal_lib\n"
"[env:b]\n"
"lib_deps =\n"
" esphome/noise-c @ 1.0\n"
)
mod = _load_script()
args = Namespace(libraries=True, platforms=True, tools=False)
libs, platforms, tools = mod.parse_specs(str(ini), args)
# exact-string duplicates collapse; distinct version pins survive
assert libs == ["esphome/noise-c @ 1.0"]
assert platforms == ["fake/platform@1"]
assert tools == []
assert mod.build_cli_args(libs, platforms, tools) == [
"-l",
"esphome/noise-c @ 1.0",
"-p",
"fake/platform@1",
]
class _FakeManager:
"""Scripted manager_cls: records installs, raises on demand."""
installed: set = set()
fail: set = set()
calls: list = []
lock_events: list = []
base_dir: str = "" # per-test tmp base; set by _reset_fake
def __init__(self, package_dir) -> None:
assert package_dir is None
@staticmethod
def _key(spec) -> str:
return spec if isinstance(spec, str) else str(spec)
def get_package(self, spec):
if self._key(spec) in self.installed:
return SimpleNamespace(path="/tmp/fake-pkg", spec=self._key(spec))
return None
def memcache_reset(self) -> None:
type(self).resets = getattr(type(self), "resets", 0) + 1
@property
def package_dir(self) -> str:
return str(Path(type(self).base_dir) / "packages")
def get_download_dir(self) -> str:
return str(Path(type(self).base_dir) / "downloads")
def get_tmp_dir(self) -> str:
return str(Path(type(self).base_dir) / "tmp")
def lock(self) -> None:
type(self).lock_events.append("lock")
def unlock(self) -> None:
type(self).lock_events.append("unlock")
def _install(self, spec, skip_dependencies, compatibility=None):
assert skip_dependencies is True
if self._key(spec) in self.fail:
raise RuntimeError("boom")
type(self).calls.append(spec)
type(self).compat_calls.append((self._key(spec), compatibility))
type(self).installed.add(self._key(spec)) # atomic under the GIL
def get_pkg_dependencies(self, pkg):
return getattr(type(self), "deps", {}).get(pkg.spec)
dependency_to_spec = staticmethod(BasePackageManager.dependency_to_spec)
def _reset_fake(base_dir: str = "", **kwargs) -> type:
# A fresh subclass per test: nothing leaks between tests through the
# class-level scripted state
return type(
"_ScriptedManager",
(_FakeManager,),
{
"base_dir": base_dir,
"installed": kwargs.get("installed", set()),
"fail": kwargs.get("fail", set()),
"calls": [],
"compat_calls": [],
"lock_events": [],
},
)
def test_parallel_install_empty_specs_is_a_no_op(tmp_path: Path) -> None:
mod = _load_script()
cls = _reset_fake(str(tmp_path))
mod.parallel_install(cls, [])
assert cls.calls == [] and cls.lock_events == []
def test_parallel_install_behavior(tmp_path: Path) -> None:
"""Duplicates collapse to one install, installed specs are filtered,
URL specs stay out of the wave, and the lock wraps the pool."""
mod = _load_script()
cls = _reset_fake(str(tmp_path), installed={"esphome/already @ 1.0"})
mod.parallel_install(
cls,
[
"esphome/noise-c @ 1.0",
"esphome/noise-c @ 1.0",
"esphome/already @ 1.0",
"https://x/framework.tar.xz",
],
)
assert cls.calls == ["esphome/noise-c @ 1.0"]
assert cls.lock_events == ["lock", "unlock"]
def test_parallel_install_failure_cleans_torn_destination(
tmp_path: Path, capsys
) -> None:
"""A failed install resets the memcache, removes what get_package can
see, and reports; the others still install."""
mod = _load_script()
cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
removed = []
torn = str(tmp_path / "packages" / "torn-pkg") # never created; only rmtree'd
def get_package(self, spec):
if spec == "esphome/bad @ 1.0" and getattr(cls, "resets", 0):
return SimpleNamespace(path=torn, spec=spec)
return _FakeManager.get_package(self, spec)
cls.get_package = get_package # throwaway subclass; nothing to restore
with patch.object(mod.fs, "rmtree", side_effect=removed.append):
mod.parallel_install(cls, ["esphome/bad @ 1.0", "esphome/good @ 1.0"])
assert "esphome/good @ 1.0" in cls.calls
assert removed == [torn]
out = capsys.readouterr().out
assert "Pre-install of esphome/bad @ 1.0 failed" in out
assert "Pre-install failed for 1 of 2 package(s)" in out
def test_parallel_install_runs_dependency_waves(tmp_path: Path) -> None:
"""Dependencies of wave-installed packages install in a second wave,
deduped by name; name-only platform libs stay with the serial pass."""
mod = _load_script()
cls = _reset_fake(str(tmp_path))
cls.deps = {
"esphome/noise-c @ 1.0": [
{"owner": "esphome", "name": "libsodium", "version": "^1.0"},
{"name": "SPI"},
],
"esphome/wg @ 1.0": [
{"owner": "esphome", "name": "libsodium", "version": "^1.0"},
],
}
mod.parallel_install(cls, ["esphome/noise-c @ 1.0", "esphome/wg @ 1.0"])
assert len(cls.calls) == 3 # the shared dep installs exactly once
assert {mod.spec_key(c) for c in cls.calls} == {"noise-c", "wg", "libsodium"}
# Wave-1 strings carry no compatibility; the dependency wave does
compats = dict(cls.compat_calls)
assert compats["esphome/noise-c @ 1.0"] is None
dep_compat = next(v for k, v in cls.compat_calls if "libsodium" in k)
assert dep_compat is not None # mirrors pio's install_dependency
def test_dependency_wave_excludes_url_specs(tmp_path: Path) -> None:
"""A dependency pinned to a URL surfaces as spec.uri; it must stay out
of the wave like string URL specs do."""
mod = _load_script()
cls = _reset_fake(str(tmp_path))
cls.deps = {
"esphome/noise-c @ 1.0": [
{"name": "vendored", "version": "https://github.com/x/y.git"},
],
}
mod.parallel_install(cls, ["esphome/noise-c @ 1.0"])
assert {mod.spec_key(c) for c in cls.calls} == {"noise-c"}
def test_failed_cleanup_fails_the_build(tmp_path: Path) -> None:
"""A torn destination still on disk after rmtree must fail the build:
fs.rmtree never raises (its onexc handler prints), so only the
destination's absence proves the cleanup worked."""
mod = _load_script()
cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
torn = tmp_path / "packages" / "torn-pkg"
torn.mkdir(parents=True)
def get_package(self, spec):
if getattr(cls, "resets", 0):
return SimpleNamespace(path=str(torn), spec=spec)
return None
cls.get_package = get_package # throwaway subclass; nothing to restore
with (
patch.object(mod.fs, "rmtree", lambda path: None), # onexc swallowed
pytest.raises(mod.CleanupError, match="could not remove"),
):
mod.parallel_install(cls, ["esphome/bad @ 1.0"])
assert cls.lock_events == ["lock", "unlock"] # still released
def test_unverifiable_torn_destination_fails_the_build(tmp_path: Path) -> None:
"""When the scan fails, the spec's own .piopm decides: an unremovable
leftover fails the build."""
mod = _load_script()
cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
dest = Path(cls.base_dir) / "packages" / "bad"
dest.mkdir(parents=True)
(dest / ".piopm").write_text('{"spec": {"owner": "esphome", "name": "bad"}}')
def bad_reset(self):
raise OSError("scan broken")
cls.memcache_reset = bad_reset
with (
patch.object(mod.fs, "rmtree", lambda path: None), # onexc swallowed
pytest.raises(mod.CleanupError, match="could not remove"),
):
mod.parallel_install(cls, ["esphome/bad @ 1.0"])
def test_unverifiable_scan_without_leftover_degrades(tmp_path: Path, capsys) -> None:
"""A failing scan with no destination on disk is never a build
failure blaming this spec."""
mod = _load_script()
cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
resets = {"n": 0}
def bad_reset(self):
# Fail clean_torn's reset; the coordinator's later reset works
resets["n"] += 1
if resets["n"] <= 1:
raise OSError("scan broken")
cls.memcache_reset = bad_reset
mod.parallel_install(cls, ["esphome/bad @ 1.0"])
assert "No resolvable destination to clean" in capsys.readouterr().out
def test_unresolvable_torn_destination_is_printed(tmp_path: Path, capsys) -> None:
"""A failed install with no resolvable package prints, so an invisible
torn directory is at least traceable."""
mod = _load_script()
cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
mod.parallel_install(cls, ["esphome/bad @ 1.0"])
assert "No resolvable destination to clean" in capsys.readouterr().out
def test_unparsable_torn_destination_is_removed(tmp_path: Path, capsys) -> None:
"""A torn dir get_package cannot resolve but whose .piopm names the
spec is removed instead of surviving into the serial pass."""
mod = _load_script()
cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
dest = Path(cls.base_dir) / "packages" / "bad"
dest.mkdir(parents=True)
(dest / ".piopm").write_text('{"spec": {"owner": "esphome", "name": "bad"}}')
with patch.object(mod.fs, "rmtree", shutil.rmtree):
mod.parallel_install(cls, ["esphome/bad @ 1.0"])
assert not dest.exists()
assert "Removed torn destination" in capsys.readouterr().out
def test_parse_specs_tools_branch(tmp_path: Path) -> None:
"""platform_packages parsing keeps owner'd tools and rewrites github
URL pins to bare URLs the wave then skips via parsed.uri."""
mod = _load_script()
ini = tmp_path / "platformio.ini"
ini.write_text(
"[env:t]\n"
"platform_packages =\n"
" ${common.platform_packages}\n"
" platformio/tool-scons@~4.40801.0\n"
" framework-arduinopico@https://github.com/earlephilhower/arduino-pico/releases/download/6.0.0/rp2040-6.0.0.zip\n"
)
args = Namespace(libraries=False, platforms=False, tools=True)
libs, platforms, tools = mod.parse_specs(str(ini), args)
assert libs == [] and platforms == []
assert tools == [
"platformio/tool-scons@~4.40801.0",
"https://github.com/earlephilhower/arduino-pico/releases/download/6.0.0/rp2040-6.0.0.zip",
]
assert mod.build_cli_args([], [], tools)[:2] == ["-t", tools[0]]
def test_warm_store_still_walks_dependencies(tmp_path: Path) -> None:
"""Already-installed top-level packages still feed the dependency
wave; a warm store can be missing a transitive dep."""
mod = _load_script()
cls = _reset_fake(str(tmp_path), installed={"esphome/noise-c @ 1.0"})
cls.deps = {
"esphome/noise-c @ 1.0": [
{"owner": "esphome", "name": "libsodium", "version": "^1.0"},
],
}
mod.parallel_install(cls, ["esphome/noise-c @ 1.0"])
assert [mod.spec_key(c) for c in cls.calls] == ["libsodium"]
def test_worker_system_exit_still_cleans(tmp_path: Path, capsys) -> None:
"""A worker SystemExit runs the torn cleanup before propagating; the
serial pass must never trust its leftovers."""
mod = _load_script()
cls = _reset_fake(str(tmp_path))
torn = tmp_path / "packages" / "torn-pkg"
torn.mkdir(parents=True)
def exiting_install(self, spec, skip_dependencies, compatibility=None):
raise SystemExit(0)
def get_package(self, spec):
if getattr(cls, "resets", 0):
return SimpleNamespace(path=str(torn), spec=spec)
return None
cls._install = exiting_install
cls.get_package = get_package
def real_rmtree(path):
Path(path).rmdir()
with (
patch.object(mod.fs, "rmtree", real_rmtree),
pytest.raises(SystemExit),
):
mod.parallel_install(cls, ["esphome/bad @ 1.0"])
assert not torn.exists()
def test_unlock_failure_is_fatal(tmp_path: Path) -> None:
"""A failed unlock must fail the build: the serial pass in another
process would block on the held flock."""
mod = _load_script()
cls = _reset_fake(str(tmp_path))
def bad_unlock(self):
raise OSError("flock broke")
cls.unlock = bad_unlock
with pytest.raises(mod.LockReleaseError, match="manager lock"):
mod.parallel_install(cls, ["esphome/good @ 1.0"])
def test_unlock_failure_keeps_inflight_error_as_context(tmp_path: Path) -> None:
"""An in-flight CleanupError stays attached when the unlock fault
takes over the raise."""
mod = _load_script()
cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
torn = tmp_path / "packages" / "bad"
torn.mkdir(parents=True)
def get_package(self, spec):
if getattr(cls, "resets", 0):
return SimpleNamespace(path=str(torn), spec=spec)
return None
def bad_unlock(self):
raise OSError("flock broke")
cls.get_package = get_package
cls.unlock = bad_unlock
with (
patch.object(mod.fs, "rmtree", lambda path: None), # leaves torn
pytest.raises(mod.LockReleaseError) as err,
):
mod.parallel_install(cls, ["esphome/bad @ 1.0"])
assert isinstance(err.value.__cause__.__context__, mod.CleanupError)
def test_chdir_failure_does_not_fail_the_wave(tmp_path: Path, monkeypatch) -> None:
"""A lost cwd is suppressed: further waves may misbehave and fall to
the serial pass, whose cwd is pinned."""
mod = _load_script()
cls = _reset_fake(str(tmp_path))
monkeypatch.setattr(mod.os, "chdir", MagicMock(side_effect=OSError("gone")))
mod.parallel_install(cls, ["esphome/good @ 1.0"])
assert cls.calls == ["esphome/good @ 1.0"]
def test_piopm_match_removes_manifest_named_torn_dir(tmp_path: Path, capsys) -> None:
"""A torn dir named by its manifest (not the registry spec) is found
through its .piopm and removed."""
mod = _load_script()
cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
torn = tmp_path / "packages" / "ManifestName"
torn.mkdir(parents=True)
(torn / ".piopm").write_text('{"spec": {"owner": "esphome", "name": "bad"}}')
innocent = tmp_path / "packages" / "innocent"
innocent.mkdir()
(innocent / ".piopm").write_text('{"spec": {"owner": "o", "name": "other"}}')
with patch.object(mod.fs, "rmtree", shutil.rmtree):
mod.parallel_install(cls, ["esphome/bad @ 1.0"])
assert not torn.exists()
assert innocent.exists() # another package's valid metadata survives
assert "Removed torn destination" in capsys.readouterr().out
def test_unscannable_package_dir_fails_the_build(tmp_path: Path) -> None:
"""A storage dir the cleanup cannot scan is not proof of cleanliness."""
mod = _load_script()
cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
real_iterdir = Path.iterdir
def broken_iterdir(self):
if self.name == "packages":
raise PermissionError("denied")
return real_iterdir(self)
with (
patch.object(Path, "iterdir", broken_iterdir),
pytest.raises(mod.CleanupError, match="cleanup failed"),
):
mod.parallel_install(cls, ["esphome/bad @ 1.0"])
def test_stray_file_in_package_dir_is_ignored(tmp_path: Path) -> None:
"""A plain file (or a pio-link) beside the packages is skipped by
pio's own scan and must never hard-fail the build."""
mod = _load_script()
cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
(tmp_path / "packages").mkdir(parents=True)
(tmp_path / "packages" / "stray.pio-link").write_text("x")
(tmp_path / "packages" / "no-metadata").mkdir() # pio overwrites these
mod.parallel_install(cls, ["esphome/bad @ 1.0"])
assert (tmp_path / "packages" / "stray.pio-link").exists()
assert (tmp_path / "packages" / "no-metadata").exists()
def test_unreadable_piopm_dir_is_removed(tmp_path: Path) -> None:
"""A persistently corrupt .piopm under this spec's own name would
crash pio's storage scan; the dir is removed rather than left to
break the serial pass."""
mod = _load_script()
cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
torn = tmp_path / "packages" / "bad"
torn.mkdir(parents=True)
(torn / ".piopm").write_text("{not json")
mod.parallel_install(cls, ["esphome/bad @ 1.0"])
assert not torn.exists()
def test_unreadable_piopm_under_other_name_survives(tmp_path: Path) -> None:
"""A corrupt .piopm in another package's dir may be a worker mid-copy;
a failing spec must not remove a directory it does not own."""
mod = _load_script()
cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
other = tmp_path / "packages" / "innocent"
other.mkdir(parents=True)
(other / ".piopm").write_text("{not json")
mod.parallel_install(cls, ["esphome/bad @ 1.0"])
assert other.exists()
def test_unexpected_cleanup_class_becomes_cleanup_error(tmp_path: Path) -> None:
"""Cleanup failures of any class fail the build; nothing may be
downgraded to the serial fallback over a torn directory."""
mod = _load_script()
cls = _reset_fake(str(tmp_path), fail={"esphome/bad @ 1.0"})
with (
patch.object(
mod, "piopm_matches", MagicMock(side_effect=ValueError("bad spec"))
),
pytest.raises(mod.CleanupError, match="cleanup failed"),
):
mod.parallel_install(cls, ["esphome/bad @ 1.0"])
def test_main_cleanup_error_fails_before_generic_fallback(tmp_path: Path) -> None:
"""A CleanupError must escape main's serial fallback: the clause order
decides whether a stuck torn package fails the image build."""
mod = _load_script()
ini = tmp_path / "platformio.ini"
ini.write_text("[env:t]\nlib_deps =\n esphome/x @ 1.0\n")
with (
patch.object(
mod, "parallel_install", side_effect=mod.CleanupError("stuck torn pkg")
),
patch.object(mod.subprocess, "check_call"),
patch.object(sys, "argv", ["platformio_install_deps.py", str(ini), "-l"]),
pytest.raises(mod.CleanupError),
):
mod.main()
def test_main_generic_failure_still_runs_serial_pass(tmp_path: Path) -> None:
"""A non-CleanupError wave failure prints, dumps the traceback, and
still reaches the authoritative serial pass with the pinned cwd."""
mod = _load_script()
ini = tmp_path / "platformio.ini"
ini.write_text("[env:t]\nlib_deps =\n esphome/x @ 1.0\n")
with (
patch.object(mod, "parallel_install", side_effect=RuntimeError("boom")),
patch.object(mod.subprocess, "check_call") as mock_call,
patch.object(sys, "argv", ["platformio_install_deps.py", str(ini), "-l"]),
):
mod.main()
mock_call.assert_called_once()
args, kwargs = mock_call.call_args
assert args[0][:4] == ["platformio", "pkg", "install", "-g"]
assert "esphome/x @ 1.0" in args[0]
assert kwargs["cwd"] == Path.cwd()
def test_content_cache_creates_its_dir(tmp_path: Path, monkeypatch) -> None:
"""The cold-cache hardening relies on ContentCache.__init__ creating
the namespace dir; pin the side effect, not mere callability."""
monkeypatch.setenv("PLATFORMIO_CACHE_DIR", str(tmp_path / "cache"))
ContentCache("http")
assert (tmp_path / "cache" / "http").is_dir()
def test_piopm_matches_without_name_matches_nothing(tmp_path: Path) -> None:
"""A spec with no derivable name can never match a directory."""
mod = _load_script()
assert mod.piopm_matches(str(tmp_path), "") == []
def test_unresolvable_spec_stays_out_of_the_wave(tmp_path: Path, capsys) -> None:
"""A spec with no derivable name is left to the serial pass; a raw
string key would break the one-per-destination dedupe."""
mod = _load_script()
cls = _reset_fake(str(tmp_path))
nameless = PackageSpec(requirements="^1.0")
mod.parallel_install(cls, [nameless])
assert cls.calls == []
assert "Skipping unresolvable spec" in capsys.readouterr().out
def test_parallel_install_unlocks_when_pool_fails(tmp_path: Path) -> None:
mod = _load_script()
cls = _reset_fake(str(tmp_path))
with (
patch.object(mod, "ThreadPoolExecutor", side_effect=RuntimeError("no")),
pytest.raises(RuntimeError),
):
mod.parallel_install(cls, ["esphome/a @ 1.0"])
assert cls.lock_events == ["lock", "unlock"]
def test_parse_specs_unreadable_ini_fails_loudly(tmp_path: Path) -> None:
"""A bad path must not silently build an image with no dependencies."""
mod = _load_script()
args = Namespace(libraries=True, platforms=False, tools=False)
with pytest.raises(SystemExit):
mod.parse_specs(str(tmp_path / "missing.ini"), args)
def test_platformio_surface_for_install_deps_script() -> None:
"""A PlatformIO bump that changes these members must fail here, not
silently turn the docker image's parallel preinstall into a no-op."""
# The script calls these positionally; pin the positions, not just
# membership, so a parameter reorder trips the wire too
params = inspect.signature(PackageManagerInstallMixin._install).parameters
assert list(params)[1] == "spec"
assert "skip_dependencies" in params
assert "compatibility" in params
for cls in (ToolPackageManager, LibraryPackageManager):
assert list(inspect.signature(cls.__init__).parameters)[1] == "package_dir"
for name in (
"lock",
"unlock",
"get_package",
"memcache_reset",
"get_pkg_dependencies",
"dependency_to_spec",
"get_download_dir",
"get_tmp_dir",
):
assert callable(getattr(BasePackageManager, name))
# Losing any of these turns the wave into main()'s silent serial
# fallback: ensure_spec runs in the coordinator, the spec attributes
# feed the dedupe, cleanup, and dependency filters
assert callable(BasePackageManager.ensure_spec)
spec = PackageSpec("owner/name @ ^1.0")
assert spec.name == "name"
assert spec.owner == "owner"
assert spec.uri is None
assert spec.external is False
assert Version("1.5.0") in spec.requirements
# The failure-cleanup path degrades to a single line if these vanish
assert callable(fs.rmtree)
assert callable(fs.load_json)
# piopm_matches only tolerates a corrupt .piopm through this base;
# losing it would flip a wave failure from degrade to build failure
assert issubclass(InvalidJSONFile, ValueError)
assert PackageItem("pkg-dir").path == "pkg-dir"
assert callable(PackageCompatibility.from_dependency)