Merge remote-tracking branch 'origin/dev' into store-yaml-core

esphome/config_validation.py

@@ -570,8 +570,7 @@ class SensitiveValidator:
     def __call__(self, value: typing.Any) -> typing.Any:
         validated = self.inner(value)
         # Already-tagged values are returned as-is so nested cv.sensitive is
-        # idempotent; bool is an int subclass and must stay bool. Other types
-        # (e.g. lambdas) pass through unmasked.
+        # idempotent; bool is an int subclass and must stay bool.
         if isinstance(validated, (Sensitive, bool)):
             return validated
         if isinstance(validated, str):
@@ -580,6 +579,9 @@ class SensitiveValidator:
             return (SensitiveHexInt if isinstance(validated, HexInt) else SensitiveInt)(
                 validated
             )
+        if isinstance(validated, Lambda):
+            # Marked, not masked: the dumper keeps the lambda representer.
+            return add_class_to_obj(validated, Sensitive)
         return validated
 
     def __repr__(self) -> str:

esphome/git.py

@@ -145,13 +145,17 @@ class GitRepositoryError(GitException):
     """Exception raised when a git repository is in an invalid state."""
 
 
+# The userinfo part of a URL, which can carry credentials.
+URL_USERINFO_RE = re.compile(r"://([^/@\s]+)@")
+
+
 def _redact_url_credentials(text: str) -> str:
     """Mask userinfo in any URLs embedded in ``text``.
 
     Users can put credentials directly in a git URL, and log output is
     routinely pasted into public issues.
     """
-    return re.sub(r"://[^/@\s]+@", "://***@", text)
+    return URL_USERINFO_RE.sub("://***@", text)
 
 
 def _is_transient_git_error(stderr: str) -> bool:

esphome/yaml_util.py

@@ -54,7 +54,7 @@ _LOADED_SECRET_NAMES: set[str] = set()
 # Set while a secret_values_registered context is active; the dumper records
 # every emitted `!secret` name into it.
 _EMITTED_SECRET_NAMES: set[str] | None = None
-# Called with every file the YAML loader opens; ``track_yaml_loads`` builds on it.
+# Called with every file the YAML loader opens.
 # Used by device-builder (esphome/device-builder), whose listener filters by
 # thread because the hook is process-global; keep the name and the per-file
 # ``Path`` signature. Not otherwise a supported API: use at your own risk.
@@ -118,25 +118,6 @@ class SensitiveHexInt(core.HexInt, SensitiveInt):
     """Keep hexadecimal formatting and code generation for sensitive integers."""
 
 
-@contextmanager
-def track_yaml_loads() -> Generator[list[Path]]:
-    """Context manager that records every file loaded by the YAML loader.
-
-    Yields a list that is populated with resolved Path objects for every
-    file loaded through ``_load_yaml_internal`` while the context is active.
-    """
-    loaded: list[Path] = []
-
-    def _on_load(fname: Path) -> None:
-        loaded.append(Path(fname).resolve())
-
-    load_listeners.append(_on_load)
-    try:
-        yield loaded
-    finally:
-        load_listeners.remove(_on_load)
-
-
 class ESPHomeDataBase:
     @property
     def esp_range(self):
