esphome/tests/unit_tests/test_espidf_framework.py

"""Tests for esphome.espidf.framework helpers."""

# pylint: disable=protected-access

import io
import json
from pathlib import Path
import tarfile
from types import SimpleNamespace
from unittest.mock import patch

import pytest

from esphome.espidf.framework import (
    _check_stamp,
    _clone_idf_with_submodules,
    _get_framework_path,
    _get_idf_tool_paths,
    _get_idf_tools_path,
    _get_idf_version,
    _get_python_env_path,
    _get_python_version,
    _parse_git_source,
    _patch_tools_json_for_linux_arm64,
    _write_idf_version_txt,
    _write_stamp,
    check_esp_idf_install,
    get_framework_env,
)
from esphome.framework_helpers import _tar_extract_all, get_python_env_executable_path


@pytest.mark.parametrize(
    ("source", "expected"),
    [
        # github:// shorthand
        (
            "github://espressif/esp-idf",
            ("https://github.com/espressif/esp-idf.git", None),
        ),
        (
            "github://espressif/esp-idf@master",
            ("https://github.com/espressif/esp-idf.git", "master"),
        ),
        (
            "github://espressif/esp-idf@release/v6.0",
            ("https://github.com/espressif/esp-idf.git", "release/v6.0"),
        ),
        # explicit https://github.com/...git URL
        (
            "https://github.com/espressif/esp-idf.git",
            ("https://github.com/espressif/esp-idf.git", None),
        ),
        (
            "https://github.com/espressif/esp-idf.git@master",
            ("https://github.com/espressif/esp-idf.git", "master"),
        ),
        (
            "https://github.com/espressif/esp-idf.git@v6.0.1",
            ("https://github.com/espressif/esp-idf.git", "v6.0.1"),
        ),
        # Tolerate a trailing ".git" on the shorthand so the user doesn't
        # silently end up with a doubled "...esp-idf.git.git" URL.
        (
            "github://espressif/esp-idf.git",
            ("https://github.com/espressif/esp-idf.git", None),
        ),
        (
            "github://espressif/esp-idf.git@master",
            ("https://github.com/espressif/esp-idf.git", "master"),
        ),
    ],
)
def test_parse_git_source_recognized(
    source: str, expected: tuple[str, str | None]
) -> None:
    assert _parse_git_source(source) == expected


@pytest.mark.parametrize(
    "source",
    [
        # archive URLs fall through to the existing download path
        "https://github.com/espressif/esp-idf/archive/refs/heads/master.zip",
        "https://dl.espressif.com/dl/esp-idf/v6.0.1/esp-idf-v6.0.1.zip",
        "https://github.com/esphome-libs/esp-idf/releases/download/v5.5.4/esp-idf-v5.5.4.tar.xz",
        # SSH and other git protocols are intentionally rejected — match
        # external_components, which only recognizes github:// + structured
        # dicts for these.
        "git@github.com:espressif/esp-idf.git",
        "ssh://git@github.com/espressif/esp-idf.git",
        "git://github.com/espressif/esp-idf.git",
        # non-GitHub .git URLs are intentionally rejected for the same reason
        "https://gitlab.com/foo/bar.git",
        "https://github.example.com/foo/bar.git",
    ],
)
def test_parse_git_source_rejected(source: str) -> None:
    assert _parse_git_source(source) is None


def _make_idf_tree(framework_path: Path) -> None:
    """Create the minimum tree _clone_idf_with_submodules sanity-checks for."""
    (framework_path / "tools").mkdir(parents=True)
    (framework_path / "tools" / "idf_tools.py").write_text("# stub\n")


def test_clone_idf_with_submodules_without_ref(tmp_path: Path) -> None:
    framework_path = tmp_path / "idf"
    framework_path.mkdir()
    _make_idf_tree(framework_path)

    with patch("esphome.git.run_git_command", return_value="") as run_git_command_mock:
        _clone_idf_with_submodules(
            framework_path, "https://github.com/espressif/esp-idf.git", None
        )

    # No ref -> just clone + submodule update, no fetch/reset.
    calls = [c.args[0] for c in run_git_command_mock.call_args_list]
    assert calls[0] == [
        "git",
        "clone",
        "--depth=1",
        "--",
        "https://github.com/espressif/esp-idf.git",
        str(framework_path),
    ]
    assert calls[-1][:5] == ["git", "submodule", "update", "--init", "--recursive"]
    assert not any(c[1] == "fetch" for c in calls)
    assert not any(c[1] == "reset" for c in calls)


def test_clone_idf_with_submodules_with_ref(tmp_path: Path) -> None:
    framework_path = tmp_path / "idf"
    framework_path.mkdir()
    _make_idf_tree(framework_path)

    with patch("esphome.git.run_git_command", return_value="") as run_git_command_mock:
        _clone_idf_with_submodules(
            framework_path,
            "https://github.com/espressif/esp-idf.git",
            "master",
        )

    calls = [c.args[0] for c in run_git_command_mock.call_args_list]
    # clone, fetch ref, reset hard, submodule update
    assert calls[0][:2] == ["git", "clone"]
    assert calls[1] == [
        "git",
        "fetch",
        "--depth=1",
        "--",
        "origin",
        "master",
    ]
    assert calls[2] == ["git", "reset", "--hard", "FETCH_HEAD"]
    assert calls[3][:5] == ["git", "submodule", "update", "--init", "--recursive"]


def test_clone_idf_with_submodules_raises_when_tree_missing(
    tmp_path: Path,
) -> None:
    framework_path = tmp_path / "idf"
    framework_path.mkdir()
    # Deliberately do NOT call _make_idf_tree — simulate a clone that
    # returned 0 but produced no tools/idf_tools.py.

    with (
        patch("esphome.git.run_git_command", return_value=""),
        pytest.raises(RuntimeError, match="no usable ESP-IDF tree"),
    ):
        _clone_idf_with_submodules(
            framework_path,
            "https://github.com/espressif/esp-idf.git",
            None,
        )


# ---------------------------------------------------------------------------
# Helpers for _tar_extract_all hard-link prefix-stripping tests
# ---------------------------------------------------------------------------


def _make_tar(
    members: list[tarfile.TarInfo], file_contents: dict[str, bytes]
) -> io.BytesIO:
    """Build an in-memory tar archive from a list of TarInfo objects."""
    buf = io.BytesIO()
    with tarfile.open(fileobj=buf, mode="w") as tf:
        for info in members:
            if info.isreg() and info.name in file_contents:
                data = file_contents[info.name]
                info.size = len(data)
                tf.addfile(info, io.BytesIO(data))
            else:
                tf.addfile(info)
    buf.seek(0)
    return buf


def _regular(name: str) -> tarfile.TarInfo:
    info = tarfile.TarInfo(name=name)
    info.type = tarfile.REGTYPE
    info.size = 0
    info.mode = 0o644
    return info


def _hardlink(name: str, linkname: str) -> tarfile.TarInfo:
    info = tarfile.TarInfo(name=name)
    info.type = tarfile.LNKTYPE
    info.linkname = linkname
    info.size = 0
    info.mode = 0o644
    return info


