Merge branch 'dev' into sha256_ota

tests/unit_tests/test_main.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 from collections.abc import Generator
 from dataclasses import dataclass
+import logging
 from pathlib import Path
 import re
 from typing import Any
@@ -16,6 +17,7 @@ from esphome import platformio_api
 from esphome.__main__ import (
     Purpose,
     choose_upload_log_host,
+    command_clean_all,
     command_rename,
     command_update_all,
     command_wizard,
@@ -1853,3 +1855,95 @@ esp32:
     # Should not have any Python error messages
     assert "TypeError" not in clean_output
     assert "can only concatenate str" not in clean_output
+
+
+def test_command_clean_all_success(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test command_clean_all when writer.clean_all() succeeds."""
+    args = MockArgs(configuration=["/path/to/config1", "/path/to/config2"])
+
+    # Set logger level to capture INFO messages
+    with (
+        caplog.at_level(logging.INFO),
+        patch("esphome.writer.clean_all") as mock_clean_all,
+    ):
+        result = command_clean_all(args)
+
+        assert result == 0
+        mock_clean_all.assert_called_once_with(["/path/to/config1", "/path/to/config2"])
+
+        # Check that success message was logged
+        assert "Done!" in caplog.text
+
+
+def test_command_clean_all_oserror(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test command_clean_all when writer.clean_all() raises OSError."""
+    args = MockArgs(configuration=["/path/to/config1"])
+
+    # Create a mock OSError with a specific message
+    mock_error = OSError("Permission denied: cannot delete directory")
+
+    # Set logger level to capture ERROR and INFO messages
+    with (
+        caplog.at_level(logging.INFO),
+        patch("esphome.writer.clean_all", side_effect=mock_error) as mock_clean_all,
+    ):
+        result = command_clean_all(args)
+
+        assert result == 1
+        mock_clean_all.assert_called_once_with(["/path/to/config1"])
+
+        # Check that error message was logged
+        assert (
+            "Error cleaning all files: Permission denied: cannot delete directory"
+            in caplog.text
+        )
+        # Should not have success message
+        assert "Done!" not in caplog.text
+
+
+def test_command_clean_all_oserror_no_message(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test command_clean_all when writer.clean_all() raises OSError without message."""
+    args = MockArgs(configuration=["/path/to/config1"])
+
+    # Create a mock OSError without a message
+    mock_error = OSError()
+
+    # Set logger level to capture ERROR and INFO messages
+    with (
+        caplog.at_level(logging.INFO),
+        patch("esphome.writer.clean_all", side_effect=mock_error) as mock_clean_all,
+    ):
+        result = command_clean_all(args)
+
+        assert result == 1
+        mock_clean_all.assert_called_once_with(["/path/to/config1"])
+
+        # Check that error message was logged (should show empty string for OSError without message)
+        assert "Error cleaning all files:" in caplog.text
+        # Should not have success message
+        assert "Done!" not in caplog.text
+
+
+def test_command_clean_all_args_used() -> None:
+    """Test that command_clean_all uses args.configuration parameter."""
+    # Test with different configuration paths
+    args1 = MockArgs(configuration=["/path/to/config1"])
+    args2 = MockArgs(configuration=["/path/to/config2", "/path/to/config3"])
+
+    with patch("esphome.writer.clean_all") as mock_clean_all:
+        result1 = command_clean_all(args1)
+        result2 = command_clean_all(args2)
+
+        assert result1 == 0
+        assert result2 == 0
+        assert mock_clean_all.call_count == 2
+
+        # Verify the correct configuration paths were passed
+        mock_clean_all.assert_any_call(["/path/to/config1"])
+        mock_clean_all.assert_any_call(["/path/to/config2", "/path/to/config3"])

tests/unit_tests/test_writer.py

@@ -362,11 +362,17 @@ def test_clean_build(
     assert dependencies_lock.exists()
     assert platformio_cache_dir.exists()
 
-    # Mock PlatformIO's get_project_cache_dir
+    # Mock PlatformIO's ProjectConfig cache_dir
     with patch(
-        "platformio.project.helpers.get_project_cache_dir"
-    ) as mock_get_cache_dir:
-        mock_get_cache_dir.return_value = str(platformio_cache_dir)
+        "platformio.project.config.ProjectConfig.get_instance"
+    ) as mock_get_instance:
+        mock_config = MagicMock()
+        mock_get_instance.return_value = mock_config
+        mock_config.get.side_effect = (
+            lambda section, option: str(platformio_cache_dir)
+            if (section, option) == ("platformio", "cache_dir")
+            else ""
+        )
 
         # Call the function
         with caplog.at_level("INFO"):
@@ -486,7 +492,7 @@ def test_clean_build_platformio_not_available(
 
     # Mock import error for platformio
     with (
-        patch.dict("sys.modules", {"platformio.project.helpers": None}),
+        patch.dict("sys.modules", {"platformio.project.config": None}),
         caplog.at_level("INFO"),
     ):
         # Call the function
@@ -520,11 +526,17 @@ def test_clean_build_empty_cache_dir(
     # Verify pioenvs exists before
     assert pioenvs_dir.exists()
 
-    # Mock PlatformIO's get_project_cache_dir to return whitespace
+    # Mock PlatformIO's ProjectConfig cache_dir to return whitespace
     with patch(
-        "platformio.project.helpers.get_project_cache_dir"
-    ) as mock_get_cache_dir:
-        mock_get_cache_dir.return_value = "   "  # Whitespace only
+        "platformio.project.config.ProjectConfig.get_instance"
+    ) as mock_get_instance:
+        mock_config = MagicMock()
+        mock_get_instance.return_value = mock_config
+        mock_config.get.side_effect = (
+            lambda section, option: "   "  # Whitespace only
+            if (section, option) == ("platformio", "cache_dir")
+            else ""
+        )
 
         # Call the function
         with caplog.at_level("INFO"):
@@ -723,3 +735,135 @@ def test_write_cpp_with_duplicate_markers(
     # Call should raise an error
     with pytest.raises(EsphomeError, match="Found multiple auto generate code begins"):
         write_cpp("// New code")
+
+
+@patch("esphome.writer.CORE")
+def test_clean_all(
+    mock_core: MagicMock,
+    tmp_path: Path,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test clean_all removes build and PlatformIO dirs."""
+    # Create build directories for multiple configurations
+    config1_dir = tmp_path / "config1"
+    config2_dir = tmp_path / "config2"
+    config1_dir.mkdir()
+    config2_dir.mkdir()
+
+    build_dir1 = config1_dir / ".esphome"
+    build_dir2 = config2_dir / ".esphome"
+    build_dir1.mkdir()
+    build_dir2.mkdir()
+    (build_dir1 / "dummy.txt").write_text("x")
+    (build_dir2 / "dummy.txt").write_text("x")
+
+    # Create PlatformIO directories
+    pio_cache = tmp_path / "pio_cache"
+    pio_packages = tmp_path / "pio_packages"
+    pio_platforms = tmp_path / "pio_platforms"
+    pio_core = tmp_path / "pio_core"
+    for d in (pio_cache, pio_packages, pio_platforms, pio_core):
+        d.mkdir()
+        (d / "keep").write_text("x")
+
+    # Mock ProjectConfig
+    with patch(
+        "platformio.project.config.ProjectConfig.get_instance"
+    ) as mock_get_instance:
+        mock_config = MagicMock()
+        mock_get_instance.return_value = mock_config
+
+        def cfg_get(section: str, option: str) -> str:
+            mapping = {
+                ("platformio", "cache_dir"): str(pio_cache),
+                ("platformio", "packages_dir"): str(pio_packages),
+                ("platformio", "platforms_dir"): str(pio_platforms),
+                ("platformio", "core_dir"): str(pio_core),
+            }
+            return mapping.get((section, option), "")
+
+        mock_config.get.side_effect = cfg_get
+
+        # Call
+        from esphome.writer import clean_all
+
+        with caplog.at_level("INFO"):
+            clean_all([str(config1_dir), str(config2_dir)])
+
+    # Verify deletions
+    assert not build_dir1.exists()
+    assert not build_dir2.exists()
+    assert not pio_cache.exists()
+    assert not pio_packages.exists()
+    assert not pio_platforms.exists()
+    assert not pio_core.exists()
+
+    # Verify logging mentions each
+    assert "Deleting" in caplog.text
+    assert str(build_dir1) in caplog.text
+    assert str(build_dir2) in caplog.text
+    assert "PlatformIO cache" in caplog.text
+    assert "PlatformIO packages" in caplog.text
+    assert "PlatformIO platforms" in caplog.text
+    assert "PlatformIO core" in caplog.text
+
+
+@patch("esphome.writer.CORE")
+def test_clean_all_platformio_not_available(
+    mock_core: MagicMock,
+    tmp_path: Path,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test clean_all when PlatformIO is not available."""
+    # Build dirs
+    config_dir = tmp_path / "config"
+    config_dir.mkdir()
+    build_dir = config_dir / ".esphome"
+    build_dir.mkdir()
+
+    # PlatformIO dirs that should remain untouched
+    pio_cache = tmp_path / "pio_cache"
+    pio_cache.mkdir()
+
+    from esphome.writer import clean_all
+
+    with (
+        patch.dict("sys.modules", {"platformio.project.config": None}),
+        caplog.at_level("INFO"),
+    ):
+        clean_all([str(config_dir)])
+
+    # Build dir removed, PlatformIO dirs remain
+    assert not build_dir.exists()
+    assert pio_cache.exists()
+
+    # No PlatformIO-specific logs
+    assert "PlatformIO" not in caplog.text
+
+
+@patch("esphome.writer.CORE")
+def test_clean_all_partial_exists(
+    mock_core: MagicMock,
+    tmp_path: Path,
+) -> None:
+    """Test clean_all when only some build dirs exist."""
+    config_dir = tmp_path / "config"
+    config_dir.mkdir()
+    build_dir = config_dir / ".esphome"
+    build_dir.mkdir()
+
+    with patch(
+        "platformio.project.config.ProjectConfig.get_instance"
+    ) as mock_get_instance:
+        mock_config = MagicMock()
+        mock_get_instance.return_value = mock_config
+        # Return non-existent dirs
+        mock_config.get.side_effect = lambda *_args, **_kw: str(
+            tmp_path / "does_not_exist"
+        )
+
+        from esphome.writer import clean_all
+
+        clean_all([str(config_dir)])
+
+    assert not build_dir.exists()