@@ -517,7 +498,7 @@ def force_load_include_files(
     Nested ``!include`` returns a deferred ``IncludeFile`` that is only resolved
     later (substitution / packages pass). Callers that need every referenced
     file to actually load — bundle discovery, on-device YAML recovery — invoke
-    this while a :func:`track_yaml_loads` listener is active so the underlying
+    this while a ``load_listeners`` callback is active so the underlying
     loader fires and records every reachable file.
 
     ``IncludeFile`` instances whose path contains unresolved substitution
@@ -539,6 +520,17 @@ def force_load_include_files(
     return result
 
 
+def include_enabled(include: IncludeFile) -> bool:
+    """False only for a statically false condition; an unresolved or malformed one
+    still loads."""
+    from voluptuous import Invalid
+
+    try:
+        return include.should_load()
+    except Invalid:
+        return True
+
+
 def _force_load_include_files(
     obj: Any,
     *,
@@ -568,6 +560,8 @@ def _force_load_include_files(
         if id(obj) in _seen:
             return
         _seen.add(id(obj))
+        if not include_enabled(obj):
+            return
         if obj.has_unresolved_file():
             _load_include_candidates(
                 obj,
@@ -638,13 +632,27 @@ class DiscoveredYamlFiles:
     *un-resolved* filename matched :data:`esphome.const.SECRETS_FILES` (so
     a ``secrets.yaml`` symlinked to a differently-named target is still
     flagged as secrets). ``files`` is incomplete when ``unresolved`` (templated
-    include paths) or ``load_errors`` is non-empty.
+    include paths) or ``load_errors`` is non-empty. ``aliases`` maps a resolved
+    path to every :func:`named_path` the YAML reached it by; ``links`` maps each
+    directory symlink a path went through to its target. Neither covers files a
+    templated include matched, nor links on a path through ``..``.
     """
 
     files: list[Path] = field(default_factory=list)
     secrets: set[Path] = field(default_factory=set)
     unresolved: list[str] = field(default_factory=list)
     load_errors: list[str] = field(default_factory=list)
+    aliases: dict[Path, list[Path]] = field(default_factory=dict)
+    links: dict[Path, Path] = field(default_factory=dict)
+
+
+def named_path(path: Path) -> Path:
+    """``path`` with its directory resolved but its own name kept: a symlinked file
+    keeps its name, and ``..`` after a symlinked directory goes where the OS does."""
+    path = Path(path)
+    if os.name == "nt":  # Windows collapses `..` before it follows a directory link
+        path = Path(os.path.normpath(path))
+    return path.parent.resolve() / path.name
 
 
 def discover_user_yaml_files(config_path: Path) -> DiscoveredYamlFiles:
@@ -663,35 +671,56 @@ def discover_user_yaml_files(config_path: Path) -> DiscoveredYamlFiles:
     """
     from esphome.const import SECRETS_FILES
 
+    root = Path(config_path).absolute().parent
     secrets: set[Path] = set()
+    # Insertion ordered, so its keys are the loaded files in first-seen order.
+    aliases: dict[Path, list[Path]] = {}
+    links: dict[Path, Path] = {}
 
-    def _capture_secret(fname: Path) -> None:
-        if Path(fname).name in SECRETS_FILES:
-            secrets.add(Path(fname).resolve())
+    def _on_load(fname: Path) -> None:
+        fname = Path(fname).absolute()
+        resolved = fname.resolve()
+        if fname.name in SECRETS_FILES:
+            secrets.add(resolved)
+        names = aliases.setdefault(resolved, [])
+        if (alias := named_path(fname)) not in names:
+            names.append(alias)
+        # The outermost directory symlink below the config directory; `..` after
+        # one is the OS's call.
+        if ".." not in fname.parts:
+            for directory in reversed((fname.parent, *fname.parent.parents)):
+                if (
+                    directory.is_relative_to(root)
+                    and directory != root
+                    and directory.is_symlink()
+                ):
+                    links[directory] = directory.resolve()
+                    break
 
-    with track_yaml_loads() as loaded:
-        load_listeners.append(_capture_secret)
+    unresolved: list[str] = []
+    load_listeners.append(_on_load)
+    try:
         try:
-            try:
-                data = load_yaml(config_path)
-            except EsphomeError as err:
-                return DiscoveredYamlFiles(
-                    list(loaded), secrets, load_errors=[f"{config_path}: {err}"]
-                )
+            data = load_yaml(config_path)
+        except EsphomeError as err:
+            load_errors = [f"{config_path}: {err}"]
+        else:
             unresolved, load_errors = force_load_include_files(
                 data, warn_on_unresolved=False, warn_on_load_error=False
             )
-        finally:
-            load_listeners.remove(_capture_secret)
+    finally:
+        load_listeners.remove(_on_load)
 
-    # Deduplicate while preserving first-seen order.
-    seen: set[Path] = set()
-    unique: list[Path] = []
-    for path in loaded:
-        if path not in seen:
-            seen.add(path)
-            unique.append(path)
-    return DiscoveredYamlFiles(unique, secrets, unresolved, load_errors)
+    # A `!secret` probe beside the including file is reported even when absent.
+    files = [path for path in aliases if path.is_file()]
+    return DiscoveredYamlFiles(
+        files,
+        secrets & set(files),
+        unresolved,
+        load_errors,
+        {path: aliases[path] for path in files},
+        links,
+    )
 
 
 def _add_data_ref(fn):
@@ -1150,8 +1179,13 @@ def parse_yaml(
     yaml_loader=None,
     *,
     track_document_range: bool = True,
+    resolve_document_include: bool = True,
 ) -> Any:
-    """Parse a YAML file."""
+    """Parse a YAML file.
+
+    resolve_document_include=False keeps a document that is one ``!include``
+    as the deferred IncludeFile, with its ``vars`` and ``condition``.
+    """
     if yaml_loader is None:
         # Nested loads (!include, !secret, !include_dir_*) inherit the
         # same tracking mode.
@@ -1181,7 +1215,7 @@ def parse_yaml(
             track_document_range=track_document_range,
         )
     # Resolved after the fallback so a failed first parse never loads the include twice
-    return _resolve_include(res)
+    return _resolve_include(res) if resolve_document_include else res
 
 
 def _load_yaml_internal_with_type(
@@ -1234,6 +1268,11 @@ def registered_secret_names() -> set[str]:
     return set(_LOADED_SECRET_NAMES)
 
 
+def registered_secret_values() -> dict[str, str]:
+    """Each registered secret value mapped to its ``!secret`` name."""
+    return dict(_SECRET_VALUES)
+
+
 @contextmanager
 def secret_values_registered(values: dict[str, str]) -> Generator[set[str]]:
     """Make :func:`dump` write these values as ``!secret <name>`` while active.
@@ -1563,6 +1602,23 @@ class ESPHomeDumper(yaml.SafeDumper):
             return self.represent_secret(value.value)
         return self.represent_scalar(tag="!lambda", value=value.value, style="|")
 
+    def represent_data(self, data):
+        # str, dict and list precede the ESPLiteralValue mixin in the MRO.
+        if isinstance(data, ESPLiteralValue):
+            # Not aliased, so the enclosing container's alias entry stays its own
+            self.alias_key = None
+            return self.represent_literal(data)
+        return super().represent_data(data)
+
+    def represent_literal(self, value: ESPLiteralValue) -> yaml.Node:
+        if isinstance(value, dict):
+            return self.represent_mapping("!literal", value)
+        if isinstance(value, list):
+            return self.represent_sequence("!literal", value)
+        if is_secret(value):
+            return self._represent_tagged_secret("!literal", value)
+        return self.represent_scalar(tag="!literal", value=str(value))
+
     def _represent_tagged_secret(self, tag: str, value: Any) -> yaml.ScalarNode:
         # Writing the secret beats leaking the value, but the tag is lost, so
         # the dumped config no longer means the same thing when loaded back.

tests/unit_tests/test_bundle.py

@@ -1393,6 +1393,7 @@ def test_force_load_skips_duplicate_include_file() -> None:
         def __init__(self) -> None:
             self.file = Path("dup.yaml")
             self.parent_file = Path("root.yaml")
+            self.should_load = lambda: True
             self.load_calls = 0
 
         def has_unresolved_file(self) -> bool:

tests/unit_tests/test_config_validation.py

@@ -70,9 +70,11 @@ from esphome.schema_extractors import SCHEMA_EXTRACT
 from esphome.util import Registry
 from esphome.yaml_util import (
     ESPHomeDataBase,
+    Sensitive,
     SensitiveHexInt,
     SensitiveInt,
     SensitiveStr,
+    dump,
     load_yaml,
     make_data_base,
     parse_yaml,
@@ -204,6 +206,17 @@ def test_sensitive__wraps_string_result_in_sensitive_str() -> None:
     assert result == "hunter2"
 
 
+def test_sensitive__marks_a_lambda_without_changing_its_value() -> None:
+    """A lambda credential is marked so store_yaml can find it; dumps still see a lambda."""
+    result = cv.sensitive(cv.templatable(cv.string))(Lambda("return id(pw).state;"))
+
+    assert isinstance(result, Lambda)
+    assert isinstance(result, Sensitive)
+    assert (
+        dump({"password": result}) == "password: !lambda |-\n  return id(pw).state;\n"
+    )
+
+
 @pytest.mark.parametrize(
     "pre_tagged", [SensitiveStr("hunter2"), SensitiveInt(5), SensitiveHexInt(0x12)]
 )

tests/unit_tests/test_yaml_util.py

@@ -1,10 +1,11 @@
 from collections.abc import Callable, Generator
 import importlib.util
 import io
+import os
 from pathlib import Path
 import shutil
 import sys
-from types import ModuleType
+from types import ModuleType, SimpleNamespace
 from unittest.mock import patch
 
 import pytest
@@ -455,70 +456,6 @@ def test_dump_sort_keys() -> None:
     assert sorted_dump.index("a_key:") < sorted_dump.index("z_key:")
 
 
-# ---------------------------------------------------------------------------
-# track_yaml_loads
-# ---------------------------------------------------------------------------
-
-
-def test_track_yaml_loads_records_files(
-    tmp_path: Path, yaml_util_module: ModuleType
-) -> None:
-    """yaml_util_module.track_yaml_loads records every file loaded inside the context."""
-    yaml_file = tmp_path / "test.yaml"
-    yaml_file.write_text("key: value\n")
-
-    with yaml_util_module.track_yaml_loads() as loaded:
-        yaml_util_module.load_yaml(yaml_file)
-
-    assert len(loaded) == 1
-    assert loaded[0] == yaml_file.resolve()
-
-
-def test_track_yaml_loads_records_includes(
-    tmp_path: Path, yaml_util_module: ModuleType
-) -> None:
-    """yaml_util_module.track_yaml_loads records nested !include files."""
-    inc = tmp_path / "included.yaml"
-    inc.write_text("included_key: 42\n")
-    main = tmp_path / "main.yaml"
-    main.write_text("child: !include included.yaml\n")
-
-    with yaml_util_module.track_yaml_loads() as loaded:
-        result = yaml_util_module.load_yaml(main)
-        # !include is deferred; resolve it to trigger the nested load
-        result["child"].load()
-
-    resolved = [p.name for p in loaded]
-    assert "main.yaml" in resolved
-    assert "included.yaml" in resolved
-
-
-def test_track_yaml_loads_empty_outside_context(
-    tmp_path: Path, yaml_util_module: ModuleType
-) -> None:
-    """Files loaded outside the context are not recorded."""
-    yaml_file = tmp_path / "test.yaml"
-    yaml_file.write_text("key: value\n")
-
-    with yaml_util_module.track_yaml_loads() as loaded:
-        pass  # load nothing inside
-
-    yaml_util_module.load_yaml(yaml_file)
-    assert loaded == []
-
-
-def test_track_yaml_loads_cleanup_on_exception(
-    tmp_path: Path, yaml_util_module: ModuleType
-) -> None:
-    """Listener is removed even if the body raises."""
-    before = len(yaml_util_module.load_listeners)
-
-    with pytest.raises(RuntimeError), yaml_util_module.track_yaml_loads():
-        raise RuntimeError("boom")
-
-    assert len(yaml_util_module.load_listeners) == before
-
-
 def test_load_listener_removing_itself_does_not_skip_the_next(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch, yaml_util_module: ModuleType
 ) -> None:
@@ -539,15 +476,20 @@ def test_load_listener_removing_itself_does_not_skip_the_next(
     assert listeners == [seen.append]
 
 
-def test_track_yaml_loads_no_duplicate_load_on_top_level_include_failure(
+def test_load_listeners_no_duplicate_load_on_top_level_include_failure(
     tmp_path: Path, yaml_util_module: ModuleType
 ) -> None:
-    """A failed top-level !include must not record any file twice in yaml_util_module.track_yaml_loads."""
+    """A failed top-level !include must not report any file to load_listeners twice."""
     main = tmp_path / "main.yaml"
     main.write_text("!include missing.yaml\n")
 
-    with yaml_util_module.track_yaml_loads() as loaded, pytest.raises(EsphomeError):
-        yaml_util_module.load_yaml(main)
+    loaded: list[Path] = []
+    yaml_util_module.load_listeners.append(loaded.append)
+    try:
+        with pytest.raises(EsphomeError):
+            yaml_util_module.load_yaml(main)
+    finally:
+        yaml_util_module.load_listeners.remove(loaded.append)
 
     assert len(loaded) == len(set(loaded)), (
         f"Files loaded more than once during a failed top-level include: {loaded}"
@@ -738,6 +680,25 @@ def test_represent_include_file_with_condition() -> None:
     )
 
 
+@pytest.mark.parametrize(
+    ("value", "secrets", "expected"),
+    [
+        ("${x}", {}, "key: !literal '${x}'\n"),
+        ({"a": "${x}"}, {}, "key: !literal\n  a: ${x}\n"),
+        (["${x}"], {}, "key: !literal\n  - ${x}\n"),
+        # A literal equal to a registered secret is written as the secret.
+        ("hunter2", {"hunter2": "pw"}, "key: !secret 'pw'\n"),
+    ],
+    ids=["scalar", "mapping", "sequence", "secret"],
+)
+def test_represent_literal(
+    value: object, secrets: dict[str, str], expected: str
+) -> None:
+    """`!literal` values keep their tag when dumped."""
+    yaml_util._SECRET_VALUES.update(secrets)
+    assert yaml_util.dump({"key": make_literal(value)}) == expected
+
+
 def test_represent_include_file_with_data_base_mixin() -> None:
     """Test that IncludeFile wrapped with ESPHomeDataBase mixin is also dumped correctly.
 
@@ -1347,6 +1308,7 @@ class _StubInclude:
         # stubs never glob real files during candidate expansion.
         self.file = file
         self.parent_file = parent_file or Path("/nonexistent/parent.yaml")
+        self.should_load = lambda: True
         self._unresolved = unresolved
         self._load_result = load_result if load_result is not None else {}
         self._raise = raise_on_load
@@ -1809,17 +1771,77 @@ def test_discover_user_yaml_files_tolerates_templated_top_level_include(
     assert (tmp_path / "wrapper.yaml").resolve() in discovered.files
 
 
-def test_track_yaml_loads_records_resolved_paths(
-    tmp_path: Path, yaml_util_module: ModuleType
-) -> None:
-    """`track_yaml_loads` is the building block - sanity-check it resolves
-    symlinks so callers can dedupe by identity."""
-    target = _write(tmp_path, "actual.yaml", "esphome:\n  name: t\n")
-    link = tmp_path / "alias.yaml"
+def test_discover_user_yaml_files_records_symlink_alias(tmp_path: Path) -> None:
+    """A file reached through a symlink is listed by its target and aliased to the link."""
+    target = _write(tmp_path, "shared/network.yaml", "ssid: x\n")
+    link = tmp_path / "wifi.yaml"
     link.symlink_to(target)
-    with yaml_util_module.track_yaml_loads() as loaded:
-        yaml_util_module.load_yaml(link)
-    assert target.resolve() in loaded
+    discovered = discover_user_yaml_files(_write_entry_including(tmp_path, "wifi.yaml"))
+    assert target.resolve() in discovered.files
+    assert discovered.aliases[target.resolve()] == [yaml_util.named_path(link)]
+
+
+def test_parse_yaml_can_keep_a_document_include(tmp_path: Path) -> None:
+    """resolve_document_include=False returns the deferred include with its vars."""
+    _write(tmp_path, "base.yaml", "esphome:\n  name: ${name}\n")
+    entry = _write(
+        tmp_path, "entry.yaml", "!include\n  file: base.yaml\n  vars:\n    name: k\n"
+    )
+    with entry.open(encoding="utf-8") as handle:
+        loaded = yaml_util.parse_yaml(entry, handle, resolve_document_include=False)
+    assert isinstance(loaded, yaml_util.IncludeFile)
+    assert loaded.file == "base.yaml"
+    assert loaded.vars == {"name": "k"}
+    assert yaml_util.load_yaml(entry) == {"esphome": {"name": "${name}"}}
+
+
+@pytest.mark.parametrize("condition", ["false", "true"])
+def test_discover_user_yaml_files_static_include_condition(
+    tmp_path: Path, condition: str
+) -> None:
+    """A missing file behind `condition: false` is not an error; behind `true` it is."""
+    entry = _write(
+        tmp_path,
+        "entry.yaml",
+        f"packages:\n  opt: !include {{file: missing.yaml, condition: {condition}}}\n",
+    )
+    discovered = discover_user_yaml_files(entry)
+    assert bool(discovered.load_errors) == (condition == "true")
+
+
+def test_discover_user_yaml_files_skips_a_disabled_templated_include(
+    tmp_path: Path,
+) -> None:
+    """A statically disabled include is skipped before its templated name is expanded."""
+    _write(tmp_path, "optional-x.yaml", "broken: [\n")
+    entry = _write(
+        tmp_path,
+        "entry.yaml",
+        "packages:\n"
+        '  opt: !include {file: "optional-${board}.yaml", condition: false}\n',
+    )
+    discovered = discover_user_yaml_files(entry)
+    assert not discovered.load_errors
+    assert all(p.name != "optional-x.yaml" for p in discovered.files)
+
+
+def test_discover_user_yaml_files_skips_missing_secrets_probe(
+    tmp_path: Path,
+) -> None:
+    """A `!secret` in a package first probes a secrets.yaml next to that file;
+    when only the entry directory has one, the missing probe is not reported."""
+    _write(tmp_path, "pkg/common.yaml", "wifi:\n  password: !secret pw\n")
+    secrets = _write(tmp_path, "cfg/secrets.yaml", "pw: value\n")
+    entry = _write(
+        tmp_path,
+        "cfg/entry.yaml",
+        "esphome:\n  name: test\npackages:\n  common: !include ../pkg/common.yaml\n",
+    )
+    with patch.object(core.CORE, "config_path", entry):
+        discovered = discover_user_yaml_files(entry)
+    assert discovered.load_errors == []
+    assert discovered.secrets == {secrets.resolve()}
+    assert all(p.is_file() for p in discovered.files)
 
 
 # ---------------------------------------------------------------------------
@@ -2728,3 +2750,71 @@ def test_dump_secret_valued_float_lambda_and_id() -> None:
     assert "!secret 'float_secret'" in dumped
     assert "!secret 'lambda_secret'" in dumped
     assert "!secret 'id_secret'" in dumped
+
+
+def test_named_path_collapses_dotdot_lexically_on_windows(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Windows resolves `..` before following a directory link, so named_path does too."""
+    (tmp_path / "real" / "sub").mkdir(parents=True)
+    # Nested, so following the link (real) and collapsing `..` (tmp_path) differ
+    (tmp_path / "link").symlink_to(tmp_path / "real" / "sub")
+    path = tmp_path / "link" / ".." / "x.yaml"
+    if os.name != "nt":  # POSIX follows the link first
+        assert yaml_util.named_path(path) == tmp_path.resolve() / "real" / "x.yaml"
+    # Only yaml_util sees Windows; pathlib keeps resolving with the real os
+    monkeypatch.setattr(yaml_util, "os", SimpleNamespace(name="nt", path=os.path))
+    assert yaml_util.named_path(path) == tmp_path.resolve() / "x.yaml"
+
+
+def test_discover_user_yaml_files_loads_an_include_with_a_templated_condition(
+    tmp_path: Path,
+) -> None:
+    """A condition that is not static yet is loaded anyway, so its errors still show."""
+    entry = _write(
+        tmp_path,
+        "entry.yaml",
+        'packages:\n  opt: !include {file: missing.yaml, condition: "${enable}"}\n',
+    )
+    assert discover_user_yaml_files(entry).load_errors
+
+
+def test_discover_user_yaml_files_records_directory_links(tmp_path: Path) -> None:
+    """A file reached through a symlinked directory below the config directory
+    records that link and its target."""
+    shared = tmp_path / "shared"
+    _write(shared, "wifi.yaml", "ssid: x\n")
+    config_dir = tmp_path / "config"
+    entry = _write(config_dir, "entry.yaml", "wifi: !include linked/wifi.yaml\n")
+    (config_dir / "linked").symlink_to(shared)
+    discovered = discover_user_yaml_files(entry)
+    assert discovered.links == {config_dir.resolve() / "linked": shared.resolve()}
+
+
+def test_registered_secret_values_maps_a_loaded_secret_to_its_name(
+    tmp_path: Path,
+) -> None:
+    """Values loaded through `!secret` are listed with the name they came from."""
+    _write(tmp_path, "secrets.yaml", "wifi_password: hunter2\n")
+    entry = _write(tmp_path, "entry.yaml", "wifi:\n  password: !secret wifi_password\n")
+    yaml_util.load_yaml(entry)
+    assert yaml_util.registered_secret_values()["hunter2"] == "wifi_password"
+
+
+def test_discover_user_yaml_files_lists_only_existing_files_after_a_load_error(
+    tmp_path: Path,
+) -> None:
+    """A failed top-level load still reports only files that exist."""
+    entry = _write(tmp_path, "entry.yaml", "!include missing.yaml\n")
+    discovered = discover_user_yaml_files(entry)
+    assert discovered.load_errors
+    assert discovered.files == [entry.resolve()]
+    assert list(discovered.aliases) == [entry.resolve()]
+
+
+def test_dump_literal_inside_a_shared_container_keeps_the_container() -> None:
+    """A literal must not take over the alias of the list that holds it."""
+    shared = [yaml_util.make_literal({"x": "y"})]
+    out = yaml_util.dump({"a": shared, "b": shared})
+    loaded = yaml.load(out.replace("!literal", ""), Loader=yaml.SafeLoader)
+    assert loaded == {"a": [{"x": "y"}], "b": [{"x": "y"}]}