class TestTarExtractHardLinkPrefixStripping:
    """
    Covers the hard-link prefix-stripping block in _tar_extract_all (L528-541).

    Archive layout used by every test:

        wrapper/                   ← single top-level wrapper dir (stripped)
        wrapper/target.txt         ← regular file; becomes target.txt in dest
        wrapper/link_good          ← hard link to wrapper/target.txt  (kept, linkname stripped)
        wrapper/link_exact_root    ← hard link to "wrapper"            (skipped – equals strip_root)
        wrapper/link_exact_prefix  ← hard link to "wrapper/"           (skipped – equals strip_prefix)
        wrapper/link_outside       ← hard link to "other/target.txt"   (skipped – not under prefix)
    """

    WRAPPER = "wrapper"

    def _build_archive(self) -> io.BytesIO:
        members = [
            _regular(f"{self.WRAPPER}/"),
            _regular(f"{self.WRAPPER}/target.txt"),
            _hardlink(f"{self.WRAPPER}/link_good", f"{self.WRAPPER}/target.txt"),
            _hardlink(f"{self.WRAPPER}/link_exact_root", self.WRAPPER),
            _hardlink(f"{self.WRAPPER}/link_exact_prefix", f"{self.WRAPPER}/"),
            _hardlink(f"{self.WRAPPER}/link_outside", "other/target.txt"),
        ]
        return _make_tar(members, {f"{self.WRAPPER}/target.txt": b"hello"})

    def test_good_hardlink_is_extracted_with_stripped_linkname(
        self, tmp_path: Path
    ) -> None:
        """Hard link whose linkname starts with wrapper/ is extracted and its
        linkname has the prefix removed so tarfile can resolve the target."""
        _tar_extract_all(self._build_archive(), tmp_path)
        link = tmp_path / "link_good"
        assert link.exists(), "link_good should have been extracted"
        assert link.read_bytes() == b"hello"

    def test_hardlink_equal_to_strip_root_is_skipped(self, tmp_path: Path) -> None:
        """Hard link whose linkname equals strip_root exactly must be dropped."""
        _tar_extract_all(self._build_archive(), tmp_path)
        assert not (tmp_path / "link_exact_root").exists()

    def test_hardlink_equal_to_strip_prefix_is_skipped(self, tmp_path: Path) -> None:
        """Hard link whose linkname equals strip_prefix (strip_root + '/') must be dropped."""
        _tar_extract_all(self._build_archive(), tmp_path)
        assert not (tmp_path / "link_exact_prefix").exists()

    def test_hardlink_outside_prefix_is_skipped(self, tmp_path: Path) -> None:
        """Hard link whose linkname does not start with wrapper/ must be dropped."""
        _tar_extract_all(self._build_archive(), tmp_path)
        assert not (tmp_path / "link_outside").exists()

    def test_regular_file_and_no_spurious_files(self, tmp_path: Path) -> None:
        """Sanity check: target.txt is extracted and no unexpected files appear."""
        _tar_extract_all(self._build_archive(), tmp_path)
        assert (tmp_path / "target.txt").read_bytes() == b"hello"
        extracted = {p.name for p in tmp_path.iterdir()}
        assert extracted == {"target.txt", "link_good"}


_IDF_VERSION = "5.1.2"


@pytest.fixture
def espidf_mocks(setup_core: Path):
    """Patch the heavy I/O of check_esp_idf_install and pre-create the framework dir."""
    # archive_extract_all is mocked, so pre-create the framework dir that the
    # extracted-marker touch writes into.
    _get_framework_path(_IDF_VERSION).mkdir(parents=True, exist_ok=True)
    with (
        patch("esphome.espidf.framework.rmdir"),
        patch(
            "esphome.espidf.framework.download_from_mirrors",
            return_value="https://example.com/idf.tar.xz",
        ) as download,
        patch("esphome.espidf.framework.archive_extract_all") as extract,
        patch("esphome.espidf.framework.create_venv") as venv,
        patch("esphome.espidf.framework.run_command_ok", return_value=True) as run_ok,
        patch("esphome.espidf.framework._clone_idf_with_submodules") as clone,
        patch("esphome.espidf.framework._write_idf_version_txt"),
        patch("esphome.espidf.framework._patch_tools_json_for_linux_arm64"),
        patch("esphome.espidf.framework._write_stamp"),
        patch("esphome.espidf.framework._check_stamp", return_value=True),
        patch("esphome.espidf.framework._get_idf_version", return_value=_IDF_VERSION),
        patch("esphome.espidf.framework._get_python_version", return_value="3.11.0"),
        patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
    ):
        yield SimpleNamespace(
            download=download, extract=extract, venv=venv, run_ok=run_ok, clone=clone
        )


