Improve tests of YAML config annotations (#182457)

tests/components/blueprint/test_models.py

@@ -1,13 +1,32 @@
 """Test blueprint models."""
 
+from collections.abc import Callable
 import logging
+from typing import Any
 from unittest.mock import AsyncMock, patch
 
 import pytest
 
 from homeassistant.components.blueprint import BLUEPRINT_SCHEMA, errors, models
+from homeassistant.config import _get_annotation
 from homeassistant.core import HomeAssistant
 from homeassistant.util.yaml import Input
+from homeassistant.util.yaml.objects import NodeDictClass, NodeListClass, NodeStrClass
+
+from tests.common import get_test_config_dir
+
+# A real blueprint in the test config dir, loaded from disk by DomainBlueprints.
+BLUEPRINT_PATH = "test_event_sensor.yaml"
+BLUEPRINT_FILE = get_test_config_dir("blueprints", "template", BLUEPRINT_PATH)
+
+SUBSTITUTE_XFAIL = pytest.mark.xfail(
+    strict=True,
+    reason=(
+        "annotatedyaml's substitute (input.py:51,54) rebuilds containers with "
+        "comprehensions, so the node class and the __config_file__/__line__ slots "
+        "are both dropped"
+    ),
+)
 
 
 @pytest.fixture
@@ -62,6 +81,41 @@ def blueprint_2(request: pytest.FixtureRequest) -> models.Blueprint:
     return models.Blueprint(blueprint, schema=BLUEPRINT_SCHEMA)
 
 
+@pytest.fixture
+async def yaml_blueprint(hass: HomeAssistant) -> models.Blueprint:
+    """Blueprint loaded from a real YAML file, so its nodes carry source annotations."""
+    domain_bps = models.DomainBlueprints(
+        hass,
+        "template",
+        logging.getLogger(__name__),
+        None,
+        AsyncMock(),
+        BLUEPRINT_SCHEMA,
+    )
+    return await domain_bps.async_get_blueprint(BLUEPRINT_PATH)
+
+
+@pytest.fixture
+def yaml_blueprint_inputs(
+    yaml_blueprint: models.Blueprint,
+) -> models.BlueprintInputs:
+    """Validated inputs for the YAML blueprint fixture."""
+    inputs = models.BlueprintInputs(
+        yaml_blueprint,
+        {
+            "use_blueprint": {
+                "path": BLUEPRINT_PATH,
+                "input": {
+                    "event_type": "my_event",
+                    "event_data": {"hello": "world"},
+                },
+            }
+        },
+    )
+    inputs.validate()
+    return inputs
+
+
 @pytest.fixture
 def domain_bps(hass: HomeAssistant) -> models.DomainBlueprints:
     """Domain blueprints fixture."""
@@ -227,6 +281,92 @@ def test_blueprint_inputs_override_default(blueprint_2: models.Blueprint) -> Non
     assert inputs.async_substitute() == {"example": 1, "example-default": "custom"}
 
 
+def test_yaml_blueprint_keeps_annotations_through_the_schema(
+    yaml_blueprint: models.Blueprint,
+) -> None:
+    """Test the blueprint schema hands the config through with its annotations.
+
+    BLUEPRINT_SCHEMA allows extra keys, and probatio stores the original value
+    object for those, so anything missing after async_substitute was dropped by
+    the substitution rather than by the schema.
+    """
+    data = yaml_blueprint.data
+
+    assert type(data["triggers"]) is NodeListClass
+    assert _get_annotation(data["triggers"]) == (BLUEPRINT_FILE, 18)
+    assert _get_annotation(data["triggers"][0]) == (BLUEPRINT_FILE, 18)
+    assert _get_annotation(data["sensor"]) == (BLUEPRINT_FILE, 24)
+    assert _get_annotation(data["sensor"]["attributes"]) == (BLUEPRINT_FILE, 27)
+
+
+@pytest.mark.parametrize(
+    ("select", "expected_type", "expected_annotation"),
+    [
+        pytest.param(
+            lambda config: next(key for key in config if key == "triggers"),
+            NodeStrClass,
+            (BLUEPRINT_FILE, 17),
+            id="key_at_root",
+        ),
+        pytest.param(
+            lambda config: next(
+                key for key in config["triggers"][0] if key == "trigger"
+            ),
+            NodeStrClass,
+            (BLUEPRINT_FILE, 18),
+            id="key_in_list_element",
+        ),
+        pytest.param(
+            lambda config: config["triggers"],
+            NodeListClass,
+            (BLUEPRINT_FILE, 18),
+            marks=SUBSTITUTE_XFAIL,
+            id="list_in_dict",
+        ),
+        pytest.param(
+            lambda config: config["triggers"][0],
+            NodeDictClass,
+            (BLUEPRINT_FILE, 18),
+            marks=SUBSTITUTE_XFAIL,
+            id="dict_in_list",
+        ),
+        pytest.param(
+            lambda config: config["sensor"],
+            NodeDictClass,
+            (BLUEPRINT_FILE, 24),
+            marks=SUBSTITUTE_XFAIL,
+            id="dict_in_dict",
+        ),
+        pytest.param(
+            lambda config: config["sensor"]["attributes"],
+            NodeDictClass,
+            (BLUEPRINT_FILE, 27),
+            marks=SUBSTITUTE_XFAIL,
+            id="dict_in_dict_in_dict",
+        ),
+    ],
+)
+def test_substituted_blueprint_keeps_annotations(
+    yaml_blueprint_inputs: models.BlueprintInputs,
+    select: Callable[[dict], Any],
+    expected_type: type,
+    expected_annotation: tuple[str, int],
+) -> None:
+    """Test the substituted config keeps the node classes and their locations.
+
+    The keys pass today because substitute's dict comprehension hands them
+    through unchanged, which is the only reason a blueprint-backed config still
+    reports where an error came from. The result's top level is deliberately not
+    asserted: async_substitute merges it into a fresh dict literal, so it can
+    never carry an annotation.
+    """
+    config = yaml_blueprint_inputs.async_substitute()
+    node = select(config)
+
+    assert type(node) is expected_type
+    assert _get_annotation(node) == expected_annotation
+
+
 async def test_domain_blueprints_get_blueprint_errors(
     hass: HomeAssistant, domain_bps: models.DomainBlueprints
 ) -> None:

tests/fixtures/core/config/annotations/basic/configuration.yaml

@@ -0,0 +1,9 @@
+test_domain:
+  mars: blah
+  servers:
+    - port: 8123
+      name: kitchen
+    - port: 8124
+      name: hallway
+
+included_domain: !include included.yaml

tests/fixtures/core/config/annotations/basic/included.yaml

@@ -0,0 +1,4 @@
+mars: blah
+servers:
+  - port: 9123
+    name: garage

tests/fixtures/core/config/annotations/include_dirs/configuration.yaml

@@ -0,0 +1,4 @@
+dir_list: !include_dir_list entries
+dir_merge_list: !include_dir_merge_list merge_entries
+dir_named: !include_dir_named entries
+dir_merge_named: !include_dir_merge_named entries

tests/fixtures/core/config/annotations/include_dirs/entries/one.yaml

@@ -0,0 +1,2 @@
+name: kitchen
+port: 8123

tests/fixtures/core/config/annotations/include_dirs/entries/two.yaml

@@ -0,0 +1,2 @@
+name: hallway
+port: 8124

tests/fixtures/core/config/annotations/include_dirs/merge_entries/first.yaml

@@ -0,0 +1,2 @@
+- name: garage
+  port: 8125

tests/helpers/test_config_validation.py

@@ -7,6 +7,7 @@ import enum
 from functools import partial
 import logging
 import os
+from pathlib import Path
 import re
 from socket import _GLOBAL_DEFAULT_TIMEOUT
 import threading
@@ -29,6 +30,7 @@ from homeassistant.helpers import (
 )
 from homeassistant.helpers.config_validation import TRIGGER_SCHEMA
 from homeassistant.util import dt as dt_util
+from homeassistant.util.yaml import load_yaml_dict
 
 
 def test_boolean() -> None:
@@ -1232,6 +1234,76 @@ def test_deprecated_logger_without_config_attributes(
     assert len(caplog.records) == 0
 
 
+@pytest.mark.parametrize(
+    ("validator", "option_status"),
+    [
+        pytest.param(cv.deprecated, "is deprecated", id="deprecated"),
+        pytest.param(
+            partial(cv.removed, raise_if_present=False),
+            "has been removed",
+            id="removed",
+        ),
+    ],
+)
+@pytest.mark.parametrize(
+    "preprocess",
+    [
+        pytest.param(lambda config: config, id="raw_loader_output"),
+        pytest.param(probatio.Schema(dict), id="pass_through_schema"),
+        pytest.param(
+            probatio.Schema({}, extra=probatio.ALLOW_EXTRA),
+            id="rebuilding_schema",
+            marks=pytest.mark.xfail(
+                strict=True,
+                reason=(
+                    "The mapping engine allocates a fresh container and never copies "
+                    "the __config_file__/__line__ slots, so _deprecated_or_removed "
+                    "takes its 'except AttributeError' branch and logs no location."
+                ),
+            ),
+        ),
+        pytest.param(
+            cv.PLATFORM_SCHEMA_BASE,
+            id="platform_schema_base",
+            marks=pytest.mark.xfail(
+                strict=True,
+                reason=(
+                    "Same rebuild, through the real PLATFORM_SCHEMA_BASE that "
+                    "check_config.py runs before the platform's own PLATFORM_SCHEMA."
+                ),
+            ),
+        ),
+    ],
+)
+def test_deprecated_or_removed_location_survives_schema_rebuild(
+    caplog: pytest.LogCaptureFixture,
+    tmp_path: Path,
+    validator: Callable[..., Callable[[dict], dict]],
+    option_status: str,
+    preprocess: Callable[[dict], dict],
+) -> None:
+    """Test the location prefix survives a schema the config was passed through.
+
+    The config is built with the real YAML loader so the annotation is the one
+    production gets, not a hand-set attribute.
+    """
+    # Note: Unlike find_annotation, which reads the location off the mapping key,
+    # _deprecated_or_removed reads it off the container itself, so a schema that
+    # rebuilds the container silently drops the "near <file>:<line>" prefix. That is
+    # what check_config.py does when it hands the already validated p_validated to
+    # the platform's own PLATFORM_SCHEMA, where the cv.deprecated calls of broadlink,
+    # canary, mvglive and integration live.
+
+    config_file = tmp_path / "configuration.yaml"
+    config_file.write_text("platform: test_platform\nmars: blah\n", encoding="utf8")
+    config = load_yaml_dict(str(config_file))
+
+    validator("mars", default=False)(preprocess(config))
+
+    assert len(caplog.records) == 1
+    assert f"The 'mars' option near {config_file}:1 {option_status}" in caplog.text
+
+
 def test_key_dependency() -> None:
     """Test key_dependency validator."""
     schema = probatio.Schema(cv.key_dependency("beer", "soda"))

tests/test_config_annotations.py

@@ -0,0 +1,282 @@
+"""Test that YAML source annotations survive Home Assistant's config pipeline.
+
+The loader (annotatedyaml), the validation engine (probatio) and the readers in
+homeassistant.config are three separate packages, and what breaks in production
+is their composition: a location the loader recorded is read back after a schema
+has run. No single package's test suite covers that, so it lives here rather
+than next to any one helper.
+"""
+
+from collections.abc import Callable, Generator
+import os
+from typing import Any
+
+from probatio import (
+    Coerce,
+    CompilePolicy,
+    Schema,
+    get_compile_policy,
+    set_compile_policy,
+)
+import pytest
+
+from homeassistant import config as config_util
+from homeassistant.config import _get_annotation, find_annotation
+from homeassistant.core import HomeAssistant
+from homeassistant.util.yaml.objects import NodeDictClass
+
+from .common import get_fixture_path
+
+CONFIG_DIR = str(get_fixture_path("core/config/annotations/basic"))
+CONFIGURATION_YAML = os.path.join(CONFIG_DIR, "configuration.yaml")
+INCLUDED_YAML = os.path.join(CONFIG_DIR, "included.yaml")
+
+CONTAINER_XFAIL = pytest.mark.xfail(
+    strict=True,
+    reason=(
+        "probatio's mapping and sequence engines rebuild the container as a fresh "
+        "instance of the input's class, so the __config_file__/__line__ slots "
+        "annotatedyaml set on the original are never copied."
+    ),
+)
+
+
+@pytest.fixture(
+    params=[CompilePolicy.OFF, CompilePolicy.ON],
+    ids=["interpreted", "generated"],
+)
+def probatio_compile_policy(request: pytest.FixtureRequest) -> Generator[None]:
+    """Run the test against the interpreted and the generated validator.
+
+    The generated validator bails back to the interpreted engine for anything
+    that is not exactly a dict or a list, so a node class takes the same
+    interpreted rebuild under either policy today. Running both is what guards
+    that bail-out: if generated code ever handled dict subclasses inline without
+    copying the slots, only the generated case would move.
+
+    OFF and ON are pinned rather than the AUTO default because AUTO switches from
+    one to the other after a call count probatio keeps internal and does not
+    export.
+    """
+    original = get_compile_policy()
+    set_compile_policy(request.param)
+    yield
+    set_compile_policy(original)
+
+
+@pytest.fixture
+async def annotated_config(hass: HomeAssistant) -> dict:
+    """Load the annotations fixture directory the way Home Assistant loads it."""
+    hass.config.config_dir = CONFIG_DIR
+    return await config_util.async_hass_config_yaml(hass)
+
+
+def _key(mapping: dict, name: str) -> Any:
+    """Return the key object of mapping equal to name."""
+    return next(key for key in mapping if key == name)
+
+
+def _schema() -> Schema:
+    """Return a schema for the fixture config, built fresh so the policy applies."""
+    domain = {"mars": str, "servers": [{"port": int, "name": str}]}
+    return Schema({"test_domain": domain, "included_domain": domain})
+
+
+async def test_loader_annotates_every_container(annotated_config: dict) -> None:
+    """Test the loader records a distinct file and line at each nesting depth.
+
+    Precondition for the rest of the module: if this fails the fixture config is
+    broken rather than the carry.
+    """
+    assert _get_annotation(annotated_config) == (CONFIGURATION_YAML, 1)
+    assert _get_annotation(annotated_config["test_domain"]) == (CONFIGURATION_YAML, 2)
+    assert _get_annotation(annotated_config["test_domain"]["servers"]) == (
+        CONFIGURATION_YAML,
+        4,
+    )
+    assert _get_annotation(annotated_config["test_domain"]["servers"][1]) == (
+        CONFIGURATION_YAML,
+        6,
+    )
+    assert _get_annotation(annotated_config["included_domain"]["servers"]) == (
+        INCLUDED_YAML,
+        3,
+    )
+
+
+@pytest.mark.usefixtures("probatio_compile_policy")
+async def test_validation_keeps_the_node_class(annotated_config: dict) -> None:
+    """Test a rebuilt container is still a node class; only the location is dropped.
+
+    Pinning the class separately keeps the two halves of the bug apart: an
+    annotation assertion that fails because the class changed would be a
+    different defect than the one the xfails below describe.
+    """
+    validated = _schema()(annotated_config)
+
+    assert type(validated) is NodeDictClass
+    assert type(validated["test_domain"]) is NodeDictClass
+    assert type(validated["test_domain"]["servers"][1]) is NodeDictClass
+    assert type(validated["included_domain"]) is NodeDictClass
+
+
+@pytest.mark.parametrize(
+    ("select", "expected"),
+    [
+        pytest.param(
+            lambda config: _key(config, "test_domain"),
+            (CONFIGURATION_YAML, 1),
+            id="key_at_root",
+        ),
+        pytest.param(
+            lambda config: _key(config["test_domain"], "servers"),
+            (CONFIGURATION_YAML, 3),
+            id="key_in_nested_dict",
+        ),
+        pytest.param(
+            lambda config: _key(config["test_domain"]["servers"][1], "name"),
+            (CONFIGURATION_YAML, 7),
+            id="key_in_list_element",
+        ),
+        pytest.param(
+            lambda config: _key(config["included_domain"], "servers"),
+            (INCLUDED_YAML, 2),
+            id="key_in_included_file",
+        ),
+        pytest.param(
+            lambda config: config["test_domain"]["mars"],
+            (CONFIGURATION_YAML, 2),
+            id="scalar_value",
+        ),
+        pytest.param(
+            lambda config: config,
+            (CONFIGURATION_YAML, 1),
+            marks=CONTAINER_XFAIL,
+            id="root_dict",
+        ),
+        pytest.param(
+            lambda config: config["test_domain"],
+            (CONFIGURATION_YAML, 2),
+            marks=CONTAINER_XFAIL,
+            id="dict_in_dict",
+        ),
+        pytest.param(
+            lambda config: config["test_domain"]["servers"],
+            (CONFIGURATION_YAML, 4),
+            marks=CONTAINER_XFAIL,
+            id="list_in_dict",
+        ),
+        pytest.param(
+            lambda config: config["test_domain"]["servers"][1],
+            (CONFIGURATION_YAML, 6),
+            marks=CONTAINER_XFAIL,
+            id="dict_in_list",
+        ),
+        pytest.param(
+            lambda config: config["included_domain"]["servers"],
+            (INCLUDED_YAML, 3),
+            marks=CONTAINER_XFAIL,
+            id="list_in_included_file",
+        ),
+    ],
+)
+@pytest.mark.usefixtures("probatio_compile_policy")
+async def test_annotation_survives_validation(
+    annotated_config: dict,
+    select: Callable[[dict], Any],
+    expected: tuple[str, int],
+) -> None:
+    """Test a schema rebuild keeps the file and line the loader recorded.
+
+    The key and scalar cases pass today because probatio stores the original key
+    and value objects in the rebuilt mapping, which is why config error messages
+    still carry locations at all; the container cases are the bug. The two
+    included-file cases pin the file as well as the line, so an annotation that
+    survives pointing at the including file would still fail.
+    """
+    validated = _schema()(annotated_config)
+
+    assert _get_annotation(select(validated)) == expected
+
+
+async def test_type_check_schema_returns_the_same_object(
+    annotated_config: dict,
+) -> None:
+    """Test Schema(dict) checks the type and rebuilds nothing, so nothing is lost.
+
+    This separates "the rebuild dropped the location" from "going through a
+    schema dropped the location".
+    """
+    validated = Schema(dict)(annotated_config)
+
+    assert validated is annotated_config
+    assert _get_annotation(validated) == (CONFIGURATION_YAML, 1)
+
+
+async def test_coerce_loses_the_annotation(annotated_config: dict) -> None:
+    """Test Coerce builds a plain dict, which has no slot to hold a location.
+
+    Asserted as it behaves rather than xfailed: this loss is inherent to the
+    target type, so an upstream change that started preserving it is a surprise
+    worth being told about.
+    """
+    validated = Schema(Coerce(dict))(annotated_config)
+
+    assert type(validated) is dict
+    assert _get_annotation(validated) is None
+
+
+async def test_rebuilding_validator_loses_the_annotation(
+    annotated_config: dict,
+) -> None:
+    """Test a validator rebuilding via type(value)(...) keeps the class, not the location.
+
+    Asserted as it behaves rather than xfailed: preserving the class is not
+    preserving the annotation, and a validator that rebuilds has to carry the
+    location itself.
+    """
+
+    def rebuild(value: NodeDictClass) -> NodeDictClass:
+        return type(value)((key, item) for key, item in value.items())
+
+    validated = Schema(rebuild)(annotated_config)
+
+    assert type(validated) is NodeDictClass
+    assert _get_annotation(validated) is None
+
+
+@pytest.mark.parametrize(
+    ("path", "expected"),
+    [
+        pytest.param(["test_domain", "mars"], (CONFIGURATION_YAML, 2), id="own_key"),
+        pytest.param(
+            ["test_domain", "servers"], (CONFIGURATION_YAML, 3), id="parent_key"
+        ),
+        pytest.param(
+            ["included_domain", "servers"],
+            (INCLUDED_YAML, 2),
+            id="key_in_included_file",
+        ),
+        pytest.param(
+            [], (CONFIGURATION_YAML, 1), marks=CONTAINER_XFAIL, id="root_has_no_key"
+        ),
+        pytest.param(
+            ["test_domain", "servers", 0],
+            (CONFIGURATION_YAML, 4),
+            marks=CONTAINER_XFAIL,
+            id="list_element_has_no_key",
+        ),
+    ],
+)
+@pytest.mark.usefixtures("probatio_compile_policy")
+async def test_find_annotation_after_validation(
+    annotated_config: dict, path: list[str | int], expected: tuple[str, int]
+) -> None:
+    """Test find_annotation reads the key first, so it only fails where no key is reachable.
+
+    The list element case reports the line of the "servers:" key today instead of
+    the line of the element itself.
+    """
+    validated = _schema()(annotated_config)
+
+    assert find_annotation(validated, path) == expected

tests/util/yaml/test_init.py

@@ -12,13 +12,21 @@ import probatio
 import pytest
 import yaml as pyyaml
 
-from homeassistant.config import YAML_CONFIG_FILE, load_yaml_config_file
+from homeassistant.config import (
+    YAML_CONFIG_FILE,
+    _get_annotation,
+    load_yaml_config_file,
+)
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.util import yaml as yaml_util
 from homeassistant.util.yaml import loader as yaml_loader
+from homeassistant.util.yaml.objects import NodeDictClass, NodeListClass
 
-from tests.common import extract_stack_to_frame
+from tests.common import extract_stack_to_frame, get_fixture_path
+
+INCLUDE_DIRS_FIXTURE = str(get_fixture_path("core/config/annotations/include_dirs"))
+INCLUDE_DIRS_CONFIG = os.path.join(INCLUDE_DIRS_FIXTURE, "configuration.yaml")
 
 
 @pytest.fixture(params=["enable_c_loader", "disable_c_loader"])
@@ -374,6 +382,55 @@ def test_include_dir_merge_named_recursive(mock_walk: Mock) -> None:
         }
 
 
+@pytest.mark.parametrize(
+    ("key", "expected_type", "expected_line"),
+    [
+        pytest.param(
+            "dir_list",
+            NodeListClass,
+            1,
+            marks=pytest.mark.xfail(
+                strict=True,
+                reason=(
+                    "annotatedyaml _include_dir_list_yaml returns a bare list "
+                    "comprehension and never calls _add_reference, unlike the "
+                    "three sibling include_dir tags"
+                ),
+            ),
+            id="include_dir_list",
+        ),
+        pytest.param("dir_merge_list", NodeListClass, 2, id="include_dir_merge_list"),
+        pytest.param("dir_named", NodeDictClass, 3, id="include_dir_named"),
+        pytest.param("dir_merge_named", NodeDictClass, 4, id="include_dir_merge_named"),
+    ],
+)
+@pytest.mark.usefixtures("try_both_loaders")
+def test_include_dir_annotates_the_container(
+    key: str, expected_type: type, expected_line: int
+) -> None:
+    """Test each include_dir tag annotates the container it returns."""
+    doc = yaml_loader.load_yaml_dict(INCLUDE_DIRS_CONFIG)
+
+    assert type(doc[key]) is expected_type
+    assert _get_annotation(doc[key]) == (INCLUDE_DIRS_CONFIG, expected_line)
+
+
+@pytest.mark.usefixtures("try_both_loaders")
+def test_include_dir_list_annotates_its_elements() -> None:
+    """Test each element of an include_dir_list is annotated with its own file.
+
+    The elements are what masks the unannotated list: find_annotation recurses
+    into them, so an error reported at or below an element still gets a location
+    even though the list itself carries none.
+    """
+    doc = yaml_loader.load_yaml_dict(INCLUDE_DIRS_CONFIG)
+
+    assert [_get_annotation(element) for element in doc["dir_list"]] == [
+        (os.path.join(INCLUDE_DIRS_FIXTURE, "entries", "one.yaml"), 1),
+        (os.path.join(INCLUDE_DIRS_FIXTURE, "entries", "two.yaml"), 1),
+    ]
+
+
 @patch("annotatedyaml.loader.open", create=True)
 @pytest.mark.usefixtures("try_both_loaders")
 def test_load_yaml_encoding_error(mock_open: Mock) -> None: