Merge remote-tracking branch 'origin/cv-sensitive-redact-sentinel' into integration

tests/unit_tests/components/test_rp2040.py

@@ -1,6 +1,11 @@
-"""Tests for RP2040 component public helpers."""
+"""Tests for RP2040 component public helpers and variant detection."""
 
-from esphome.components.rp2040 import board_id_has_wifi
+import pytest
+
+from esphome.components.rp2040 import _detect_variant, board_id_has_wifi
+from esphome.components.rp2040.const import VARIANT_RP2040, VARIANT_RP2350
+import esphome.config_validation as cv
+from esphome.const import CONF_BOARD, CONF_VARIANT
 
 
 def test_board_id_has_wifi_for_known_wifi_board() -> None:
@@ -27,3 +32,61 @@ def test_board_id_has_wifi_for_unknown_board_returns_true() -> None:
     "no CYW43" guard at compile time.
     """
     assert board_id_has_wifi("not-a-real-board-id") is True
+
+
+def test_detect_variant_derives_variant_from_board() -> None:
+    """Board alone resolves to the matching variant."""
+    result = _detect_variant({CONF_BOARD: "rpipicow"})
+    assert result[CONF_BOARD] == "rpipicow"
+    assert result[CONF_VARIANT] == VARIANT_RP2040
+
+
+def test_detect_variant_derives_variant_from_rp2350_board() -> None:
+    """An RP2350 board resolves to ``RP2350``."""
+    result = _detect_variant({CONF_BOARD: "rpipico2"})
+    assert result[CONF_BOARD] == "rpipico2"
+    assert result[CONF_VARIANT] == VARIANT_RP2350
+
+
+def test_detect_variant_only_picks_default_board_rp2040() -> None:
+    """Variant alone picks Pico W as the canonical RP2040 board."""
+    result = _detect_variant({CONF_VARIANT: VARIANT_RP2040})
+    assert result[CONF_BOARD] == "rpipicow"
+    assert result[CONF_VARIANT] == VARIANT_RP2040
+
+
+def test_detect_variant_only_picks_default_board_rp2350() -> None:
+    """Variant alone picks Pico 2 W as the canonical RP2350 board."""
+    result = _detect_variant({CONF_VARIANT: VARIANT_RP2350})
+    assert result[CONF_BOARD] == "rpipico2w"
+    assert result[CONF_VARIANT] == VARIANT_RP2350
+
+
+def test_detect_variant_matching_explicit_variant_passes() -> None:
+    """Specifying both a board and the matching variant is allowed."""
+    result = _detect_variant({CONF_BOARD: "rpipico2", CONF_VARIANT: VARIANT_RP2350})
+    assert result[CONF_BOARD] == "rpipico2"
+    assert result[CONF_VARIANT] == VARIANT_RP2350
+
+
+def test_detect_variant_mismatched_variant_raises() -> None:
+    """Board/variant mismatch must be rejected and name the offending board."""
+    with pytest.raises(
+        cv.Invalid, match=r"does not match the selected board 'rpipicow'"
+    ):
+        _detect_variant({CONF_BOARD: "rpipicow", CONF_VARIANT: VARIANT_RP2350})
+
+
+def test_detect_variant_unknown_board_without_variant_raises() -> None:
+    """Unknown board with no variant tells the user how to recover."""
+    with pytest.raises(cv.Invalid, match="please specify the chip variant"):
+        _detect_variant({CONF_BOARD: "not-a-real-board"})
+
+
+def test_detect_variant_unknown_board_with_variant_passes() -> None:
+    """Unknown board + explicit variant is accepted (with a warning)."""
+    result = _detect_variant(
+        {CONF_BOARD: "not-a-real-board", CONF_VARIANT: VARIANT_RP2040}
+    )
+    assert result[CONF_BOARD] == "not-a-real-board"
+    assert result[CONF_VARIANT] == VARIANT_RP2040

tests/unit_tests/components/test_time.py

@@ -70,11 +70,11 @@ def test_numeric_offset_slash() -> None:
 
 
 def test_star() -> None:
-    assert _parse_cron_part("*", 0, 59, {}) == set(range(0, 60))
+    assert _parse_cron_part("*", 0, 59, {}) == set(range(60))
 
 
 def test_question() -> None:
-    assert _parse_cron_part("?", 0, 59, {}) == set(range(0, 60))
+    assert _parse_cron_part("?", 0, 59, {}) == set(range(60))
 
 
 def test_range() -> None:

tests/unit_tests/core/test_config.py

@@ -486,7 +486,7 @@ def test_preload_core_config_basic(setup_core: Path) -> None:
     assert CONF_BUILD_PATH in config[CONF_ESPHOME]
     # Verify default build path is "build/<device_name>"
     build_path = config[CONF_ESPHOME][CONF_BUILD_PATH]
-    assert build_path.endswith(os.path.join("build", "test_device"))
+    assert build_path.endswith(str(Path("build") / "test_device"))
 
 
 def test_preload_core_config_with_build_path(setup_core: Path) -> None:
@@ -523,7 +523,7 @@ def test_preload_core_config_env_build_path(setup_core: Path) -> None:
     assert "test_device" in config[CONF_ESPHOME][CONF_BUILD_PATH]
     # Verify it uses the env var path with device name appended
     build_path = config[CONF_ESPHOME][CONF_BUILD_PATH]
-    expected_path = os.path.join("/env/build", "test_device")
+    expected_path = str(Path("/env/build") / "test_device")
     assert build_path == expected_path or build_path == expected_path.replace(
         "/", os.sep
     )
@@ -739,7 +739,7 @@ async def test_add_includes_with_single_file(
     """Test add_includes copies a single header file to build directory."""
     CORE.config_path = tmp_path / "config.yaml"
     CORE.build_path = tmp_path / "build"
-    os.makedirs(CORE.build_path, exist_ok=True)
+    CORE.build_path.mkdir(parents=True, exist_ok=True)
 
     # Create include file
     include_file = tmp_path / "my_header.h"
@@ -769,7 +769,7 @@ async def test_add_includes_with_directory_unix(
     """Test add_includes copies all files from a directory on Unix."""
     CORE.config_path = tmp_path / "config.yaml"
     CORE.build_path = tmp_path / "build"
-    os.makedirs(CORE.build_path, exist_ok=True)
+    CORE.build_path.mkdir(parents=True, exist_ok=True)
 
     # Create include directory with files
     include_dir = tmp_path / "includes"
@@ -814,7 +814,7 @@ async def test_add_includes_with_directory_windows(
     """Test add_includes copies all files from a directory on Windows."""
     CORE.config_path = tmp_path / "config.yaml"
     CORE.build_path = tmp_path / "build"
-    os.makedirs(CORE.build_path, exist_ok=True)
+    CORE.build_path.mkdir(parents=True, exist_ok=True)
 
     # Create include directory with files
     include_dir = tmp_path / "includes"
@@ -856,7 +856,7 @@ async def test_add_includes_with_multiple_sources(
     """Test add_includes with multiple files and directories."""
     CORE.config_path = tmp_path / "config.yaml"
     CORE.build_path = tmp_path / "build"
-    os.makedirs(CORE.build_path, exist_ok=True)
+    CORE.build_path.mkdir(parents=True, exist_ok=True)
 
     # Create various include sources
     single_file = tmp_path / "single.h"
@@ -884,7 +884,7 @@ async def test_add_includes_empty_directory(
     """Test add_includes with an empty directory doesn't fail."""
     CORE.config_path = tmp_path / "config.yaml"
     CORE.build_path = tmp_path / "build"
-    os.makedirs(CORE.build_path, exist_ok=True)
+    CORE.build_path.mkdir(parents=True, exist_ok=True)
 
     # Create empty directory
     empty_dir = tmp_path / "empty"
@@ -906,7 +906,7 @@ async def test_add_includes_preserves_directory_structure_unix(
     """Test that add_includes preserves relative directory structure on Unix."""
     CORE.config_path = tmp_path / "config.yaml"
     CORE.build_path = tmp_path / "build"
-    os.makedirs(CORE.build_path, exist_ok=True)
+    CORE.build_path.mkdir(parents=True, exist_ok=True)
 
     # Create nested directory structure
     lib_dir = tmp_path / "lib"
@@ -940,7 +940,7 @@ async def test_add_includes_preserves_directory_structure_windows(
     """Test that add_includes preserves relative directory structure on Windows."""
     CORE.config_path = tmp_path / "config.yaml"
     CORE.build_path = tmp_path / "build"
-    os.makedirs(CORE.build_path, exist_ok=True)
+    CORE.build_path.mkdir(parents=True, exist_ok=True)
 
     # Create nested directory structure
     lib_dir = tmp_path / "lib"
@@ -973,7 +973,7 @@ async def test_add_includes_overwrites_existing_files(
     """Test that add_includes overwrites existing files in build directory."""
     CORE.config_path = tmp_path / "config.yaml"
     CORE.build_path = tmp_path / "build"
-    os.makedirs(CORE.build_path, exist_ok=True)
+    CORE.build_path.mkdir(parents=True, exist_ok=True)
 
     # Create include file
     include_file = tmp_path / "header.h"

tests/unit_tests/test_config_normalization.py

@@ -67,7 +67,7 @@ def test_iter_component_configs_with_multi_conf(mock_get_component: Mock) -> Non
     configs = list(config.iter_component_configs(test_config))
     assert len(configs) == 2
 
-    for domain, component, conf in configs:
+    for domain, _component, conf in configs:
         assert domain == "switch"
         assert "name" in conf
 

tests/unit_tests/test_config_validation.py

@@ -27,6 +27,7 @@ from esphome.const import (
     SCHEDULER_DONT_RUN,
 )
 from esphome.core import CORE, HexInt, Lambda
+from esphome.yaml_util import SensitiveStr
 
 
 def test_check_not_templatable__invalid():
@@ -127,6 +128,85 @@ def test_string_string__invalid(value):
         config_validation.string_strict(value)
 
 
+def test_sensitive__default_delegates_to_string() -> None:
+    validator = config_validation.sensitive()
+
+    assert isinstance(validator, config_validation.SensitiveValidator)
+    assert validator.inner is config_validation.string
+    assert validator("hunter2") == "hunter2"
+    assert validator(42) == "42"
+
+
+def test_sensitive__custom_inner_delegates_validation() -> None:
+    validator = config_validation.sensitive(config_validation.string_strict)
+
+    assert validator.inner is config_validation.string_strict
+    assert validator("abc") == "abc"
+    with pytest.raises(Invalid, match="Must be string, got"):
+        validator(123)
+
+
+def test_sensitive__wraps_string_result_in_sensitive_str() -> None:
+    validator = config_validation.sensitive()
+    result = validator("hunter2")
+
+    assert isinstance(result, SensitiveStr)
+    assert isinstance(result, str)
+    assert result == "hunter2"
+
+
+def test_sensitive__does_not_double_tag_already_sensitive() -> None:
+    # If the inner validator already returns a SensitiveStr (e.g., nested
+    # cv.sensitive wrappers), re-tagging is a no-op rather than a new
+    # SensitiveStr around the same value.
+    pre_tagged = SensitiveStr("hunter2")
+
+    def inner(_value):
+        return pre_tagged
+
+    validator = config_validation.sensitive(inner)
+    result = validator("anything")
+
+    assert result is pre_tagged
+
+
+def test_sensitive__non_string_result_passes_through() -> None:
+    # If an inner validator returns something other than a string (e.g., a
+    # Lambda template), the sensitive wrapper must not coerce it.
+    sentinel = object()
+
+    def inner(_value):
+        return sentinel
+
+    validator = config_validation.sensitive(inner)
+    assert validator("anything") is sentinel
+
+
+def test_sensitive__is_detectable_via_isinstance() -> None:
+    validator = config_validation.sensitive()
+
+    assert isinstance(validator, config_validation.SensitiveValidator)
+
+
+def test_sensitive__repr_mirrors_inner() -> None:
+    # The schema dump dedups on ``repr(schema)``; mirroring the inner
+    # validator's repr keeps two ``cv.sensitive(cv.string)`` wrappers
+    # interchangeable for that purpose and avoids leaking the wrapper as
+    # noise in voluptuous error messages.
+    assert repr(config_validation.sensitive(config_validation.string)) == repr(
+        config_validation.string
+    )
+    assert repr(config_validation.sensitive(config_validation.string)) == repr(
+        config_validation.sensitive(config_validation.string)
+    )
+
+
+def test_sensitive_key_fragments__covers_common_terms() -> None:
+    assert isinstance(config_validation.SENSITIVE_KEY_FRAGMENTS, frozenset)
+    for term in ("password", "passcode", "secret", "token", "api_key", "apikey", "psk"):
+        assert term in config_validation.SENSITIVE_KEY_FRAGMENTS
+
+
 @given(
     builds(
         lambda v: "mdi:" + v,

tests/unit_tests/test_espidf_component.py

@@ -261,9 +261,14 @@ def test_check_library_data_invalid_platform(esp32_idf_core):
         _check_library_data({"platforms": ["other"], "frameworks": "*"})
 
 
-def test_check_library_data_invalid_framework(esp32_idf_core):
-    with pytest.raises(InvalidIDFComponent):
-        _check_library_data({"platforms": "*", "frameworks": ["other"]})
+def test_check_library_data_invalid_framework(
+    esp32_idf_core: None, caplog: pytest.LogCaptureFixture
+) -> None:
+    # Framework mismatch is a warning, not a hard skip: the library is still
+    # included so that PIO manifests that only list "arduino" (but actually
+    # compile under IDF) can be used without forking them.
+    _check_library_data({"name": "lib", "platforms": "*", "frameworks": ["other"]})
+    assert "do not include 'espidf'" in caplog.text
 
 
 def test_extra_script_captures_libpath_libs_and_defines(tmp_path):
@@ -288,7 +293,7 @@ def test_extra_script_captures_libpath_libs_and_defines(tmp_path):
 
     result = run_extra_script(script, library_dir=tmp_path, idf_target="esp32")
 
-    assert result.libpath == [os.path.join("src", "esp32")]
+    assert result.libpath == [str(Path("src") / "esp32")]
     assert result.libs == ["algobsec"]
     assert ("BAR", "1") in result.cppdefines
     assert "FOO" in result.cppdefines

tests/unit_tests/test_espidf_framework.py

@@ -0,0 +1,156 @@
+"""Tests for esphome.espidf.framework helpers."""
+
+# pylint: disable=protected-access
+
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from esphome.espidf.framework import _clone_idf_with_submodules, _parse_git_source
+
+
+@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,
+        )

tests/unit_tests/test_external_files.py

@@ -120,7 +120,7 @@ def test_is_file_recent_with_old_file(setup_core: Path) -> None:
 
     old_time = time.time() - 7200
     mock_stat = MagicMock()
-    mock_stat.st_ctime = old_time
+    mock_stat.st_mtime = old_time
 
     with patch.object(Path, "stat", return_value=mock_stat):
         refresh = TimePeriod(seconds=3600)
@@ -147,7 +147,7 @@ def test_is_file_recent_with_zero_refresh(setup_core: Path) -> None:
 
     # Mock stat to return a time 10 seconds ago
     mock_stat = MagicMock()
-    mock_stat.st_ctime = time.time() - 10
+    mock_stat.st_mtime = time.time() - 10
     with patch.object(Path, "stat", return_value=mock_stat):
         refresh = TimePeriod(seconds=0)
         result = external_files.is_file_recent(test_file, refresh)

tests/unit_tests/test_git.py

@@ -1,10 +1,10 @@
 """Tests for git.py module."""
 
-from datetime import datetime, timedelta
 import os
 from pathlib import Path
+import time
 from typing import Any
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
 
 import pytest
 
@@ -34,9 +34,9 @@ def _setup_old_repo(repo_dir: Path, days_old: int = 2) -> None:
     # Create FETCH_HEAD file with old timestamp
     fetch_head = git_dir / "FETCH_HEAD"
     fetch_head.write_text("test")
-    old_time = datetime.now() - timedelta(days=days_old)
+    old_time = time.time() - days_old * 86400
     fetch_head.touch()
-    os.utime(fetch_head, (old_time.timestamp(), old_time.timestamp()))
+    os.utime(fetch_head, (old_time, old_time))
 
 
 def _get_git_command_type(cmd: list[str]) -> str | None:
@@ -285,10 +285,10 @@ def test_clone_or_update_with_refresh_updates_old_repo(
     # Create FETCH_HEAD file with old timestamp (2 days ago)
     fetch_head = git_dir / "FETCH_HEAD"
     fetch_head.write_text("test")
-    old_time = datetime.now() - timedelta(days=2)
+    old_time = time.time() - 2 * 86400
     fetch_head.touch()  # Create the file
     # Set modification time to 2 days ago
-    os.utime(fetch_head, (old_time.timestamp(), old_time.timestamp()))
+    os.utime(fetch_head, (old_time, old_time))
 
     # Mock git command responses
     mock_run_git_command.return_value = "abc123"  # SHA for rev-parse
@@ -333,10 +333,10 @@ def test_clone_or_update_with_refresh_skips_fresh_repo(
     # Create FETCH_HEAD file with recent timestamp (1 hour ago)
     fetch_head = git_dir / "FETCH_HEAD"
     fetch_head.write_text("test")
-    recent_time = datetime.now() - timedelta(hours=1)
+    recent_time = time.time() - 3600
     fetch_head.touch()  # Create the file
     # Set modification time to 1 hour ago
-    os.utime(fetch_head, (recent_time.timestamp(), recent_time.timestamp()))
+    os.utime(fetch_head, (recent_time, recent_time))
 
     # Call with refresh=1d (1 day)
     refresh = TimePeriodSeconds(days=1)
@@ -409,10 +409,10 @@ def test_clone_or_update_with_none_refresh_always_updates(
     # Create FETCH_HEAD file with very recent timestamp (1 second ago)
     fetch_head = git_dir / "FETCH_HEAD"
     fetch_head.write_text("test")
-    recent_time = datetime.now() - timedelta(seconds=1)
+    recent_time = time.time() - 1
     fetch_head.touch()  # Create the file
     # Set modification time to 1 second ago
-    os.utime(fetch_head, (recent_time.timestamp(), recent_time.timestamp()))
+    os.utime(fetch_head, (recent_time, recent_time))
 
     # Mock git command responses
     mock_run_git_command.return_value = "abc123"  # SHA for rev-parse
@@ -1001,3 +1001,304 @@ def test_refresh_picks_up_new_remote_commits(
         "--hard",
         "old_sha",
     ]
+
+
+def test_resolve_symlink_stub_returns_none_on_non_windows(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """On non-Windows, resolve_symlink_stub returns None without calling git."""
+    repo_dir = tmp_path / "repo"
+    repo_dir.mkdir()
+    stub = repo_dir / "file.yaml"
+    stub.write_text("static/file.yaml")
+
+    with patch("esphome.git.sys.platform", "linux"):
+        result = git.resolve_symlink_stub(repo_dir, stub)
+
+    assert result is None
+    mock_run_git_command.assert_not_called()
+
+
+def test_resolve_symlink_stub_returns_target_for_mode_120000(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """A mode-120000 file is recognised as a stub; its target Path is returned."""
+    repo_dir = tmp_path / "repo"
+    repo_dir.mkdir()
+    (repo_dir / "static").mkdir()
+
+    target = repo_dir / "static" / "real.yaml"
+    target.write_text("esphome:\n  name: real\n")
+
+    stub = repo_dir / "real.yaml"
+    stub.write_text("static/real.yaml")
+
+    mock_run_git_command.return_value = "120000 abc123 0\treal.yaml"
+
+    with patch("esphome.git.sys.platform", "win32"):
+        result = git.resolve_symlink_stub(repo_dir, stub)
+
+    assert result == target.resolve()
+    # Stub file itself was not modified — only inspected.
+    assert stub.read_text() == "static/real.yaml"
+
+
+def test_resolve_symlink_stub_resolves_relative_parent_paths(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """Symlink targets with ``..`` segments resolve correctly within the repo."""
+    repo_dir = tmp_path / "repo"
+    (repo_dir / "subdir").mkdir(parents=True)
+    (repo_dir / "static").mkdir()
+
+    target = repo_dir / "static" / "shared.yaml"
+    target.write_text("shared content")
+
+    stub = repo_dir / "subdir" / "shared.yaml"
+    stub.write_text("../static/shared.yaml")
+
+    mock_run_git_command.return_value = "120000 abc123 0\tsubdir/shared.yaml"
+
+    with patch("esphome.git.sys.platform", "win32"):
+        result = git.resolve_symlink_stub(repo_dir, stub)
+
+    assert result == target.resolve()
+
+
+def test_resolve_symlink_stub_refuses_escape_outside_repo(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """A symlink pointing outside the repository is not followed."""
+    outside = tmp_path / "outside.yaml"
+    outside.write_text("sensitive")
+
+    repo_dir = tmp_path / "repo"
+    repo_dir.mkdir()
+
+    stub = repo_dir / "escape.yaml"
+    stub.write_text("../outside.yaml")
+
+    mock_run_git_command.return_value = "120000 abc123 0\tescape.yaml"
+
+    with patch("esphome.git.sys.platform", "win32"):
+        result = git.resolve_symlink_stub(repo_dir, stub)
+
+    assert result is None
+
+
+def test_resolve_symlink_stub_returns_none_for_real_symlink(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """A real symlink already opens transparently, so the helper short-circuits.
+
+    Skipped on Windows where symlink creation requires
+    SeCreateSymbolicLinkPrivilege.
+    """
+    if os.name == "nt":
+        pytest.skip("Requires symlink-creation privilege on Windows")
+
+    repo_dir = tmp_path / "repo"
+    repo_dir.mkdir()
+    target = repo_dir / "real.yaml"
+    target.write_text("real content")
+
+    real_link = repo_dir / "link.yaml"
+    real_link.symlink_to("real.yaml")
+
+    with patch("esphome.git.sys.platform", "win32"):
+        result = git.resolve_symlink_stub(repo_dir, real_link)
+
+    assert result is None
+    # No git call needed for real symlinks.
+    mock_run_git_command.assert_not_called()
+
+
+def test_resolve_symlink_stub_returns_none_for_regular_file(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """A regular file (mode 100644) whose content looks path-shaped is not
+    followed."""
+    repo_dir = tmp_path / "repo"
+    repo_dir.mkdir()
+
+    regular = repo_dir / "looks_like_path.txt"
+    regular.write_text("static/something.yaml")
+
+    mock_run_git_command.return_value = "100644 abc123 0\tlooks_like_path.txt"
+
+    with patch("esphome.git.sys.platform", "win32"):
+        result = git.resolve_symlink_stub(repo_dir, regular)
+
+    assert result is None
+
+
+def test_resolve_symlink_stub_returns_none_when_git_fails(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """If ``git ls-files`` fails (e.g. not a repo), the helper returns None."""
+    repo_dir = tmp_path / "repo"
+    repo_dir.mkdir()
+
+    stub = repo_dir / "real.yaml"
+    stub.write_text("static/real.yaml")
+
+    mock_run_git_command.side_effect = GitCommandError("ls-files exploded")
+
+    with patch("esphome.git.sys.platform", "win32"):
+        result = git.resolve_symlink_stub(repo_dir, stub)
+
+    assert result is None
+
+
+def test_resolve_symlink_stub_returns_none_for_non_utf8_content(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """A file whose bytes are not valid UTF-8 must not raise — return None."""
+    repo_dir = tmp_path / "repo"
+    repo_dir.mkdir()
+
+    stub = repo_dir / "binary.bin"
+    stub.write_bytes(b"\xff\xfe\x00\xff")
+
+    mock_run_git_command.return_value = "120000 abc123 0\tbinary.bin"
+
+    with patch("esphome.git.sys.platform", "win32"):
+        result = git.resolve_symlink_stub(repo_dir, stub)
+
+    assert result is None
+
+
+def test_resolve_symlink_stub_preserves_whitespace_in_target(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """Only trailing CR/LF is stripped — internal whitespace is preserved."""
+    repo_dir = tmp_path / "repo"
+    repo_dir.mkdir()
+    target_dir = repo_dir / "dir with spaces"
+    target_dir.mkdir()
+    target = target_dir / "real.yaml"
+    target.write_text("hello")
+
+    stub = repo_dir / "link.yaml"
+    # Trailing newline (as git's checkout may append) is stripped, but
+    # whitespace inside the target path itself must survive.
+    stub.write_bytes(b"dir with spaces/real.yaml\n")
+
+    mock_run_git_command.return_value = "120000 abc123 0\tlink.yaml"
+
+    with patch("esphome.git.sys.platform", "win32"):
+        result = git.resolve_symlink_stub(repo_dir, stub)
+
+    assert result == target.resolve()
+
+
+def test_resolve_symlink_stub_returns_none_for_directory_target(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """A symlink pointing at a directory has no file content to load."""
+    repo_dir = tmp_path / "repo"
+    repo_dir.mkdir()
+    (repo_dir / "dir_target").mkdir()
+
+    stub = repo_dir / "link_to_dir"
+    stub.write_text("dir_target")
+
+    mock_run_git_command.return_value = "120000 abc123 0\tlink_to_dir"
+
+    with patch("esphome.git.sys.platform", "win32"):
+        result = git.resolve_symlink_stub(repo_dir, stub)
+
+    assert result is None
+
+
+def test_resolve_symlink_stub_returns_none_when_resolve_raises(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """Path.resolve() raising (e.g. on a malformed target) must not propagate."""
+    repo_dir = tmp_path / "repo"
+    repo_dir.mkdir()
+
+    stub = repo_dir / "broken.yaml"
+    stub.write_text("ignored")
+
+    mock_run_git_command.return_value = "120000 abc123 0\tbroken.yaml"
+
+    with (
+        patch("esphome.git.sys.platform", "win32"),
+        patch.object(Path, "resolve", side_effect=OSError("bad path")),
+    ):
+        result = git.resolve_symlink_stub(repo_dir, stub)
+
+    assert result is None
+
+
+def test_resolve_symlink_stub_returns_none_when_file_missing(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """A file path that doesn't exist is rejected before git is consulted."""
+    repo_dir = tmp_path / "repo"
+    repo_dir.mkdir()
+
+    missing = repo_dir / "ghost.yaml"  # not created
+
+    with patch("esphome.git.sys.platform", "win32"):
+        result = git.resolve_symlink_stub(repo_dir, missing)
+
+    assert result is None
+    mock_run_git_command.assert_not_called()
+
+
+def test_resolve_symlink_stub_returns_none_when_path_outside_repo(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """A file path that isn't under repo_dir is rejected (ValueError from relative_to)."""
+    repo_dir = tmp_path / "repo"
+    repo_dir.mkdir()
+
+    outside = tmp_path / "stray.yaml"
+    outside.write_text("something")
+
+    with patch("esphome.git.sys.platform", "win32"):
+        result = git.resolve_symlink_stub(repo_dir, outside)
+
+    assert result is None
+    mock_run_git_command.assert_not_called()
+
+
+def test_resolve_symlink_stub_returns_none_when_untracked(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """Empty `git ls-files` output (untracked file) makes the helper return None."""
+    repo_dir = tmp_path / "repo"
+    repo_dir.mkdir()
+
+    stub = repo_dir / "untracked.yaml"
+    stub.write_text("static/foo.yaml")
+
+    mock_run_git_command.return_value = ""
+
+    with patch("esphome.git.sys.platform", "win32"):
+        result = git.resolve_symlink_stub(repo_dir, stub)
+
+    assert result is None
+
+
+def test_resolve_symlink_stub_returns_none_when_read_bytes_raises(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """An OSError from read_bytes() (e.g. file vanished mid-call) must not propagate."""
+    repo_dir = tmp_path / "repo"
+    repo_dir.mkdir()
+
+    stub = repo_dir / "racy.yaml"
+    stub.write_text("static/racy.yaml")
+
+    mock_run_git_command.return_value = "120000 abc123 0\tracy.yaml"
+
+    with (
+        patch("esphome.git.sys.platform", "win32"),
+        patch.object(Path, "read_bytes", side_effect=OSError("vanished")),
+    ):
+        result = git.resolve_symlink_stub(repo_dir, stub)
+
+    assert result is None

tests/unit_tests/test_helpers.py

@@ -7,7 +7,7 @@ import stat
 from unittest.mock import MagicMock, patch
 
 from aioesphomeapi.host_resolver import AddrInfo, IPv4Sockaddr, IPv6Sockaddr
-from hypothesis import given
+from hypothesis import given, settings
 from hypothesis.strategies import ip_addresses
 import pytest
 
@@ -151,6 +151,7 @@ def test_is_ip_address__invalid(host):
     assert actual is False
 
 
+@settings(deadline=None)
 @given(value=ip_addresses(v=4).map(str))
 def test_is_ip_address__valid(value):
     actual = helpers.is_ip_address(value)

tests/unit_tests/test_main.py

@@ -11,7 +11,7 @@ from pathlib import Path
 import re
 import sys
 import time
-from typing import Any
+from typing import Any, Self
 from unittest.mock import AsyncMock, MagicMock, Mock, patch
 
 import pytest
@@ -22,6 +22,7 @@ from esphome.__main__ import (
     Purpose,
     _get_configured_xtal_freq,
     _make_crystal_freq_callback,
+    _redact_with_legacy_fallback,
     _resolve_network_devices,
     _validate_bootloader_binary,
     _validate_partition_table_binary,
@@ -29,6 +30,7 @@ from esphome.__main__ import (
     command_analyze_memory,
     command_bundle,
     command_clean_all,
+    command_config,
     command_config_hash,
     command_rename,
     command_run,
@@ -340,6 +342,135 @@ def mock_ram_strings_analyzer() -> Generator[Mock]:
         yield mock_class
 
 
+def test_redact_with_legacy_fallback__wraps_unmarked_field(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Unmarked sensitive-shaped fields are redacted; a deprecation warning
+    is emitted naming the field."""
+    with caplog.at_level(logging.WARNING, logger="esphome.__main__"):
+        out = _redact_with_legacy_fallback("password: hunter2\n")
+    assert "password: \\033[8mhunter2\\033[28m" in out
+    assert any(
+        "password" in rec.message and "cv.sensitive" in rec.message
+        for rec in caplog.records
+    )
+
+
+def test_redact_with_legacy_fallback__skips_already_wrapped(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Values already wrapped by the SensitiveStr representer don't trigger
+    the heuristic or the warning."""
+    wrapped = "password: \\033[8mhunter2\\033[28m\n"
+    with caplog.at_level(logging.WARNING, logger="esphome.__main__"):
+        out = _redact_with_legacy_fallback(wrapped)
+    assert out == wrapped
+    assert not any("legacy substring" in rec.message for rec in caplog.records)
+
+
+def test_redact_with_legacy_fallback__captures_full_field_name(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """The warning names the actual field, not just the matched fragment."""
+    with caplog.at_level(logging.WARNING, logger="esphome.__main__"):
+        _redact_with_legacy_fallback("encryption_key: abc\n")
+    assert any("encryption_key" in rec.message for rec in caplog.records)
+
+
+def test_redact_with_legacy_fallback__deduplicates_warnings(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """One warning per unique field name even if it appears many times."""
+    text = "password: a\npassword: b\npassword: c\n"
+    with caplog.at_level(logging.WARNING, logger="esphome.__main__"):
+        _redact_with_legacy_fallback(text)
+    password_warnings = [rec for rec in caplog.records if "'password'" in rec.message]
+    assert len(password_warnings) == 1
+
+
+def test_redact_with_legacy_fallback__skips_lambda_values(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """``!lambda`` first line is structural, body is unreachable by a
+    single-line regex anyway, and tagged fields shouldn't trigger a warning."""
+    text = '          ssid: !lambda |-\n            return "x";\n'
+    with caplog.at_level(logging.WARNING, logger="esphome.__main__"):
+        out = _redact_with_legacy_fallback(text)
+    assert out == text
+    assert not any("legacy substring" in rec.message for rec in caplog.records)
+
+
+def test_redact_with_legacy_fallback__skips_secret_references(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """``!secret name`` is the dumper's user-friendly representation; the
+    name isn't the secret, so wrapping it would clobber the round-trip."""
+    text = "          password: !secret wifi_password\n"
+    with caplog.at_level(logging.WARNING, logger="esphome.__main__"):
+        out = _redact_with_legacy_fallback(text)
+    assert out == text
+    assert not any("legacy substring" in rec.message for rec in caplog.records)
+
+
+def test_redact_with_legacy_fallback__does_not_match_fragment_in_middle(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Fragment must end the field name; embedded matches like
+    ``key_value_pair`` are unrelated to a sensitive key and must not be
+    redacted (matching the prior regex's scope)."""
+    with caplog.at_level(logging.WARNING, logger="esphome.__main__"):
+        out = _redact_with_legacy_fallback("key_value_pair: abc\n")
+    assert "\\033[8m" not in out
+    assert not any("legacy substring" in rec.message for rec in caplog.records)
+
+
+def test_redact_with_legacy_fallback__does_not_match_fragment_as_suffix(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Fragment must start the name or follow ``_``; ``monkey:`` shouldn't
+    fire a 'legacy heuristic' warning because there's no sensitive field
+    here — the user has nothing to migrate."""
+    with caplog.at_level(logging.WARNING, logger="esphome.__main__"):
+        out = _redact_with_legacy_fallback("monkey: 1234\n")
+    assert "\\033[8m" not in out
+    assert not any("legacy substring" in rec.message for rec in caplog.records)
+
+
+def test_command_config__invokes_legacy_fallback_when_redacting(
+    tmp_path: Path, capfd: CaptureFixture[str]
+) -> None:
+    """``command_config`` runs the legacy fallback on the dumped output when
+    ``--show-secrets`` is off. Cover the wiring (not just the helper).
+    """
+    setup_core(tmp_path=tmp_path, config={"esphome": {"name": "test"}})
+    args = MockArgs()
+    args.show_secrets = False
+
+    result = command_config(args, {"wifi": {"password": "hunter2"}})
+
+    assert result == 0
+    output = capfd.readouterr().out
+    assert "\\033[8mhunter2\\033[28m" in output
+
+
+def test_command_config__show_secrets_skips_redaction(
+    tmp_path: Path, capfd: CaptureFixture[str]
+) -> None:
+    """With ``--show-secrets`` the helper isn't invoked and the value
+    renders raw.
+    """
+    setup_core(tmp_path=tmp_path, config={"esphome": {"name": "test"}})
+    args = MockArgs()
+    args.show_secrets = True
+
+    result = command_config(args, {"wifi": {"password": "hunter2"}})
+
+    assert result == 0
+    output = capfd.readouterr().out
+    assert "hunter2" in output
+    assert "\\033[8m" not in output
+
+
 def test_choose_upload_log_host_with_string_default() -> None:
     """Test with a single string default device."""
     setup_core()
@@ -5110,11 +5241,11 @@ class MockSerial:
         self.timeout = 0.1
         self._is_open = False
 
-    def __enter__(self) -> MockSerial:
+    def __enter__(self) -> Self:
         self._is_open = True
         return self
 
-    def __exit__(self, *args: Any) -> None:
+    def __exit__(self, *args: object) -> None:
         self._is_open = False
 
     @property

tests/unit_tests/test_storage_json.py

@@ -576,8 +576,8 @@ def test_esphome_storage_json_last_update_check_property() -> None:
     assert result.hour == 10
     assert result.minute == 30
 
-    # Test setter
-    new_date = datetime(2024, 2, 20, 15, 45, 30)
+    # Test setter — naive datetime matches the storage round-trip format.
+    new_date = datetime(2024, 2, 20, 15, 45, 30)  # noqa: DTZ001
     storage.last_update_check = new_date
     assert storage.last_update_check_str == "2024-02-20T15:45:30"
 

tests/unit_tests/test_substitutions.py

@@ -1,4 +1,3 @@
-import glob
 import logging
 from pathlib import Path
 from typing import Any
@@ -106,7 +105,7 @@ REMOTES = {
 # Collect all input YAML files for test_substitutions_fixtures parametrized tests:
 HERE = Path(__file__).parent
 BASE_DIR = HERE / "fixtures" / "substitutions"
-SOURCES = sorted(glob.glob(str(BASE_DIR / "*.input.yaml")))
+SOURCES = sorted(str(p) for p in BASE_DIR.glob("*.input.yaml"))
 assert SOURCES, f"test_substitutions_fixtures: No input YAML files found in {BASE_DIR}"
 
 
@@ -838,3 +837,86 @@ def test_include_vars_applied_to_lambda_value(tmp_path: Path) -> None:
 
     assert isinstance(result["value"], Lambda)
     assert result["value"].value == 'return "bar";'
+
+
+@patch("esphome.git.resolve_symlink_stub")
+@patch("esphome.git.clone_or_update")
+def test_remote_package_symlink_stub_is_followed(
+    mock_clone_or_update: MagicMock,
+    mock_resolve_symlink_stub: MagicMock,
+    tmp_path: Path,
+) -> None:
+    """When a package YAML is a scalar (symlink stub) and resolve_symlink_stub
+    returns a target, the loader follows the target and uses its content."""
+    CORE.config_path = tmp_path / "test.yaml"
+
+    repo_dir = tmp_path / "repo"
+    repo_dir.mkdir()
+    (repo_dir / "static").mkdir()
+
+    # Stub file: content is the target path string (simulating Windows behavior).
+    stub = repo_dir / "file1.yaml"
+    stub.write_text("static/file1.yaml")
+
+    # Real target with valid YAML mapping.
+    target = repo_dir / "static" / "file1.yaml"
+    target.write_text("substitutions:\n  hello: world\n")
+
+    mock_clone_or_update.return_value = (repo_dir, None)
+    mock_resolve_symlink_stub.return_value = target
+
+    config: dict[str, Any] = {
+        "packages": {
+            "test_package": {
+                "url": "https://github.com/esphome/repo1",
+                "ref": "main",
+                "files": ["file1.yaml"],
+            }
+        }
+    }
+
+    # Must succeed (does not raise the helpful cv.Invalid) because the stub
+    # was followed and a valid mapping was loaded from the target.
+    do_packages_pass(config)
+    assert mock_resolve_symlink_stub.called
+
+
+@patch("esphome.git.clone_or_update")
+def test_remote_package_scalar_yaml_raises_helpful_error(
+    mock_clone_or_update: MagicMock, tmp_path: Path
+) -> None:
+    """A remote package YAML that is a top-level scalar (e.g. an unmaterialized
+    git symlink on Windows) raises a clear cv.Invalid, not AttributeError.
+
+    Regression test for the case where a repo containing a YAML symlink,
+    checked out on Windows without symlink privilege, lands as a short text
+    file containing the symlink target path. PyYAML parses that as a bare
+    string scalar; the package loader must reject it with a human-readable
+    error instead of dying inside ``.get()``.
+    """
+    CORE.config_path = tmp_path / "test.yaml"
+
+    repo_dir = tmp_path / "repo"
+    repo_dir.mkdir()
+    # Simulate the broken-symlink state: a YAML file whose entire content is
+    # the symlink target string. PyYAML parses this as a top-level scalar.
+    (repo_dir / "file1.yaml").write_text("static/file1.yaml")
+
+    mock_clone_or_update.return_value = (repo_dir, None)
+
+    config: dict[str, Any] = {
+        "packages": {
+            "test_package": {
+                "url": "https://github.com/esphome/repo1",
+                "ref": "main",
+                "files": ["file1.yaml"],
+            }
+        }
+    }
+
+    with pytest.raises(cv.Invalid) as exc_info:
+        do_packages_pass(config)
+
+    msg = str(exc_info.value)
+    assert "mapping at the top level" in msg
+    assert "file1.yaml" in msg

tests/unit_tests/test_writer.py

@@ -1361,7 +1361,7 @@ def test_clean_build_handles_readonly_files(
     # Create a read-only file (simulating git pack files on Windows)
     readonly_file = git_dir / "pack-abc123.pack"
     readonly_file.write_text("pack data")
-    os.chmod(readonly_file, stat.S_IRUSR)  # Read-only
+    readonly_file.chmod(stat.S_IRUSR)  # Read-only
 
     # Setup mocks
     mock_core.relative_pioenvs_path.return_value = pioenvs_dir
@@ -1396,7 +1396,7 @@ def test_clean_all_handles_readonly_files(
     subdir.mkdir()
     readonly_file = subdir / "readonly.txt"
     readonly_file.write_text("content")
-    os.chmod(readonly_file, stat.S_IRUSR)  # Read-only
+    readonly_file.chmod(stat.S_IRUSR)  # Read-only
 
     # Verify file is read-only
     assert not os.access(readonly_file, os.W_OK)
@@ -1425,7 +1425,7 @@ def test_clean_build_reraises_for_other_errors(
     test_file.write_text("content")
 
     # Make subdir read-only so files inside can't be deleted
-    os.chmod(subdir, stat.S_IRUSR | stat.S_IXUSR)
+    subdir.chmod(stat.S_IRUSR | stat.S_IXUSR)
 
     # Setup mocks
     mock_core.relative_pioenvs_path.return_value = pioenvs_dir
@@ -1443,7 +1443,7 @@ def test_clean_build_reraises_for_other_errors(
             clean_build()
     finally:
         # Cleanup - restore write permission so tmp_path cleanup works
-        os.chmod(subdir, stat.S_IRWXU)
+        subdir.chmod(stat.S_IRWXU)
 
 
 # Tests for get_build_info()

tests/unit_tests/test_yaml_util.py

@@ -15,6 +15,7 @@ from esphome.yaml_util import (
     DiscoveredYamlFiles,
     ESPHomeDataBase,
     ESPLiteralValue,
+    SensitiveStr,
     discover_user_yaml_files,
     force_load_include_files,
     format_path,
@@ -907,7 +908,7 @@ def test_format_path_current_obj_without_location_falls_back_to_key():
     """An ESPHomeDataBase current_obj with no esp_range falls back to the key's location."""
 
     class _NoRange(ESPHomeDataBase, str):
-        pass
+        __slots__ = ()
 
     obj = _NoRange.__new__(_NoRange, "value")
     str.__init__(obj)
@@ -1340,3 +1341,57 @@ def test_frontmatter_included_file_stored(tmp_path: Path) -> None:
     assert main.resolve() not in core.CORE.frontmatter
     # Included file's frontmatter is captured
     assert core.CORE.frontmatter[inc.resolve()]["child_meta"] == "hello"
+
+
+def test_sensitive_str__is_a_str_subclass() -> None:
+    value = SensitiveStr("hunter2")
+    assert isinstance(value, str)
+    assert value == "hunter2"
+
+
+def test_dump__redacts_sensitive_str_by_default() -> None:
+    out = yaml_util.dump({"password": SensitiveStr("hunter2")})
+    assert "\\033[8mhunter2\\033[28m" in out
+    assert "hunter2" not in out.replace(
+        "\\033[8mhunter2\\033[28m", ""
+    )  # the raw value is only present inside the wrap
+
+
+def test_dump__show_secrets_emits_sensitive_str_raw() -> None:
+    out = yaml_util.dump({"password": SensitiveStr("hunter2")}, show_secrets=True)
+    assert "hunter2" in out
+    assert "\\033[8m" not in out
+    assert "\\033[28m" not in out
+
+
+def test_dump__plain_str_is_not_redacted() -> None:
+    out = yaml_util.dump({"hostname": "myserver"})
+    assert "myserver" in out
+    assert "\\033[8m" not in out
+
+
+def test_dump__secret_reference_wins_over_redaction() -> None:
+    # If the value also has an entry in _SECRET_VALUES (i.e., it was loaded
+    # via !secret), the dump should render it as !secret <name>, not as a
+    # redacted scalar. SensitiveStr layered on top must not change that.
+    value = SensitiveStr("hunter2")
+    yaml_util._SECRET_VALUES[str(value)] = "my_secret_name"
+    try:
+        out = yaml_util.dump({"password": value})
+        assert "!secret" in out
+        assert "my_secret_name" in out
+        assert "\\033[8m" not in out
+    finally:
+        yaml_util._SECRET_VALUES.clear()
+
+
+def test_dump__redaction_flag_does_not_leak_between_calls() -> None:
+    # Per-call _Dumper subclass means show_secrets in one call doesn't
+    # affect another. Run them in both orders to catch any leakage.
+    redacted = yaml_util.dump({"password": SensitiveStr("hunter2")})
+    raw = yaml_util.dump({"password": SensitiveStr("hunter2")}, show_secrets=True)
+    redacted_again = yaml_util.dump({"password": SensitiveStr("hunter2")})
+
+    assert "\\033[8m" in redacted
+    assert "\\033[8m" not in raw
+    assert "\\033[8m" in redacted_again