def test_check_esp_idf_install_fresh(espidf_mocks: SimpleNamespace) -> None:
    """A forced install drives download/extract, venv creation, and pip installs."""
    framework_path, python_env_path = check_esp_idf_install(_IDF_VERSION, force=True)

    assert framework_path == _get_framework_path(_IDF_VERSION)
    assert python_env_path == _get_python_env_path(_IDF_VERSION)
    # framework tarball + python-env constraints file are both downloaded
    assert espidf_mocks.download.call_count == 2
    espidf_mocks.extract.assert_called_once()
    espidf_mocks.venv.assert_called_once()
    espidf_mocks.clone.assert_not_called()


def test_check_esp_idf_install_git_source(espidf_mocks: SimpleNamespace) -> None:
    """A git source_url clones instead of downloading; explicit tools skip discovery."""
    check_esp_idf_install(
        _IDF_VERSION,
        force=True,
        source_url="https://github.com/espressif/esp-idf.git",
        tools=["xtensa-esp-elf"],
    )

    espidf_mocks.clone.assert_called_once()
    # framework is cloned, so only the python-env constraints file is downloaded
    assert espidf_mocks.download.call_count == 1


def test_check_esp_idf_install_already_installed(espidf_mocks: SimpleNamespace) -> None:
    """Marker + matching stamps + existing python env → nothing is re-installed."""
    framework_path = _get_framework_path(_IDF_VERSION)
    (framework_path / ".esphome_extracted").touch()
    python_env_path = _get_python_env_path(_IDF_VERSION)
    env_python = get_python_env_executable_path(python_env_path, "python")
    env_python.parent.mkdir(parents=True, exist_ok=True)
    env_python.touch()

    check_esp_idf_install(_IDF_VERSION)

    espidf_mocks.extract.assert_not_called()
    espidf_mocks.venv.assert_not_called()


def test_check_esp_idf_install_framework_failure(espidf_mocks: SimpleNamespace) -> None:
    """A failing idf_tools install raises."""
    espidf_mocks.run_ok.side_effect = [False]
    with pytest.raises(RuntimeError, match="framework installation failure"):
        check_esp_idf_install(_IDF_VERSION, force=True)


def test_check_esp_idf_install_pip_upgrade_failure(
    espidf_mocks: SimpleNamespace,
) -> None:
    """A failing pip upgrade in the python env raises (framework install ok)."""
    espidf_mocks.run_ok.side_effect = [True, False]
    with pytest.raises(RuntimeError, match="Python environment packages failure"):
        check_esp_idf_install(_IDF_VERSION, force=True)


def test_check_esp_idf_install_feature_failure(espidf_mocks: SimpleNamespace) -> None:
    """A failing feature requirements install raises."""
    espidf_mocks.run_ok.side_effect = [True, True, False]
    with pytest.raises(RuntimeError, match="Python dependencies for"):
        check_esp_idf_install(_IDF_VERSION, force=True, features=["fb"])


def _mark_installed() -> None:
    """Create the extracted marker and python-env interpreter so the install
    check takes the already-installed path rather than force-installing."""
    (_get_framework_path(_IDF_VERSION) / ".esphome_extracted").touch()
    env_python = get_python_env_executable_path(
        _get_python_env_path(_IDF_VERSION), "python"
    )
    env_python.parent.mkdir(parents=True, exist_ok=True)
    env_python.touch()


def test_check_esp_idf_install_stamp_mismatch_reinstalls(
    espidf_mocks: SimpleNamespace,
) -> None:
    """A stamp mismatch reinstalls tools (marker present, so no re-extract)."""
    _mark_installed()
    with patch("esphome.espidf.framework._check_stamp", return_value=False):
        check_esp_idf_install(_IDF_VERSION)

    espidf_mocks.extract.assert_not_called()  # marker present -> no re-extract
    espidf_mocks.venv.assert_called_once()  # tools reinstall -> venv rebuilt


