From bb1318dce1a2555ab3713313880f4954d998bd37 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Sun, 19 Jul 2026 19:32:38 -1000 Subject: [PATCH] [core] Auto-clean the PlatformIO build environment when the Python version changes (#17671) Co-authored-by: pre-commit-ci-lite[bot] <117423508+pre-commit-ci-lite[bot]@users.noreply.github.com> --- esphome/platformio/toolchain.py | 187 ++++++++- esphome/writer.py | 15 +- requirements.txt | 1 + tests/unit_tests/test_platformio_toolchain.py | 360 ++++++++++++++++++ 4 files changed, 550 insertions(+), 13 deletions(-) diff --git a/esphome/platformio/toolchain.py b/esphome/platformio/toolchain.py index c97df812e3..105d4a8283 100644 --- a/esphome/platformio/toolchain.py +++ b/esphome/platformio/toolchain.py @@ -1,17 +1,35 @@ +from collections.abc import Iterable import json import logging import os from pathlib import Path import re import sys +from typing import TYPE_CHECKING from esphome.const import CONF_COMPILE_PROCESS_LIMIT, CONF_ESPHOME, KEY_CORE from esphome.core import CORE, EsphomeError -from esphome.helpers import add_git_ceiling_directory +from esphome.helpers import add_git_ceiling_directory, rmtree, write_file from esphome.util import FlashImage, run_external_process +if TYPE_CHECKING: + from platformio.project.config import ProjectConfig + _LOGGER = logging.getLogger(__name__) +# PlatformIO cache subdirs resolved via ProjectConfig. A full ``clean-all`` wipes +# these plus the whole ``core_dir``; a Python-version heal wipes these plus the +# penv while keeping ``core_dir`` (so the sibling stamp/lock survive). +_PIO_CACHE_DIRS = ("cache_dir", "packages_dir", "platforms_dir") + +# Marker recording the Python major.minor the PlatformIO cache was provisioned +# under, plus the lock guarding the check/wipe. Both live in the dir resolved +# by ``_pio_stamp_dir`` (NOT wiped by the heal), so they survive the wipe and +# are rewritten after it. +_PIO_PYTHON_STAMP_FILE = ".esphome.pio.stamp.json" +_PIO_PYTHON_STAMP_LOCK = ".esphome.pio.stamp.lock" +_PIO_PYTHON_STAMP_SCHEMA = "0" + def _strip_win_long_path_prefix(path: str) -> str: r"""Strip the Windows extended-length path prefix from ``path``. @@ -44,7 +62,174 @@ def _strip_win_long_path_prefix(path: str) -> str: return path +def get_platformio_config() -> "ProjectConfig | None": + """Return PlatformIO's ``ProjectConfig``, or None when PlatformIO is absent.""" + try: + from platformio.project.config import ProjectConfig + except ImportError: + return None + return ProjectConfig.get_instance() + + +def _pio_stamp_dir(config: "ProjectConfig") -> Path: + """Return the persistent home for the python-version stamp and lock. + + The parent of ``platforms_dir``, not ``core_dir``: the container/add-on + images relocate the platform/package caches to a persistent volume while + ``core_dir`` stays at the ephemeral default (its ``appstate.json`` must not + move), so a stamp under ``core_dir`` would be wiped on every image update + while the stale cache it guards survives. Everywhere else ``platforms_dir`` + sits inside ``core_dir`` and this resolves to ``core_dir``. + """ + return Path(config.get("platformio", "platforms_dir")).parent + + +def _delete_platformio_dirs(config: "ProjectConfig", pio_dirs: Iterable[str]) -> None: + """Delete each named PlatformIO dir resolved from *config*.""" + for pio_dir in pio_dirs: + path = Path(config.get("platformio", pio_dir)) + if path.is_dir(): + _LOGGER.info("Deleting PlatformIO %s %s", pio_dir, path) + rmtree(path) + + +def clean_platformio_cache() -> None: + """Wipe the whole PlatformIO cache (cache/packages/platforms/core). + + The full set ``clean-all`` (Reset Build Environment) clears. No-op when + PlatformIO is unavailable. + """ + config = get_platformio_config() + if config is None: + return + _delete_platformio_dirs(config, [*_PIO_CACHE_DIRS, "core_dir"]) + + +def _clean_platformio_python_env(config: "ProjectConfig", core_dir: Path) -> None: + """Wipe the cache subdirs + penv for a Python-version change. + + Keeps ``core_dir`` itself (and the stamp/lock siblings under it); otherwise + the same cache set ``clean-all`` clears. + """ + _delete_platformio_dirs(config, _PIO_CACHE_DIRS) + penv = core_dir / "penv" + if penv.is_dir(): + _LOGGER.info("Deleting PlatformIO penv %s", penv) + rmtree(penv) + + +def _current_python_minor() -> str: + """Return the running interpreter's ``major.minor`` (e.g. ``3.13``).""" + return f"{sys.version_info.major}.{sys.version_info.minor}" + + +def _read_pio_stamp_python(stamp_file: Path) -> str | None: + """Return the ``python_version`` recorded in *stamp_file*, or None.""" + try: + with stamp_file.open(encoding="utf-8") as f: + data = json.load(f) + except FileNotFoundError: + return None + except (json.JSONDecodeError, OSError) as err: + # A present-but-unreadable stamp is a distinct signal from an absent + # one, and it drives a cache clean; surface why at normal verbosity. + _LOGGER.warning("Could not read %s: %s", stamp_file, err) + return None + if not isinstance(data, dict): + return None + version = data.get("python_version") + return version if isinstance(version, str) else None + + +def _write_pio_stamp_python(stamp_file: Path, python_version: str) -> None: + """Atomically write the PlatformIO python-version stamp.""" + write_file( + stamp_file, + json.dumps( + { + "schema_version": _PIO_PYTHON_STAMP_SCHEMA, + "python_version": python_version, + } + ), + ) + + +def heal_platformio_python_env() -> None: + """Wipe the PlatformIO cache unless it is stamped for the running Python. + + A PlatformIO platform/tool package pins the Python versions it accepts when + it is provisioned, and ESPHome pins platforms to exact, immutable versions, + so a later interpreter bump (a container upgrading its base Python) leaves + the cached platform rejecting the new interpreter ("Python version must be + between ...") until the cache is wiped. A stamp records the ``major.minor`` + the cache was provisioned for; when it doesn't match the running + interpreter (or has never been written for an existing cache), the same + PlatformIO dirs ``clean-all`` wipes are cleaned so PlatformIO + re-provisions, matching Reset Build Environment automatically. The native + ESP-IDF toolchain already self-heals through its own stamp; this covers the + PlatformIO path. No-op when PlatformIO is unavailable. + """ + config = get_platformio_config() + if config is None: + return + try: + _check_platformio_python_stamp(config) + except (EsphomeError, OSError) as err: + # The check is a best-effort repair; a full or read-only cache volume + # must not abort a build that might otherwise work. The stamp write + # surfaces as EsphomeError (write_file wraps OSError). + _LOGGER.warning("PlatformIO build environment check failed: %s", err) + + +def _check_platformio_python_stamp(config: "ProjectConfig") -> None: + """Compare the stamp to the running interpreter; wipe and restamp on mismatch.""" + current = _current_python_minor() + stamp_dir = _pio_stamp_dir(config) + # Host the stamp/lock even before PlatformIO's first run creates the dir. + stamp_dir.mkdir(parents=True, exist_ok=True) + stamp_file = stamp_dir / _PIO_PYTHON_STAMP_FILE + + from filelock import FileLock + + with FileLock(str(stamp_dir / _PIO_PYTHON_STAMP_LOCK)): + provisioned = _read_pio_stamp_python(stamp_file) + if provisioned == current: + return + core_dir = Path(config.get("platformio", "core_dir")) + has_cache = ( + any( + Path(config.get("platformio", pio_dir)).is_dir() + for pio_dir in _PIO_CACHE_DIRS + ) + or (core_dir / "penv").is_dir() + ) + if has_cache: + if provisioned is None: + # An existing cache with no stamp predates the stamp: its + # provisioning interpreter is unknown, so clean once rather + # than leave a possibly-stale cache failing every build. + _LOGGER.info( + "Cleaning the PlatformIO build environment once so it " + "re-provisions for Python %s", + current, + ) + else: + _LOGGER.info( + "Python version changed (%s -> %s); cleaning PlatformIO " + "build environment so it re-provisions for the new " + "interpreter", + provisioned, + current, + ) + _clean_platformio_python_env(config, core_dir) + _write_pio_stamp_python(stamp_file, current) + + def run_platformio_cli(*args, **kwargs) -> str | int: + # Re-provision the PlatformIO cache if the interpreter's major.minor changed + # since it was last built; a stale platform otherwise rejects the new Python + # with "Python version must be between ..." until Reset Build Environment. + heal_platformio_python_env() os.environ["PLATFORMIO_FORCE_COLOR"] = "true" os.environ["PLATFORMIO_BUILD_DIR"] = str(CORE.relative_pioenvs_path().absolute()) os.environ.setdefault( diff --git a/esphome/writer.py b/esphome/writer.py index b7eeec916d..866377d2f5 100644 --- a/esphome/writer.py +++ b/esphome/writer.py @@ -670,18 +670,9 @@ def clean_all(configuration: list[str]): rmtree(install_path) # Clean PlatformIO project files - try: - from platformio.project.config import ProjectConfig - except ImportError: - # PlatformIO is not available, skip cleaning - pass - else: - config = ProjectConfig.get_instance() - for pio_dir in ["cache_dir", "packages_dir", "platforms_dir", "core_dir"]: - path = Path(config.get("platformio", pio_dir)) - if path.is_dir(): - _LOGGER.info("Deleting PlatformIO %s %s", pio_dir, path) - rmtree(path) + from esphome.platformio.toolchain import clean_platformio_cache + + clean_platformio_cache() GITIGNORE_CONTENT = """# Gitignore settings for ESPHome diff --git a/requirements.txt b/requirements.txt index fbfc034268..9dfff1452c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -27,6 +27,7 @@ smpclient==7.2.0 requests==2.34.2 py7zr==1.1.3 platformdirs==4.10.0 # native esp-idf toolchain global cache dir +filelock==3.29.0 # lock guarding the PlatformIO python-version cache heal # esp-idf >= 5.0 requires this pyparsing >= 3.3.2 diff --git a/tests/unit_tests/test_platformio_toolchain.py b/tests/unit_tests/test_platformio_toolchain.py index 568b43a259..013030d38f 100644 --- a/tests/unit_tests/test_platformio_toolchain.py +++ b/tests/unit_tests/test_platformio_toolchain.py @@ -2,12 +2,14 @@ # pylint: disable=protected-access +from collections.abc import Generator from contextlib import contextmanager from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import json import os from pathlib import Path import shutil +import sys import threading from types import SimpleNamespace from unittest.mock import MagicMock, Mock, call, patch @@ -1093,3 +1095,361 @@ def test_filter_platformio_lines_blocks_noisy_messages(msg: str) -> None: def test_filter_platformio_lines_allows_other_messages(msg: str) -> None: """Test that non-noisy platformio output lines pass through RedirectText.""" assert _filter_through_redirect(msg) == msg + "\n" + + +# --------------------------------------------------------------------------- +# PlatformIO python-version cache heal +# --------------------------------------------------------------------------- + +_CURRENT_MINOR = f"{sys.version_info.major}.{sys.version_info.minor}" +# Captured before the autouse guard patches the name, so tests can exercise the +# real implementation. +_REAL_GET_PLATFORMIO_CONFIG = toolchain.get_platformio_config + + +@pytest.fixture(autouse=True) +def _guard_real_platformio() -> Generator[None, None, None]: + """Default the PlatformIO config lookup to None so no test in this module + touches a real ~/.platformio; the heal tests re-patch it at a temp dir.""" + with patch.object(toolchain, "get_platformio_config", return_value=None): + yield + + +def _pio_layout(core_dir: Path) -> dict[str, Path]: + """Return the PlatformIO dir layout with cache/packages/platforms under core.""" + return { + "core_dir": core_dir, + "packages_dir": core_dir / "packages", + "platforms_dir": core_dir / "platforms", + "cache_dir": core_dir / ".cache", + } + + +def _split_pio_layout(tmp_path: Path) -> dict[str, Path]: + """Container-shape layout: caches on a persistent root, core_dir ephemeral.""" + persistent = tmp_path / "data" / "platformio" + return { + "core_dir": tmp_path / "root" / ".platformio", + "platforms_dir": persistent / "platforms", + "packages_dir": persistent / "packages", + "cache_dir": persistent / "cache", + } + + +def _seed_layout(layout: dict[str, Path]) -> None: + """Populate each cache dir (and the core penv) with a marker file.""" + for key in ("platforms_dir", "packages_dir", "cache_dir"): + layout[key].mkdir(parents=True, exist_ok=True) + (layout[key] / "marker").write_text("x", encoding="utf-8") + penv = layout["core_dir"] / "penv" + penv.mkdir(parents=True, exist_ok=True) + (penv / "marker").write_text("x", encoding="utf-8") + + +def _make_pio_config(layout: dict[str, Path] | Path) -> MagicMock: + """A ProjectConfig stand-in resolving platformio dir options from *layout*.""" + resolved = _pio_layout(layout) if isinstance(layout, Path) else layout + config = MagicMock() + config.get.side_effect = lambda section, option: ( + str(resolved[option]) if section == "platformio" else "" + ) + return config + + +@contextmanager +def _use_pio_config(layout: dict[str, Path] | Path) -> Generator[MagicMock, None, None]: + """Point ``get_platformio_config`` at a temp layout for the block.""" + config = _make_pio_config(layout) + with patch.object(toolchain, "get_platformio_config", return_value=config): + yield config + + +def _stamp_version(core_dir: Path) -> str | None: + """Read the python version recorded in the heal stamp under *core_dir*.""" + return toolchain._read_pio_stamp_python(core_dir / toolchain._PIO_PYTHON_STAMP_FILE) + + +def _cache_wiped(core_dir: Path) -> bool: + """True when the seeded cache subdir markers are gone.""" + return not any( + (core_dir / sub / "marker").exists() + for sub in ("packages", "platforms", ".cache") + ) + + +@pytest.fixture +def pio_core_dir(tmp_path: Path) -> Path: + """A populated PlatformIO core dir (packages/platforms/.cache/penv seeded).""" + core = tmp_path / "dot-platformio" + for sub in ("packages", "platforms", ".cache", "penv"): + seeded = core / sub + seeded.mkdir(parents=True) + (seeded / "marker").write_text("x", encoding="utf-8") + return core + + +def test_current_python_minor_matches_running_interpreter() -> None: + """_current_python_minor returns major.minor of the running interpreter.""" + assert toolchain._current_python_minor() == _CURRENT_MINOR + + +def test_pio_stamp_round_trip(tmp_path: Path) -> None: + """The stamp writer/reader round-trips and records the schema version.""" + stamp = tmp_path / toolchain._PIO_PYTHON_STAMP_FILE + toolchain._write_pio_stamp_python(stamp, "3.13") + assert toolchain._read_pio_stamp_python(stamp) == "3.13" + assert json.loads(stamp.read_text()) == { + "schema_version": toolchain._PIO_PYTHON_STAMP_SCHEMA, + "python_version": "3.13", + } + + +def test_read_pio_stamp_missing(tmp_path: Path) -> None: + """A missing stamp file yields None.""" + assert toolchain._read_pio_stamp_python(tmp_path / "nope.json") is None + + +def test_read_pio_stamp_malformed(tmp_path: Path) -> None: + """A corrupt stamp file yields None instead of raising.""" + stamp = tmp_path / "bad.json" + stamp.write_text("{not json", encoding="utf-8") + assert toolchain._read_pio_stamp_python(stamp) is None + + +def test_read_pio_stamp_unreadable_logs_warning( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A present-but-unreadable stamp yields None and warns.""" + stamp = tmp_path / "stamp.json" + stamp.mkdir() + with caplog.at_level("WARNING"): + assert toolchain._read_pio_stamp_python(stamp) is None + assert "Could not read" in caplog.text + + +def test_read_pio_stamp_without_python_version(tmp_path: Path) -> None: + """A stamp missing python_version yields None.""" + stamp = tmp_path / "s.json" + stamp.write_text(json.dumps({"schema_version": "0"}), encoding="utf-8") + assert toolchain._read_pio_stamp_python(stamp) is None + + +@pytest.mark.parametrize("payload", ["42", '"x"', "[1, 2]", "null"]) +def test_read_pio_stamp_non_object_json(tmp_path: Path, payload: str) -> None: + """Valid-but-non-object JSON in the stamp yields None, not a crash.""" + stamp = tmp_path / "s.json" + stamp.write_text(payload, encoding="utf-8") + assert toolchain._read_pio_stamp_python(stamp) is None + + +def test_clean_platformio_cache_none_config_is_noop() -> None: + """clean_platformio_cache is a no-op when PlatformIO is unavailable.""" + with patch.object(toolchain, "get_platformio_config", return_value=None): + toolchain.clean_platformio_cache() + + +def test_clean_platformio_cache_wipes_everything(pio_core_dir: Path) -> None: + """clean_platformio_cache removes cache/packages/platforms and core_dir.""" + with _use_pio_config(pio_core_dir): + toolchain.clean_platformio_cache() + assert not pio_core_dir.exists() + + +def test_heal_none_config_is_noop() -> None: + """Heal is a no-op (no error) when PlatformIO is unavailable.""" + with patch.object(toolchain, "get_platformio_config", return_value=None): + toolchain.heal_platformio_python_env() + + +def test_heal_fresh_cache_stamps_without_wipe(tmp_path: Path) -> None: + """A fresh core dir (no stamp, no penv) is stamped, not wiped.""" + core = tmp_path / "pio" + with _use_pio_config(core): + toolchain.heal_platformio_python_env() + assert _stamp_version(core) == _CURRENT_MINOR + + +def test_heal_stamp_matches_current_no_wipe(pio_core_dir: Path) -> None: + """A stamp matching the running interpreter leaves the cache untouched.""" + toolchain._write_pio_stamp_python( + pio_core_dir / toolchain._PIO_PYTHON_STAMP_FILE, _CURRENT_MINOR + ) + with _use_pio_config(pio_core_dir): + toolchain.heal_platformio_python_env() + assert not _cache_wiped(pio_core_dir) + assert (pio_core_dir / "penv" / "marker").exists() + + +def test_heal_stale_stamp_wipes_and_restamps(pio_core_dir: Path) -> None: + """A stamp from an older interpreter triggers a wipe + restamp; core_dir stays.""" + toolchain._write_pio_stamp_python( + pio_core_dir / toolchain._PIO_PYTHON_STAMP_FILE, "2.7" + ) + with _use_pio_config(pio_core_dir): + toolchain.heal_platformio_python_env() + assert _cache_wiped(pio_core_dir) + assert not (pio_core_dir / "penv").exists() + assert pio_core_dir.is_dir() + assert _stamp_version(pio_core_dir) == _CURRENT_MINOR + + +def test_heal_no_stamp_existing_cache_wipes_once( + pio_core_dir: Path, caplog: pytest.LogCaptureFixture +) -> None: + """An existing cache with no stamp is cleaned once and stamped.""" + with _use_pio_config(pio_core_dir), caplog.at_level("INFO"): + toolchain.heal_platformio_python_env() + assert _cache_wiped(pio_core_dir) + assert not (pio_core_dir / "penv").exists() + assert _stamp_version(pio_core_dir) == _CURRENT_MINOR + assert "once" in caplog.text + + +def test_heal_no_stamp_penv_only_counts_as_cache(tmp_path: Path) -> None: + """A core dir holding only a penv still triggers the one-time clean.""" + core = tmp_path / "pio" + penv = core / "penv" + penv.mkdir(parents=True) + (penv / "marker").write_text("x", encoding="utf-8") + with _use_pio_config(core): + toolchain.heal_platformio_python_env() + assert not penv.exists() + assert _stamp_version(core) == _CURRENT_MINOR + + +def test_heal_oserror_is_nonfatal( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A filesystem failure during the check warns instead of aborting the build.""" + blocker = tmp_path / "pio" + blocker.write_text("not a directory", encoding="utf-8") + with _use_pio_config(blocker), caplog.at_level("WARNING"): + toolchain.heal_platformio_python_env() + assert "build environment check failed" in caplog.text + + +def test_heal_stamp_write_failure_is_nonfatal( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A failed stamp write (EsphomeError from write_file) warns, not aborts.""" + with ( + _use_pio_config(tmp_path / "pio"), + patch.object( + toolchain, + "_write_pio_stamp_python", + side_effect=EsphomeError("disk full"), + ), + caplog.at_level("WARNING"), + ): + toolchain.heal_platformio_python_env() + assert "build environment check failed" in caplog.text + + +def test_heal_is_idempotent_across_runs(pio_core_dir: Path) -> None: + """After a heal writes the stamp, a re-provisioned cache is not wiped again.""" + with _use_pio_config(pio_core_dir): + toolchain.heal_platformio_python_env() + repop = pio_core_dir / "packages" + repop.mkdir(exist_ok=True) + (repop / "marker").write_text("x", encoding="utf-8") + toolchain.heal_platformio_python_env() + assert (pio_core_dir / "packages" / "marker").exists() + + +def test_pio_stamp_dir_is_platforms_parent(tmp_path: Path) -> None: + """The stamp home is the parent of platforms_dir, not core_dir.""" + layout = _split_pio_layout(tmp_path) + config = _make_pio_config(layout) + assert toolchain._pio_stamp_dir(config) == layout["platforms_dir"].parent + nested = _make_pio_config(tmp_path / "pio") + assert toolchain._pio_stamp_dir(nested) == tmp_path / "pio" + + +def test_heal_container_layout_stamps_persistent_root(tmp_path: Path) -> None: + """Container shape: the stamp lands on the persistent cache root.""" + layout = _split_pio_layout(tmp_path) + with _use_pio_config(layout): + toolchain.heal_platformio_python_env() + persistent = layout["platforms_dir"].parent + assert _stamp_version(persistent) == _CURRENT_MINOR + assert not (layout["core_dir"] / toolchain._PIO_PYTHON_STAMP_FILE).exists() + + +def test_heal_container_layout_stale_stamp_wipes_persistent_cache( + tmp_path: Path, +) -> None: + """Container shape: a stale stamp wipes the relocated persistent caches.""" + layout = _split_pio_layout(tmp_path) + _seed_layout(layout) + persistent = layout["platforms_dir"].parent + toolchain._write_pio_stamp_python( + persistent / toolchain._PIO_PYTHON_STAMP_FILE, "2.7" + ) + with _use_pio_config(layout): + toolchain.heal_platformio_python_env() + for key in ("platforms_dir", "packages_dir", "cache_dir"): + assert not layout[key].exists() + assert not (layout["core_dir"] / "penv").exists() + assert _stamp_version(persistent) == _CURRENT_MINOR + + +def test_heal_container_layout_survives_core_dir_wipe(tmp_path: Path) -> None: + """A python change is still detected after an image update wiped core_dir.""" + layout = _split_pio_layout(tmp_path) + _seed_layout(layout) + shutil.rmtree(layout["core_dir"]) + persistent = layout["platforms_dir"].parent + toolchain._write_pio_stamp_python( + persistent / toolchain._PIO_PYTHON_STAMP_FILE, "2.7" + ) + with _use_pio_config(layout): + toolchain.heal_platformio_python_env() + for key in ("platforms_dir", "packages_dir", "cache_dir"): + assert not layout[key].exists() + assert _stamp_version(persistent) == _CURRENT_MINOR + + +def test_get_platformio_config_returns_project_config() -> None: + """The real lookup returns a usable ProjectConfig when PlatformIO is present.""" + config = _REAL_GET_PLATFORMIO_CONFIG() + assert config is not None + assert hasattr(config, "get") + + +def test_get_platformio_config_none_when_platformio_absent() -> None: + """The lookup returns None when PlatformIO cannot be imported.""" + with patch.dict(sys.modules, {"platformio.project.config": None}): + assert _REAL_GET_PLATFORMIO_CONFIG() is None + + +def test_delete_platformio_dirs_skips_missing(tmp_path: Path) -> None: + """A named dir that does not exist is skipped without error.""" + (tmp_path / "packages").mkdir() + (tmp_path / "packages" / "marker").write_text("x", encoding="utf-8") + config = _make_pio_config(tmp_path) + # platforms_dir does not exist; packages_dir does. + toolchain._delete_platformio_dirs(config, ["packages_dir", "platforms_dir"]) + assert not (tmp_path / "packages").exists() + + +def test_heal_stale_stamp_wipes_when_penv_absent(pio_core_dir: Path) -> None: + """The penv wipe is skipped cleanly when no penv exists.""" + shutil.rmtree(pio_core_dir / "penv") + toolchain._write_pio_stamp_python( + pio_core_dir / toolchain._PIO_PYTHON_STAMP_FILE, "2.7" + ) + with _use_pio_config(pio_core_dir): + toolchain.heal_platformio_python_env() + assert _cache_wiped(pio_core_dir) + assert _stamp_version(pio_core_dir) == _CURRENT_MINOR + + +def test_run_platformio_cli_invokes_heal( + setup_core: Path, mock_run_external_process: Mock +) -> None: + """run_platformio_cli runs the heal before spawning PlatformIO.""" + CORE.build_path = str(setup_core / "build" / "test") + mock_run_external_process.return_value = 0 + with patch.object(toolchain, "heal_platformio_python_env") as mock_heal: + toolchain.run_platformio_cli("test") + mock_heal.assert_called_once()