esphome/tests/unit_tests/test_nrf52_framework.py

"""Tests for esphome.components.nrf52.framework helpers."""

import errno
import hashlib
import os
from pathlib import Path
import sys
from types import SimpleNamespace
from unittest.mock import ANY, call, patch

import platformdirs
import pytest

from esphome.components.nrf52 import _resolve_toolchain
from esphome.components.nrf52.framework import (
    _PLATFORMIO_PENV_REQUIREMENTS,
    _REQUIREMENTS,
    DEFAULT_WEST_PROJECTS,
    TOOLCHAIN_VERSION,
    _get_penv_site_packages,
    _get_platformio_penv_path,
    _get_toolchain_platform_info,
    _needs_venv_rebuild,
    _wanted_west_projects,
    check_and_install,
    get_build_env,
    get_sdk_nrf_tools_path,
    include_west_project,
    setup_platformio_python_env,
)
from esphome.components.zephyr.const import KEY_SYSBUILD, KEY_ZEPHYR
import esphome.config_validation as cv
from esphome.config_validation import Version
from esphome.const import KEY_CORE, KEY_FRAMEWORK_VERSION, Toolchain
from esphome.core import CORE, EsphomeError
from esphome.framework_helpers import get_python_env_executable_path


@pytest.fixture(autouse=True)
def _isolate_sdk_nrf_install_path(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Pin the sdk-nrf install root to a tmp dir for every test.

    The default location is the OS user cache dir, so without this any test
    that builds framework paths or pre-creates the install dir would touch
    the real ``~/.cache/esphome`` on the developer's machine. Tests that need
    to exercise the override or default-resolution logic clear/override the
    env themselves.
    """
    monkeypatch.setenv("ESPHOME_SDK_NRF_PREFIX", str(tmp_path / "sdk_nrf_install"))


@pytest.mark.parametrize(
    ("system", "machine", "expected"),
    [
        # default — no branch hit
        ("Linux", "x86_64", ("linux", "x86_64", "tar.xz")),
        # arm64 → aarch64 rename
        ("Linux", "arm64", ("linux", "aarch64", "tar.xz")),
        # darwin → macos rename only
        ("Darwin", "x86_64", ("macos", "x86_64", "tar.xz")),
        # both renames apply
        ("Darwin", "arm64", ("macos", "aarch64", "tar.xz")),
        # windows forces x86_64 + 7z; arm64 rename is overwritten
        ("Windows", "arm64", ("windows", "x86_64", "7z")),
    ],
)
def test_get_toolchain_platform_info(
    system: str, machine: str, expected: tuple[str, str, str]
) -> None:
    with (
        patch("platform.system", return_value=system),
        patch("platform.machine", return_value=machine),
    ):
        assert _get_toolchain_platform_info() == expected


# ---------------------------------------------------------------------------
# Helpers and fixtures for check_and_install tests
# ---------------------------------------------------------------------------

_TEST_SDK_VERSION = "2.9.0"
# The filter that keeps only DEFAULT_WEST_PROJECTS
_DEFAULTS_FILTER = "-.*,+cmsis,+hal_nordic,+nrfxlib,+zephyr"


@pytest.fixture
def nrf52_dirs(setup_core: Path) -> SimpleNamespace:
    """Populate CORE and pre-create SDK directories so sentinel.touch() succeeds."""
    CORE.data[KEY_CORE] = {KEY_FRAMEWORK_VERSION: Version.parse(_TEST_SDK_VERSION)}
    tools = get_sdk_nrf_tools_path()
    python_env = tools / "penvs" / f"v{_TEST_SDK_VERSION}"
    framework = tools / "frameworks" / f"v{_TEST_SDK_VERSION}"
    toolchain_dir = tools / "toolchains" / TOOLCHAIN_VERSION
    for d in (python_env, framework, toolchain_dir):
        d.mkdir(parents=True, exist_ok=True)
    zephyr_scripts = framework / "zephyr" / "scripts"
    zephyr_scripts.mkdir(parents=True, exist_ok=True)
    (zephyr_scripts / "requirements.txt").touch()
    return SimpleNamespace(
        python_env=python_env,
        framework=framework,
        toolchain=toolchain_dir,
    )


@pytest.fixture
def mock_nrf52_ops():
    """Patch all heavy I/O operations used by check_and_install."""
    with (
        patch("esphome.components.nrf52.framework.rmdir") as mock_rmdir,
        patch("esphome.components.nrf52.framework.create_venv") as mock_create_venv,
        patch(
            "esphome.components.nrf52.framework.run_command_ok", return_value=True
        ) as mock_run_cmd,
        # download_and_extract resolves its internals in framework_helpers,
        # so the download/extract seams are patched there.
        patch(
            "esphome.framework_helpers.download_from_mirrors",
            return_value="https://example.com/tc.tar.xz",
        ) as mock_download,
        patch("esphome.framework_helpers.archive_extract_all") as mock_extract,
    ):
        yield SimpleNamespace(
            rmdir=mock_rmdir,
            create_venv=mock_create_venv,
            run_command_ok=mock_run_cmd,
            download_from_mirrors=mock_download,
            archive_extract_all=mock_extract,
        )


# ---------------------------------------------------------------------------
# check_and_install tests
# ---------------------------------------------------------------------------


def _mark_west_initialized(framework: Path) -> None:
    """What a finished ``west init`` leaves behind."""
    (framework / ".west").mkdir()
    (framework / ".west" / "config").touch()


def _touch_penv_python(penv: Path) -> None:
    """Create the interpreter file so the rebuild gate sees a live venv."""
    python = get_python_env_executable_path(penv, "python")
    python.parent.mkdir(parents=True, exist_ok=True)
    python.touch()


def _subcommand(cmd: list[str]) -> str:
    tool = "west" if "west" in cmd else "pip"
    return cmd[cmd.index(tool) + 1]


def _subcommands(run_command_ok) -> list[str]:
    """The west or pip subcommand of each command run, in order."""
    return [_subcommand(c.args[0]) for c in run_command_ok.call_args_list]


def _project_filter(run_command_ok) -> str:
    """The manifest.project-filter value of the last ``west config`` run."""
    for west_call in reversed(run_command_ok.call_args_list):
        cmd = west_call.args[0]
        if "manifest.project-filter" in cmd:
            return cmd[-1]
    raise AssertionError("no west config command ran")


def _mark_installed(dirs: SimpleNamespace) -> None:
    """Every install step finished: venv, zephyr requirements, SDK, toolchain."""
    _mark_venv_ready(dirs.python_env)
    (dirs.python_env / ".zephyr_reqs_ready").touch()
    (dirs.framework / ".ready").touch()
    (dirs.toolchain / ".ready").touch()


def _mark_venv_ready(python_env: Path) -> None:
    """Write the venv sentinel with the current requirements hash and a
    present interpreter so the rebuild gate passes."""
    requirements_hash = hashlib.sha256(_REQUIREMENTS.read_bytes()).hexdigest()
    (python_env / ".ready").write_text(requirements_hash, encoding="utf-8")
    _touch_penv_python(python_env)


class TestCheckAndInstall:
    def test_all_installed_skips_all_steps(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """All three sentinels present → nothing downloaded or compiled."""
        _mark_installed(nrf52_dirs)

        check_and_install()

        mock_nrf52_ops.create_venv.assert_not_called()
        mock_nrf52_ops.run_command_ok.assert_not_called()
        mock_nrf52_ops.download_from_mirrors.assert_not_called()
        mock_nrf52_ops.archive_extract_all.assert_not_called()

    def test_missing_interpreter_rebuilds_venv(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """A valid sentinel must not mask a missing interpreter (a cached venv
        restored after a host interpreter upgrade)."""
        requirements_hash = hashlib.sha256(_REQUIREMENTS.read_bytes()).hexdigest()
        (nrf52_dirs.python_env / ".ready").write_text(
            requirements_hash, encoding="utf-8"
        )
        # no interpreter on disk

        check_and_install()

        mock_nrf52_ops.create_venv.assert_called_once()

    def test_fresh_install_runs_all_steps(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """No sentinels → venv created, west installed, SDK init+update, toolchain downloaded."""
        check_and_install()

        mock_nrf52_ops.create_venv.assert_called_once()
        assert _subcommands(mock_nrf52_ops.run_command_ok) == [
            "install",  # requirements
            "init",
            "config",
            "update",
            "list",
            "install",  # zephyr requirements
        ]
        # minimal SDK + per-arch toolchain
        assert mock_nrf52_ops.download_from_mirrors.call_count == 2
        assert mock_nrf52_ops.archive_extract_all.call_count == 2
        assert (nrf52_dirs.python_env / ".ready").exists()
        assert (nrf52_dirs.python_env / ".zephyr_reqs_ready").exists()
        assert (nrf52_dirs.framework / ".ready").exists()
        assert (nrf52_dirs.toolchain / ".ready").exists()

    def test_venv_exists_installs_framework_and_toolchain(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """Venv ready but framework missing → skip venv creation, run SDK init+update."""
        _mark_venv_ready(nrf52_dirs.python_env)

        check_and_install()

        mock_nrf52_ops.create_venv.assert_not_called()
        assert _subcommands(mock_nrf52_ops.run_command_ok) == [
            "init",
            "config",
            "update",
            "list",
            "install",
        ]
        # minimal SDK + per-arch toolchain
        assert mock_nrf52_ops.download_from_mirrors.call_count == 2

    def test_toolchain_only_missing(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """Venv and framework ready → only toolchain downloaded and extracted."""
        _mark_venv_ready(nrf52_dirs.python_env)
        (nrf52_dirs.python_env / ".zephyr_reqs_ready").touch()
        (nrf52_dirs.framework / ".ready").touch()

        check_and_install()

        mock_nrf52_ops.create_venv.assert_not_called()
        mock_nrf52_ops.run_command_ok.assert_not_called()
        # minimal SDK + per-arch toolchain
        assert mock_nrf52_ops.download_from_mirrors.call_count == 2
        assert mock_nrf52_ops.archive_extract_all.call_count == 2

    def test_framework_clone_is_shallow(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """Both the manifest repository and every project are fetched at depth 1."""
        _mark_venv_ready(nrf52_dirs.python_env)

        check_and_install()

        init, _, update = mock_nrf52_ops.run_command_ok.call_args_list[:3]
        assert "-o=--depth=1" in init.args[0]
        assert "--fetch-opt=--depth=1" in update.args[0]
        # Streamed, so the long clone's progress reaches the log
        assert init.kwargs["stream_output"] is True
        assert update.kwargs["stream_output"] is True

    def test_interrupted_download_resumes(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """A workspace left by a cut-short download is updated in place, not
        wiped and cloned again."""
        _mark_venv_ready(nrf52_dirs.python_env)
        _mark_west_initialized(nrf52_dirs.framework)

        # A marker left by an earlier failed resume
        (nrf52_dirs.framework / ".resume_failed").touch()

        check_and_install()

        assert (
            call(nrf52_dirs.framework, msg=ANY)
            not in mock_nrf52_ops.rmdir.call_args_list
        )
        assert not (nrf52_dirs.framework / ".resume_failed").exists()
        # west update in the workspace (no init), then pip install zephyr reqs
        assert _subcommands(mock_nrf52_ops.run_command_ok) == [
            "config",
            "update",
            "list",
            "install",
        ]
        update = mock_nrf52_ops.run_command_ok.call_args_list[1]
        assert update.kwargs["cwd"] == nrf52_dirs.framework
        assert (nrf52_dirs.framework / ".ready").exists()

    def test_failed_resume_keeps_the_download_once(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """A first failed resume keeps what was fetched (the network likely
        dropped again) and is retried on the next build."""
        _mark_venv_ready(nrf52_dirs.python_env)
        _mark_west_initialized(nrf52_dirs.framework)
        # config succeeds, the resumed update fails
        mock_nrf52_ops.run_command_ok.side_effect = [True, False]

        with pytest.raises(EsphomeError, match="Can't resume"):
            check_and_install()

        assert (
            call(nrf52_dirs.framework, msg=ANY)
            not in mock_nrf52_ops.rmdir.call_args_list
        )
        assert (nrf52_dirs.framework / ".resume_failed").exists()

    def test_cut_short_init_starts_over(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """A ``.west`` without its config (init cut short) clones clean."""
        _mark_venv_ready(nrf52_dirs.python_env)
        (nrf52_dirs.framework / ".west").mkdir()

        check_and_install()

        mock_nrf52_ops.rmdir.assert_any_call(nrf52_dirs.framework, msg=ANY)
        first = mock_nrf52_ops.run_command_ok.call_args_list[0]
        assert "init" in first.args[0]

    def test_second_failed_resume_starts_over(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """A resume failing twice in a row wipes the workspace and clones clean."""
        _mark_venv_ready(nrf52_dirs.python_env)
        _mark_west_initialized(nrf52_dirs.framework)
        (nrf52_dirs.framework / ".resume_failed").touch()
        # resumed update fails; the clean clone and zephyr reqs succeed
        mock_nrf52_ops.run_command_ok.side_effect = [True, False, *[True] * 5]

        check_and_install()

        mock_nrf52_ops.rmdir.assert_any_call(nrf52_dirs.framework, msg=ANY)
        assert _subcommands(mock_nrf52_ops.run_command_ok) == [
            "config",
            "update",
            "init",
            "config",
            "update",
            "list",
            "install",
        ]

    def test_requirements_install_failure_raises(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """Failing pip install -r requirements.txt raises EsphomeError."""
        mock_nrf52_ops.run_command_ok.return_value = False

        with pytest.raises(EsphomeError, match="Install requirements"):
            check_and_install()

    def test_framework_init_failure_raises(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """Failing west init raises EsphomeError."""
        _mark_venv_ready(nrf52_dirs.python_env)
        mock_nrf52_ops.run_command_ok.return_value = False

        with pytest.raises(EsphomeError, match="Can't initialize"):
            check_and_install()

    def test_framework_update_failure_raises(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """Failing west update raises EsphomeError."""
        _mark_venv_ready(nrf52_dirs.python_env)
        # init and config succeed, update fails
        mock_nrf52_ops.run_command_ok.side_effect = [True, True, False]

        with pytest.raises(EsphomeError, match="Can't update"):
            check_and_install()

    def test_fresh_install_fetches_only_default_projects(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """A fresh clone leaves every west project out except the defaults."""
        _mark_venv_ready(nrf52_dirs.python_env)

        check_and_install()

        assert _project_filter(mock_nrf52_ops.run_command_ok) == (
            "-.*,+cmsis,+hal_nordic,+nrfxlib,+zephyr"
        )
        stamp = nrf52_dirs.framework / ".west_projects"
        assert stamp.read_text(encoding="utf-8").split() == sorted(
            DEFAULT_WEST_PROJECTS
        )

    def test_included_project_joins_the_filter(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """A project a component includes is fetched with the defaults."""
        _mark_venv_ready(nrf52_dirs.python_env)
        include_west_project("oberon-psa-crypto")

        check_and_install()

        assert "+oberon-psa-crypto" in _project_filter(
            mock_nrf52_ops.run_command_ok
        ).split(",")

    def test_installed_sdk_fetches_a_newly_needed_project(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """A finished install gains a project another config left out, keeping
        what it already has."""
        _mark_installed(nrf52_dirs)
        (nrf52_dirs.framework / ".west_projects").write_text(
            "cmsis\nhal_nordic\nnrfxlib\ntinycrypt\nzephyr", encoding="utf-8"
        )
        include_west_project("openthread")

        check_and_install()

        # The names are checked before anything is fetched
        assert _subcommands(mock_nrf52_ops.run_command_ok) == [
            "list",
            "config",
            "update",
        ]
        wanted = "-.*,+cmsis,+hal_nordic,+nrfxlib,+openthread,+tinycrypt,+zephyr"
        assert _project_filter(mock_nrf52_ops.run_command_ok) == wanted
        mock_nrf52_ops.rmdir.assert_not_called()

    @pytest.mark.parametrize(
        "stamp",
        [
            pytest.param(None, id="install_from_before_the_filter"),
            pytest.param(
                "cmsis\nhal_nordic\nnrfxlib\nopenthread\nzephyr",
                id="project_already_fetched",
            ),
        ],
    )
    def test_installed_sdk_with_the_project_fetches_nothing(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
        stamp: str | None,
    ) -> None:
        """No fetch when the install already has every wanted project; an
        install without the stamp (west config, no filter) has them all. The
        names are still checked, which only reads the manifest."""
        _mark_installed(nrf52_dirs)
        _mark_west_initialized(nrf52_dirs.framework)
        if stamp is not None:
            (nrf52_dirs.framework / ".west_projects").write_text(
                stamp, encoding="utf-8"
            )
        include_west_project("openthread")

        check_and_install()

        assert _subcommands(mock_nrf52_ops.run_command_ok) == ["list"]

    def test_sysbuild_fetches_mcuboot(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """Sysbuild always builds the MCUboot image, so it needs the project."""
        _mark_venv_ready(nrf52_dirs.python_env)
        CORE.data[KEY_ZEPHYR] = {KEY_SYSBUILD: True}

        check_and_install()

        assert "+mcuboot" in _project_filter(mock_nrf52_ops.run_command_ok).split(",")

    @pytest.mark.parametrize(
        ("sdk_version", "has_cmsis_6"),
        [("2.9.2", False), ("3.1.0", True), ("3.2.0", True)],
    )
    def test_sdk_3_1_and_later_want_cmsis_6(
        self, setup_core: Path, sdk_version: str, has_cmsis_6: bool
    ) -> None:
        """Zephyr 4.1 moved the Cortex-M core headers to the cmsis_6 module."""
        CORE.data[KEY_CORE] = {KEY_FRAMEWORK_VERSION: Version.parse(sdk_version)}

        assert ("cmsis_6" in _wanted_west_projects()) is has_cmsis_6

    def test_default_projects_never_read_the_stamp(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """A build wanting only the defaults has them on any install; an
        unreadable stamp shows the check never looked."""
        _mark_installed(nrf52_dirs)
        (nrf52_dirs.framework / ".west_projects").mkdir()

        check_and_install()

        mock_nrf52_ops.run_command_ok.assert_not_called()

    def test_failed_project_fetch_raises(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """A failed fetch of a newly needed project raises, keeps the stamp and
        puts the workspace filter back to the stamp's projects and the defaults."""
        _mark_venv_ready(nrf52_dirs.python_env)
        (nrf52_dirs.framework / ".ready").touch()
        stamp = nrf52_dirs.framework / ".west_projects"
        stamp.write_text("zephyr", encoding="utf-8")
        include_west_project("openthread")
        # list and config succeed, update fails, the restoring config succeeds
        mock_nrf52_ops.run_command_ok.side_effect = [True, True, False, True]

        with pytest.raises(EsphomeError, match="Can't update"):
            check_and_install()

        assert stamp.read_text(encoding="utf-8") == "zephyr"
        assert _subcommands(mock_nrf52_ops.run_command_ok) == [
            "list",
            "config",
            "update",
            "config",
        ]
        # The defaults always stay in the restored filter
        assert _project_filter(mock_nrf52_ops.run_command_ok) == _DEFAULTS_FILTER

    def test_failed_fetch_with_a_lost_stamp_keeps_the_defaults_active(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """An unknown installed set must not leave a filter with every module off."""
        _mark_installed(nrf52_dirs)
        _mark_west_initialized(nrf52_dirs.framework)
        (nrf52_dirs.framework / ".west" / "config").write_text(
            "[manifest]\nproject-filter = -.*,+zephyr\n", encoding="utf-8"
        )
        include_west_project("openthread")
        # list and config succeed, update fails, the restoring config succeeds
        mock_nrf52_ops.run_command_ok.side_effect = [True, True, False, True]

        with pytest.raises(EsphomeError, match="Can't update"):
            check_and_install()

        assert _project_filter(mock_nrf52_ops.run_command_ok) == _DEFAULTS_FILTER

    def test_failed_filter_restore_is_logged(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
        caplog: pytest.LogCaptureFixture,
    ) -> None:
        """When the filter can't be put back after a failed fetch, the user is told."""
        _mark_venv_ready(nrf52_dirs.python_env)
        (nrf52_dirs.framework / ".ready").touch()
        (nrf52_dirs.framework / ".west_projects").write_text("zephyr", encoding="utf-8")
        include_west_project("openthread")
        # list and config succeed, update fails, the restoring config fails too
        mock_nrf52_ops.run_command_ok.side_effect = [True, True, False, False]

        with pytest.raises(EsphomeError, match="Can't update"):
            check_and_install()

        assert "Couldn't put the nRF Connect SDK" in caplog.text

    def test_lost_stamp_on_a_filtered_install_fetches_again(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """A workspace with a project filter but no stamp is a filtered install
        whose record was lost, so the wanted projects are fetched, not assumed."""
        _mark_installed(nrf52_dirs)
        _mark_west_initialized(nrf52_dirs.framework)
        (nrf52_dirs.framework / ".west" / "config").write_text(
            "[manifest]\nproject-filter = -.*,+zephyr\n", encoding="utf-8"
        )
        include_west_project("openthread")

        check_and_install()

        assert _subcommands(mock_nrf52_ops.run_command_ok) == [
            "list",
            "config",
            "update",
        ]

    def test_install_waits_for_another_build_holding_the_lock(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
        caplog: pytest.LogCaptureFixture,
    ) -> None:
        """A held lock is waited on, never mistaken for a filesystem that
        cannot lock (filelock's Timeout is an OSError as well)."""
        from filelock import Timeout

        _mark_installed(nrf52_dirs)

        with (
            caplog.at_level("INFO"),
            patch("filelock.FileLock") as file_lock,
        ):
            file_lock.return_value.acquire.side_effect = [Timeout("install.lock"), None]
            check_and_install()

        assert file_lock.return_value.acquire.call_count == 2
        assert "Waiting for another build" in caplog.text
        assert "continuing without a lock" not in caplog.text

    def test_install_from_before_the_filter_still_checks_the_names(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """An install with every project fetches nothing, but an unknown name
        is still rejected so a mistake shows on every install alike."""
        _mark_installed(nrf52_dirs)
        _mark_west_initialized(nrf52_dirs.framework)
        include_west_project("openthread")

        check_and_install()

        assert _subcommands(mock_nrf52_ops.run_command_ok) == ["list"]

        mock_nrf52_ops.run_command_ok.reset_mock()
        mock_nrf52_ops.run_command_ok.return_value = False
        include_west_project("no_such_project")
        with pytest.raises(EsphomeError, match="west list failed"):
            check_and_install()

    def test_install_lock_is_per_sdk_version(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """Installs of different SDK versions do not wait on each other; the
        toolchain they share is only locked while it is missing."""
        _mark_installed(nrf52_dirs)

        with patch("filelock.FileLock") as file_lock:
            check_and_install()

        lock_files = [Path(c.args[0]).name for c in file_lock.call_args_list]
        assert lock_files == [f"sdk-v{_TEST_SDK_VERSION}.lock"]

    def test_install_runs_unlocked_where_the_filesystem_cannot_lock(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
        caplog: pytest.LogCaptureFixture,
    ) -> None:
        """No soft-lock fallback: its marker outlives a killed build and would
        hang every later one, so the install goes ahead without a lock."""
        _mark_installed(nrf52_dirs)

        with patch("filelock.FileLock") as file_lock:
            file_lock.return_value.acquire.side_effect = OSError(
                errno.ENOSYS, "Function not implemented"
            )
            check_and_install()

        assert file_lock.call_args.kwargs == {"fallback_to_soft": False}
        assert "continuing without a lock" in caplog.text

    def test_unknown_project_raises_before_anything_is_written(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """A filter naming a project the manifest lacks fetches nothing, so
        the names are checked once the update resolved the manifest, and an
        unknown one is never recorded as installed."""
        _mark_venv_ready(nrf52_dirs.python_env)
        include_west_project("no_such_project")
        # init, config and update succeed, list fails
        mock_nrf52_ops.run_command_ok.side_effect = [True, True, True, False]

        with pytest.raises(EsphomeError, match="west list failed .*no_such_project"):
            check_and_install()

        assert _subcommands(mock_nrf52_ops.run_command_ok) == [
            "init",
            "config",
            "update",
            "list",
        ]
        assert "no_such_project" in mock_nrf52_ops.run_command_ok.call_args.args[0]
        assert not (nrf52_dirs.framework / ".west_projects").exists()

    def test_unknown_project_on_an_installed_sdk_fetches_nothing(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """On a finished install the names are checked first, so an unknown one
        costs no fetch and leaves the workspace as it was."""
        _mark_venv_ready(nrf52_dirs.python_env)
        (nrf52_dirs.framework / ".ready").touch()
        (nrf52_dirs.framework / ".west_projects").write_text("zephyr", encoding="utf-8")
        include_west_project("no_such_project")
        # list fails before anything is fetched or changed
        mock_nrf52_ops.run_command_ok.side_effect = [False]

        with pytest.raises(EsphomeError, match="west list failed"):
            check_and_install()

        assert _subcommands(mock_nrf52_ops.run_command_ok) == ["list"]
        assert (nrf52_dirs.framework / ".west_projects").read_text(
            encoding="utf-8"
        ) == "zephyr"

    def test_missing_stamp_and_west_config_fetches_again(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """With neither the stamp nor .west/config left, nothing says what the
        install has, so the wanted projects are fetched rather than assumed."""
        _mark_installed(nrf52_dirs)
        include_west_project("openthread")

        check_and_install()

        assert _subcommands(mock_nrf52_ops.run_command_ok) == [
            "list",
            "config",
            "update",
        ]

    def test_toolchain_download_passes_platform_substitutions(
        self,
        nrf52_dirs: SimpleNamespace,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """download_from_mirrors receives VERSION + platform triple from _get_toolchain_platform_info."""
        _mark_venv_ready(nrf52_dirs.python_env)
        (nrf52_dirs.framework / ".ready").touch()

        with patch(
            "esphome.components.nrf52.framework._get_toolchain_platform_info",
            return_value=("linux", "x86_64", "tar.xz"),
        ):
            check_and_install()

        args, _ = mock_nrf52_ops.download_from_mirrors.call_args
        substitutions = args[1]
        assert substitutions["VERSION"] == TOOLCHAIN_VERSION
        assert substitutions["sysname"] == "linux"
        assert substitutions["machine"] == "x86_64"
        assert substitutions["extension"] == "tar.xz"


# ---------------------------------------------------------------------------
# setup_platformio_python_env tests
# ---------------------------------------------------------------------------


def _platformio_requirements_hash() -> str:
    return hashlib.sha256(
        _REQUIREMENTS.read_bytes()
        + "\n".join(_PLATFORMIO_PENV_REQUIREMENTS).encode()
        + f"python{sys.version_info.major}.{sys.version_info.minor}".encode()
    ).hexdigest()


@pytest.fixture
def platformio_penv_dir() -> Path:
    """Pre-create the PlatformIO penv dir so sentinel writes succeed.

    create_venv is mocked in these tests, so the directory it would have
    created must exist for ``sentinel.write_text`` to work.
    """
    penv_path = _get_platformio_penv_path()
    penv_path.mkdir(parents=True, exist_ok=True)
    return penv_path


class TestSetupPlatformioPythonEnv:
    def test_fresh_install_creates_venv_and_sets_env(
        self,
        platformio_penv_dir: Path,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """No sentinel → venv created, requirements installed, env exported."""
        with patch.dict(os.environ):
            os.environ.pop("PYTHONPATH", None)

            setup_platformio_python_env()

            mock_nrf52_ops.rmdir.assert_called_once()
            mock_nrf52_ops.create_venv.assert_called_once_with(
                platformio_penv_dir, msg="PlatformIO toolchain"
            )
            mock_nrf52_ops.run_command_ok.assert_called_once()
            cmd = mock_nrf52_ops.run_command_ok.call_args[0][0]
            assert cmd[1:4] == ["-m", "pip", "install"]
            assert "-r" in cmd
            assert str(_REQUIREMENTS) in cmd
            for requirement in _PLATFORMIO_PENV_REQUIREMENTS:
                assert requirement in cmd
            sentinel = platformio_penv_dir / ".ready"
            assert sentinel.read_text(encoding="utf-8") == (
                _platformio_requirements_hash()
            )

            assert os.environ["VIRTUAL_ENV"] == str(platformio_penv_dir)
            site_packages = str(_get_penv_site_packages(platformio_penv_dir))
            assert os.environ["PYTHONPATH"] == site_packages
            bin_dir = str(
                get_python_env_executable_path(platformio_penv_dir, "python").parent
            )
            assert os.environ["PATH"].split(os.pathsep)[0] == bin_dir

    def test_ready_sentinel_skips_install_but_sets_env(
        self,
        platformio_penv_dir: Path,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """Current sentinel → no install work, env vars still exported."""
        (platformio_penv_dir / ".ready").write_text(
            _platformio_requirements_hash(), encoding="utf-8"
        )
        _touch_penv_python(platformio_penv_dir)

        with patch.dict(os.environ):
            setup_platformio_python_env()

            mock_nrf52_ops.rmdir.assert_not_called()
            mock_nrf52_ops.create_venv.assert_not_called()
            mock_nrf52_ops.run_command_ok.assert_not_called()
            assert os.environ["VIRTUAL_ENV"] == str(platformio_penv_dir)

    def test_stale_sentinel_reinstalls(
        self,
        platformio_penv_dir: Path,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """A sentinel from different requirements → venv rebuilt from scratch."""
        sentinel = platformio_penv_dir / ".ready"
        sentinel.write_text("stale-hash", encoding="utf-8")

        with patch.dict(os.environ):
            setup_platformio_python_env()

        mock_nrf52_ops.rmdir.assert_called_once()
        mock_nrf52_ops.create_venv.assert_called_once()
        mock_nrf52_ops.run_command_ok.assert_called_once()
        assert sentinel.read_text(encoding="utf-8") == _platformio_requirements_hash()

    def test_install_failure_raises(
        self,
        platformio_penv_dir: Path,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """Failing pip install raises EsphomeError and writes no sentinel."""
        mock_nrf52_ops.run_command_ok.return_value = False

        with (
            patch.dict(os.environ),
            pytest.raises(
                EsphomeError, match="Install requirements for PlatformIO toolchain"
            ),
        ):
            setup_platformio_python_env()

        assert not (platformio_penv_dir / ".ready").exists()

    def test_missing_interpreter_reinstalls(
        self,
        platformio_penv_dir: Path,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """A valid sentinel must not mask a missing interpreter."""
        (platformio_penv_dir / ".ready").write_text(
            _platformio_requirements_hash(), encoding="utf-8"
        )
        # no interpreter on disk

        with patch.dict(os.environ):
            setup_platformio_python_env()

        mock_nrf52_ops.create_venv.assert_called_once()

    def test_repeated_calls_do_not_duplicate_env_entries(
        self,
        platformio_penv_dir: Path,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """Compile then upload in one process must not grow PYTHONPATH/PATH."""
        (platformio_penv_dir / ".ready").write_text(
            _platformio_requirements_hash(), encoding="utf-8"
        )
        _touch_penv_python(platformio_penv_dir)
        site_packages = str(_get_penv_site_packages(platformio_penv_dir))
        bin_dir = str(
            get_python_env_executable_path(platformio_penv_dir, "python").parent
        )

        with patch.dict(os.environ):
            setup_platformio_python_env()
            setup_platformio_python_env()

            assert os.environ["PYTHONPATH"].split(os.pathsep).count(site_packages) == 1
            assert os.environ["PATH"].split(os.pathsep).count(bin_dir) == 1

    def test_existing_pythonpath_preserved(
        self,
        platformio_penv_dir: Path,
        mock_nrf52_ops: SimpleNamespace,
    ) -> None:
        """A pre-existing PYTHONPATH keeps its entries after the venv entry."""
        (platformio_penv_dir / ".ready").write_text(
            _platformio_requirements_hash(), encoding="utf-8"
        )
        _touch_penv_python(platformio_penv_dir)
        site_packages = str(_get_penv_site_packages(platformio_penv_dir))

        with patch.dict(os.environ, {"PYTHONPATH": "/existing/path"}):
            setup_platformio_python_env()

            assert os.environ["PYTHONPATH"] == os.pathsep.join(
                [site_packages, "/existing/path"]
            )


@pytest.mark.parametrize(
    ("os_name", "expected_parts"),
    [
        (
            "posix",
            (
                "lib",
                f"python{sys.version_info.major}.{sys.version_info.minor}",
                "site-packages",
            ),
        ),
        ("nt", ("Lib", "site-packages")),
    ],
)
def test_get_penv_site_packages(
    tmp_path: Path, os_name: str, expected_parts: tuple[str, ...]
) -> None:
    penv_path = tmp_path / "penv"
    with patch("os.name", os_name):
        assert _get_penv_site_packages(penv_path) == penv_path.joinpath(*expected_parts)


# ---------------------------------------------------------------------------
# get_build_env tests
# ---------------------------------------------------------------------------


def test_get_build_env(
    nrf52_dirs: SimpleNamespace, monkeypatch: pytest.MonkeyPatch
) -> None:
    """get_build_env exposes ZEPHYR_SDK_INSTALL_DIR pointing at the toolchain root.

    ZEPHYR_SDK_INSTALL_DIR is the variable Zephyr's FindZephyr-sdk.cmake
    explicitly consumes (from the environment) and uses as a find_package
    HINT. The old Zephyr-sdk_DIR environment hint proved unreliable in
    containerized non-root builds and was removed.
    """
    monkeypatch.setenv("SOME_PREEXISTING_VAR", "kept")

    env = get_build_env()

    tools = get_sdk_nrf_tools_path()
    venv_bin_dir = get_python_env_executable_path(
        tools / "penvs" / f"v{_TEST_SDK_VERSION}", "python"
    ).parent
    assert env["PATH"].startswith(str(venv_bin_dir) + os.pathsep)
    assert env["ZEPHYR_BASE"] == str(
        tools / "frameworks" / f"v{_TEST_SDK_VERSION}" / "zephyr"
    )
    # Toolchain root, not the cmake/ subdir
    assert env["ZEPHYR_SDK_INSTALL_DIR"] == str(
        tools / "toolchains" / TOOLCHAIN_VERSION
    )
    assert "Zephyr-sdk_DIR" not in env
    # The rest of the process environment is inherited
    assert env["SOME_PREEXISTING_VAR"] == "kept"


# ---------------------------------------------------------------------------
# get_sdk_nrf_tools_path tests
# ---------------------------------------------------------------------------


def testget_tools_path_env_override(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    override = tmp_path / "custom" / "sdk-nrf"
    monkeypatch.setenv("ESPHOME_SDK_NRF_PREFIX", str(override))
    assert get_sdk_nrf_tools_path() == override.resolve()


@pytest.mark.parametrize("value", ["", "   "])
def testget_tools_path_blank_env_falls_back_to_default(
    value: str, monkeypatch: pytest.MonkeyPatch
) -> None:
    """A blank ESPHOME_SDK_NRF_PREFIX is treated as unset, not as CWD.

    Path("") would resolve to the working directory, which clean-all could
    then delete by accident.
    """

    monkeypatch.setenv("ESPHOME_SDK_NRF_PREFIX", value)
    expected = (
        Path(platformdirs.user_cache_dir("esphome", appauthor=False)) / "sdk-nrf"
    ).resolve()
    assert get_sdk_nrf_tools_path() == expected


def testget_tools_path_default_is_global_cache(
    monkeypatch: pytest.MonkeyPatch,
) -> None:

    monkeypatch.delenv("ESPHOME_SDK_NRF_PREFIX", raising=False)
    expected = (
        Path(platformdirs.user_cache_dir("esphome", appauthor=False)) / "sdk-nrf"
    ).resolve()
    assert get_sdk_nrf_tools_path() == expected


def test_needs_venv_rebuild_gates(tmp_path: Path) -> None:
    """The shared penv gate rebuilds on any missing or stale piece."""
    penv = tmp_path / "penv"
    penv.mkdir()
    python = penv / "python"
    sentinel = penv / ".ready"
    good_hash = "abc123"

    # Nothing in place yet
    assert _needs_venv_rebuild(python, sentinel, good_hash)

    python.write_text("")
    # Interpreter present but no sentinel
    assert _needs_venv_rebuild(python, sentinel, good_hash)

    sentinel.write_text(good_hash, encoding="utf-8")
    # Everything in place
    assert not _needs_venv_rebuild(python, sentinel, good_hash)

    # Stale requirements hash
    assert _needs_venv_rebuild(python, sentinel, "otherhash")


@pytest.mark.skipif(
    sys.platform == "win32", reason="symlink creation needs privileges on Windows"
)
def test_needs_venv_rebuild_on_dangling_interpreter_symlink(tmp_path: Path) -> None:
    """A cached venv restored after a host interpreter upgrade has a
    bin/python symlink whose target is gone; the valid sentinel must not
    mask it."""
    penv = tmp_path / "penv"
    penv.mkdir()
    python = penv / "python"
    sentinel = penv / ".ready"
    sentinel.write_text("abc123", encoding="utf-8")
    python.symlink_to(tmp_path / "hostedtoolcache" / "3.12.14" / "python3")
    assert python.is_symlink()
    assert not python.exists()

    assert _needs_venv_rebuild(python, sentinel, "abc123")


def test_resolve_toolchain_rejects_unsupported() -> None:
    """A --toolchain nRF52 cannot serve fails instead of degrading silently."""

    CORE.toolchain = Toolchain.ARDUINO
    with pytest.raises(cv.Invalid, match="Unsupported toolchain 'arduino'"):
        _resolve_toolchain({})