def test_check_esp_idf_install_check_command_failure_reinstalls(
    espidf_mocks: SimpleNamespace,
) -> None:
    """A failing idf_tools check reinstalls tools (marker present, no re-extract)."""
    _mark_installed()
    # idf_tools check fails -> install stays True; the later installs succeed.
    espidf_mocks.run_ok.side_effect = [False, True, True, True]
    check_esp_idf_install(_IDF_VERSION, features=["fb"])

    espidf_mocks.extract.assert_not_called()
    espidf_mocks.venv.assert_called_once()


def test_check_esp_idf_install_unknown_python_version_reinstalls(
    espidf_mocks: SimpleNamespace,
) -> None:
    """An undeterminable python version rebuilds the venv (framework stamp still ok)."""
    _mark_installed()
    with patch("esphome.espidf.framework._get_python_version", return_value=None):
        check_esp_idf_install(_IDF_VERSION)

    espidf_mocks.extract.assert_not_called()  # framework stamp matched
    espidf_mocks.venv.assert_called_once()  # python env rebuilt


def test_check_esp_idf_install_python_stamp_mismatch_rebuilds_venv(
    espidf_mocks: SimpleNamespace,
) -> None:
    """Framework stamp matches but the python-env stamp does not -> venv rebuilt."""

    # _check_stamp passes for the framework (no python_version key) and fails
    # for the python env (carries python_version), so only the venv rebuilds.
    def stamp_ok(_stamp_file, info: dict) -> bool:
        return "python_version" not in info

    _mark_installed()
    with patch("esphome.espidf.framework._check_stamp", side_effect=stamp_ok):
        check_esp_idf_install(_IDF_VERSION)

    espidf_mocks.extract.assert_not_called()
    espidf_mocks.venv.assert_called_once()


def test_check_esp_idf_install_unparseable_version(
    espidf_mocks: SimpleNamespace,
) -> None:
    """A non-semver version skips the MAJOR/MINOR substitutions without erroring."""
    bad_version = "main"
    _get_framework_path(bad_version).mkdir(parents=True, exist_ok=True)
    check_esp_idf_install(bad_version, force=True)

    espidf_mocks.extract.assert_called_once()


# ---------------------------------------------------------------------------
# _patch_tools_json_for_linux_arm64 (arm64-only ninja backport)
# ---------------------------------------------------------------------------


def _write_tools_json(framework_path: Path, data: dict) -> Path:
    tools_dir = framework_path / "tools"
    tools_dir.mkdir(parents=True, exist_ok=True)
    tools_json = tools_dir / "tools.json"
    tools_json.write_text(json.dumps(data), encoding="utf-8")
    return tools_json


def test_patch_tools_json_non_aarch64_is_noop(tmp_path: Path) -> None:
    tools_json = _write_tools_json(
        tmp_path, {"tools": [{"name": "ninja", "versions": [{"name": "1.12.1"}]}]}
    )
    before = tools_json.read_text(encoding="utf-8")
    with patch("esphome.espidf.framework.platform.machine", return_value="x86_64"):
        _patch_tools_json_for_linux_arm64(tmp_path)
    assert tools_json.read_text(encoding="utf-8") == before


def test_patch_tools_json_missing_file_is_noop(tmp_path: Path) -> None:
    with patch("esphome.espidf.framework.platform.machine", return_value="aarch64"):
        _patch_tools_json_for_linux_arm64(tmp_path)  # no tools/tools.json present


def test_patch_tools_json_corrupt_file_warns_and_skips(tmp_path: Path) -> None:
    (tmp_path / "tools").mkdir()
    (tmp_path / "tools" / "tools.json").write_text("{ not json", encoding="utf-8")
    with patch("esphome.espidf.framework.platform.machine", return_value="aarch64"):
        _patch_tools_json_for_linux_arm64(tmp_path)  # JSONDecodeError -> skip


