esphome/tests/unit_tests/test_yaml_edit.py

"""Tests for rewriting single lines of a yaml file."""

from __future__ import annotations

from pathlib import Path
import sys

import pytest

from esphome import yaml_util
from esphome.config import do_substitution_pass
from esphome.const import CONF_ESPHOME, CONF_NAME
from esphome.core import CORE, EsphomeError
from esphome.yaml_edit import (
    LineEdit,
    field_line_re,
    line_at,
    read_text,
    rewrite,
    rewritten_text,
    source_of,
    write_keeping_mode,
)

YAML = """esphome:
  name: kitchen  # the device

wifi:
  ssid: kitchen
"""


def _setup(tmp_path: Path, yaml_text: str) -> Path:
    """Write the yaml, point CORE at it and load it the way read_config does,
    so every node carries its source range."""
    CORE.reset()
    CORE.config_path = tmp_path / "test.yaml"
    # Bytes, so Windows does not turn the newlines into CRLF on the way in
    CORE.config_path.write_bytes(yaml_text.encode())
    CORE.raw_config = do_substitution_pass(yaml_util.load_yaml(CORE.config_path), None)
    return CORE.config_path


def _name_edit(new_name: str) -> LineEdit:
    doc, line_no = source_of(CORE.raw_config[CONF_ESPHOME], CONF_NAME)
    text = line_at(doc, line_no)
    match = field_line_re(CONF_NAME, "kitchen").match(text)
    return LineEdit(doc, line_no, text, rewrite(match, new_name))


def test_rewrite_keeps_the_rest_of_the_line(tmp_path: Path) -> None:
    """The value changes; indentation, quotes and the comment stay."""
    path = _setup(tmp_path, YAML.replace("name: kitchen", "name: 'kitchen'"))
    edit = _name_edit("garage")
    assert (edit.line, edit.new_line) == (1, "  name: 'garage'  # the device")
    assert rewritten_text(read_text(path), [edit]) == YAML.replace(
        "name: kitchen", "name: 'garage'"
    )


def test_rewrite_can_force_quotes() -> None:
    match = field_line_re(CONF_NAME, "kitchen").match("  name: kitchen  # x")
    assert rewrite(match, "garage", quote='"') == '  name: "garage"  # x'


def test_only_the_located_line_changes(tmp_path: Path) -> None:
    """A lookalike `name:` under another block has its own range."""
    yaml_text = YAML + "sensor:\n  - platform: template\n    name: kitchen\n"
    path = _setup(tmp_path, yaml_text)
    text = rewritten_text(read_text(path), [_name_edit("garage")])
    assert text.endswith("    name: kitchen\n")
    assert "  name: garage  # the device" in text


def test_line_endings_are_kept(tmp_path: Path) -> None:
    path = _setup(tmp_path, YAML.replace("\n", "\r\n"))
    assert rewritten_text(read_text(path), [_name_edit("garage")]) == YAML.replace(
        "\n", "\r\n"
    ).replace("name: kitchen", "name: garage")


def test_stale_line_is_refused(tmp_path: Path) -> None:
    path = _setup(tmp_path, YAML)
    edit = _name_edit("garage")
    with pytest.raises(EsphomeError, match="changed since it was read"):
        rewritten_text(YAML.replace("kitchen  #", "pantry  #"), [edit])
    with pytest.raises(EsphomeError, match="changed since it was read"):
        rewritten_text("esphome:\n", [edit])
    with pytest.raises(EsphomeError, match="changed since it was read"):
        line_at(path, 5)


def test_a_comment_needs_whitespace_and_a_scalar_is_not_empty() -> None:
    """`abc#def` is one value to the loader, and a bare `key:` heads a block."""
    assert field_line_re("key", "abc").match("key: abc#def") is None
    assert field_line_re("key").match("key:") is None
    assert field_line_re("key").match("key: abc  # c")["trail"] == "  # c"
    assert field_line_re("key", "abc#def").match("key: abc#def") is not None


def test_source_of_is_none_for_a_value_validation_added(tmp_path: Path) -> None:
    """Only a key read from a file carries a range, and only a mapping read
    from a file can place its keys; a mapping built in code cannot."""
    assert source_of({"name": "kitchen"}, "name") is None
    _setup(tmp_path, YAML)
    loaded_key = next(iter(CORE.raw_config[CONF_ESPHOME]))
    assert source_of({loaded_key: "kitchen"}, CONF_NAME) is None


def test_source_of_is_none_for_a_merged_key(tmp_path: Path) -> None:
    """A key a merge brought in points at the anchor, which other mappings
    may merge as well; it is not this mapping's own line."""
    _setup(
        tmp_path,
        "named: &named\n  name: kitchen\n\nesphome:\n  <<: *named\n  friendly_name: x\n",
    )
    assert source_of(CORE.raw_config[CONF_ESPHOME], CONF_NAME) is None
    assert source_of(CORE.raw_config[CONF_ESPHOME], "friendly_name") == (
        tmp_path / "test.yaml",
        5,
    )


def test_source_of_names_the_file_the_loader_read(tmp_path: Path) -> None:
    """An include has its own document; a symlink is reported as given."""
    (tmp_path / "base.yaml").write_bytes(b"name: kitchen\n")
    _setup(tmp_path, "esphome: !include base.yaml\n")
    assert source_of(CORE.raw_config[CONF_ESPHOME], CONF_NAME) == (
        tmp_path / "base.yaml",
        0,
    )
    target = tmp_path / "shared" / "test.yaml"
    target.parent.mkdir()
    target.write_bytes(YAML.encode())
    CORE.config_path.unlink()
    CORE.config_path.symlink_to(target)
    CORE.raw_config = yaml_util.load_yaml(CORE.config_path)
    assert _name_edit("garage").path == CORE.config_path


@pytest.mark.skipif(sys.platform == "win32", reason="posix file modes")
def test_write_keeps_the_mode_of_the_file_or_another(tmp_path: Path) -> None:
    path = _setup(tmp_path, YAML)
    path.chmod(0o600)
    write_keeping_mode(path, YAML)
    assert path.stat().st_mode & 0o777 == 0o600
    other = tmp_path / "other.yaml"
    write_keeping_mode(other, YAML, like=path)
    assert other.stat().st_mode & 0o777 == 0o600


def test_write_failures_say_which_step_and_why(tmp_path: Path) -> None:
    """A missing mode source, a write that fails, and a mode that cannot be
    put back after the write are three different situations."""
    from unittest.mock import patch

    path = _setup(tmp_path, YAML)
    with pytest.raises(EsphomeError, match="Could not read the mode of .*gone.yaml"):
        write_keeping_mode(path, YAML, like=tmp_path / "gone.yaml")
    with (
        patch("pathlib.Path.chmod", side_effect=OSError("denied")),
        pytest.raises(
            EsphomeError, match="was written but could not get its mode back: denied"
        ),
    ):
        write_keeping_mode(path, YAML)

    def refuse(*_args: object, **_kwargs: object) -> None:
        raise EsphomeError(f"Could not write file at {path}") from OSError("disk full")

    with (
        patch("esphome.yaml_edit.write_file", side_effect=refuse),
        pytest.raises(EsphomeError, match="Could not write file at .*: disk full"),
    ):
        write_keeping_mode(path, YAML)


def test_read_text_reports_a_file_it_cannot_decode(tmp_path: Path) -> None:
    path = tmp_path / "latin1.yaml"
    path.write_bytes(b"caf\xe9: 1\n")
    with pytest.raises(EsphomeError, match="Error reading file"):
        read_text(path)