def test_patch_tools_json_injects_ninja_arm64(tmp_path: Path) -> None:
    tools_json = _write_tools_json(
        tmp_path,
        {
            "tools": [
                {"name": "ninja", "versions": [{"name": "1.12.1"}]},
                {"name": "cmake", "versions": [{"name": "3.24.0"}]},
            ]
        },
    )
    with patch("esphome.espidf.framework.platform.machine", return_value="aarch64"):
        _patch_tools_json_for_linux_arm64(tmp_path)

    data = json.loads(tools_json.read_text(encoding="utf-8"))
    ninja = next(t for t in data["tools"] if t["name"] == "ninja")
    assert "linux-arm64" in ninja["versions"][0]
    assert ninja["versions"][0]["linux-arm64"]["size"] == 121787


def test_patch_tools_json_already_patched_is_noop(tmp_path: Path) -> None:
    tools_json = _write_tools_json(
        tmp_path,
        {
            "tools": [
                {
                    "name": "ninja",
                    "versions": [{"name": "1.12.1", "linux-arm64": {"url": "x"}}],
                }
            ]
        },
    )
    before = tools_json.read_text(encoding="utf-8")
    with patch("esphome.espidf.framework.platform.machine", return_value="aarch64"):
        _patch_tools_json_for_linux_arm64(tmp_path)
    assert tools_json.read_text(encoding="utf-8") == before


# ---------------------------------------------------------------------------
# Subprocess-backed helpers (_exec -> run_command rename) and get_framework_env
# ---------------------------------------------------------------------------


def test_get_idf_version_parses_stdout(tmp_path: Path) -> None:
    with patch(
        "esphome.espidf.framework.run_command", return_value=(True, "5.1.2\n", "")
    ):
        assert _get_idf_version(tmp_path) == "5.1.2"


def test_get_idf_version_raises_on_failure(tmp_path: Path) -> None:
    with (
        patch("esphome.espidf.framework.run_command", return_value=(False, "", "boom")),
        pytest.raises(RuntimeError, match="Can't get ESP-IDF version"),
    ):
        _get_idf_version(tmp_path)


def test_get_idf_tool_paths_parses_json(tmp_path: Path) -> None:
    payload = json.dumps({"paths_to_export": ["/a", "/b"], "export_vars": {"X": "1"}})
    with patch(
        "esphome.espidf.framework.run_command", return_value=(True, payload, "")
    ):
        paths, export_vars = _get_idf_tool_paths(tmp_path)
    assert paths == ["/a", "/b"]
    assert export_vars == {"X": "1"}


def test_get_idf_tool_paths_raises_on_bad_json(tmp_path: Path) -> None:
    with (
        patch(
            "esphome.espidf.framework.run_command", return_value=(True, "not json", "")
        ),
        pytest.raises(RuntimeError, match="Can't extract ESP-IDF tool paths"),
    ):
        _get_idf_tool_paths(tmp_path)


def test_get_idf_tool_paths_raises_on_failure(tmp_path: Path) -> None:
    with (
        patch("esphome.espidf.framework.run_command", return_value=(False, "", "err")),
        pytest.raises(RuntimeError, match="Can't get ESP-IDF tool paths"),
    ):
        _get_idf_tool_paths(tmp_path)


def test_get_python_version_parses_stdout(tmp_path: Path) -> None:
    with patch(
        "esphome.espidf.framework.run_command", return_value=(True, "3.11.0\n", "")
    ):
        assert _get_python_version(tmp_path / "python") == "3.11.0"


def test_get_python_version_returns_falsy_on_failure(tmp_path: Path) -> None:
    with patch("esphome.espidf.framework.run_command", return_value=(False, "", "")):
        # non-throwing failure returns the (empty) stdout as-is
        assert not _get_python_version(tmp_path / "python")


def test_get_python_version_raises_when_requested(tmp_path: Path) -> None:
    with (
        patch("esphome.espidf.framework.run_command", return_value=(False, "", "")),
        pytest.raises(RuntimeError, match="Can't get Python version"),
    ):
        _get_python_version(tmp_path / "python", throw_exception=True)


def test_write_stamp_writes_json(tmp_path: Path) -> None:
    stamp = tmp_path / "stamp.json"
    _write_stamp(stamp, {"a": "1", "b": "2"})
    assert json.loads(stamp.read_text(encoding="utf-8")) == {"a": "1", "b": "2"}


def test_get_framework_env_with_python_env(tmp_path: Path) -> None:
    with (
        patch(
            "esphome.espidf.framework._get_idf_tools_path",
            return_value=tmp_path / "tools",
        ),
        patch("esphome.espidf.framework._get_idf_version", return_value="5.1.2"),
        patch(
            "esphome.espidf.framework._get_idf_tool_paths",
            return_value=(["/tool/bin"], {"IDF_X": "1"}),
        ),
    ):
        env = get_framework_env(
            tmp_path / "fw", tmp_path / "penv", {"PATH": "/usr/bin"}
        )

    assert env["IDF_PATH"] == str(tmp_path / "fw")
    assert env["ESP_IDF_VERSION"] == "5.1.2"
    assert env["IDF_X"] == "1"
    assert env["IDF_PYTHON_ENV_PATH"] == str(tmp_path / "penv")
    assert "/tool/bin" in env["PATH"]


def test_get_framework_env_without_python_env_uses_os_path(tmp_path: Path) -> None:
    with (
        patch(
            "esphome.espidf.framework._get_idf_tools_path",
            return_value=tmp_path / "tools",
        ),
        patch("esphome.espidf.framework._get_idf_version", return_value="5.1.2"),
        patch("esphome.espidf.framework._get_idf_tool_paths", return_value=([], {})),
    ):
        env = get_framework_env(tmp_path / "fw")

    assert "IDF_PYTHON_ENV_PATH" not in env
    assert env["PATH"]  # taken from os.environ


# ---------------------------------------------------------------------------
# _check_stamp / _write_idf_version_txt / _get_idf_tools_path
# ---------------------------------------------------------------------------


def test_check_stamp_matches(tmp_path: Path) -> None:
    f = tmp_path / "s.json"
    f.write_text(json.dumps({"a": "1"}), encoding="utf-8")
    assert _check_stamp(f, {"a": "1"}) is True


def test_check_stamp_mismatch(tmp_path: Path) -> None:
    f = tmp_path / "s.json"
    f.write_text(json.dumps({"a": "1"}), encoding="utf-8")
    assert _check_stamp(f, {"a": "2"}) is False


def test_check_stamp_missing_file(tmp_path: Path) -> None:
    assert _check_stamp(tmp_path / "nope.json", {"a": "1"}) is False


def test_check_stamp_corrupt_file(tmp_path: Path) -> None:
    f = tmp_path / "s.json"
    f.write_text("{ not json", encoding="utf-8")
    assert _check_stamp(f, {"a": "1"}) is False


def test_write_idf_version_txt_writes_when_missing(tmp_path: Path) -> None:
    _write_idf_version_txt(tmp_path, "5.1.2")
    assert (tmp_path / "version.txt").read_text(encoding="utf-8") == "v5.1.2\n"


def test_write_idf_version_txt_skips_when_present(tmp_path: Path) -> None:
    (tmp_path / "version.txt").write_text("existing\n", encoding="utf-8")
    _write_idf_version_txt(tmp_path, "5.1.2")
    assert (tmp_path / "version.txt").read_text(encoding="utf-8") == "existing\n"


def test_get_idf_tools_path_env_override(tmp_path: Path) -> None:
    override = str(tmp_path / "custom-idf")
    with patch.dict("os.environ", {"ESPHOME_ESP_IDF_PREFIX": override}):
        assert _get_idf_tools_path() == Path(override)


def test_write_idf_version_txt_warns_on_write_error(tmp_path: Path) -> None:
    with patch("pathlib.Path.write_text", side_effect=OSError("denied")):
        # write failure is caught and warned, not raised
        _write_idf_version_txt(tmp_path, "5.1.2")