Skip external component updates when running logs command (#10756)

Co-authored-by: Jesse Hills <3060199+jesserockz@users.noreply.github.com>
Co-authored-by: pre-commit-ci-lite[bot] <117423508+pre-commit-ci-lite[bot]@users.noreply.github.com>

esphome/__main__.py

@@ -1273,7 +1273,12 @@ def run_esphome(argv):
         CORE.config_path = conf_path
         CORE.dashboard = args.dashboard
 
-        config = read_config(dict(args.substitution) if args.substitution else {})
+        # For logs command, skip updating external components
+        skip_external = args.command == "logs"
+        config = read_config(
+            dict(args.substitution) if args.substitution else {},
+            skip_external_update=skip_external,
+        )
         if config is None:
             return 2
         CORE.config = config

esphome/components/external_components/__init__.py

@@ -39,11 +39,13 @@ async def to_code(config):
     pass
 
 
-def _process_git_config(config: dict, refresh) -> str:
+def _process_git_config(config: dict, refresh, skip_update: bool = False) -> str:
+    # When skip_update is True, use NEVER_REFRESH to prevent updates
+    actual_refresh = git.NEVER_REFRESH if skip_update else refresh
     repo_dir, _ = git.clone_or_update(
         url=config[CONF_URL],
         ref=config.get(CONF_REF),
-        refresh=refresh,
+        refresh=actual_refresh,
         domain=DOMAIN,
         username=config.get(CONF_USERNAME),
         password=config.get(CONF_PASSWORD),
@@ -70,12 +72,12 @@ def _process_git_config(config: dict, refresh) -> str:
     return components_dir
 
 
-def _process_single_config(config: dict):
+def _process_single_config(config: dict, skip_update: bool = False):
     conf = config[CONF_SOURCE]
     if conf[CONF_TYPE] == TYPE_GIT:
         with cv.prepend_path([CONF_SOURCE]):
             components_dir = _process_git_config(
-                config[CONF_SOURCE], config[CONF_REFRESH]
+                config[CONF_SOURCE], config[CONF_REFRESH], skip_update
             )
     elif conf[CONF_TYPE] == TYPE_LOCAL:
         components_dir = Path(CORE.relative_config_path(conf[CONF_PATH]))
@@ -105,7 +107,7 @@ def _process_single_config(config: dict):
     loader.install_meta_finder(components_dir, allowed_components=allowed_components)
 
 
-def do_external_components_pass(config: dict) -> None:
+def do_external_components_pass(config: dict, skip_update: bool = False) -> None:
     conf = config.get(DOMAIN)
     if conf is None:
         return
@@ -113,4 +115,4 @@ def do_external_components_pass(config: dict) -> None:
         conf = CONFIG_SCHEMA(conf)
         for i, c in enumerate(conf):
             with cv.prepend_path(i):
-                _process_single_config(c)
+                _process_single_config(c, skip_update)

esphome/components/packages/__init__.py

@@ -106,11 +106,13 @@ CONFIG_SCHEMA = cv.Any(
 )
 
 
-def _process_base_package(config: dict) -> dict:
+def _process_base_package(config: dict, skip_update: bool = False) -> dict:
+    # When skip_update is True, use NEVER_REFRESH to prevent updates
+    actual_refresh = git.NEVER_REFRESH if skip_update else config[CONF_REFRESH]
     repo_dir, revert = git.clone_or_update(
         url=config[CONF_URL],
         ref=config.get(CONF_REF),
-        refresh=config[CONF_REFRESH],
+        refresh=actual_refresh,
         domain=DOMAIN,
         username=config.get(CONF_USERNAME),
         password=config.get(CONF_PASSWORD),
@@ -180,16 +182,16 @@ def _process_base_package(config: dict) -> dict:
     return {"packages": packages}
 
 
-def _process_package(package_config, config):
+def _process_package(package_config, config, skip_update: bool = False):
     recursive_package = package_config
     if CONF_URL in package_config:
-        package_config = _process_base_package(package_config)
+        package_config = _process_base_package(package_config, skip_update)
     if isinstance(package_config, dict):
-        recursive_package = do_packages_pass(package_config)
+        recursive_package = do_packages_pass(package_config, skip_update)
     return merge_config(recursive_package, config)
 
 
-def do_packages_pass(config: dict):
+def do_packages_pass(config: dict, skip_update: bool = False):
     if CONF_PACKAGES not in config:
         return config
     packages = config[CONF_PACKAGES]
@@ -198,10 +200,10 @@ def do_packages_pass(config: dict):
         if isinstance(packages, dict):
             for package_name, package_config in reversed(packages.items()):
                 with cv.prepend_path(package_name):
-                    config = _process_package(package_config, config)
+                    config = _process_package(package_config, config, skip_update)
         elif isinstance(packages, list):
             for package_config in reversed(packages):
-                config = _process_package(package_config, config)
+                config = _process_package(package_config, config, skip_update)
         else:
             raise cv.Invalid(
                 f"Packages must be a key to value mapping or list, got {type(packages)} instead"

esphome/config.py

@@ -846,7 +846,9 @@ class PinUseValidationCheck(ConfigValidationStep):
 
 
 def validate_config(
-    config: dict[str, Any], command_line_substitutions: dict[str, Any]
+    config: dict[str, Any],
+    command_line_substitutions: dict[str, Any],
+    skip_external_update: bool = False,
 ) -> Config:
     result = Config()
 
@@ -859,7 +861,7 @@ def validate_config(
 
         result.add_output_path([CONF_PACKAGES], CONF_PACKAGES)
         try:
-            config = do_packages_pass(config)
+            config = do_packages_pass(config, skip_update=skip_external_update)
         except vol.Invalid as err:
             result.update(config)
             result.add_error(err)
@@ -896,7 +898,7 @@ def validate_config(
 
         result.add_output_path([CONF_EXTERNAL_COMPONENTS], CONF_EXTERNAL_COMPONENTS)
         try:
-            do_external_components_pass(config)
+            do_external_components_pass(config, skip_update=skip_external_update)
         except vol.Invalid as err:
             result.update(config)
             result.add_error(err)
@@ -1020,7 +1022,9 @@ class InvalidYAMLError(EsphomeError):
         self.base_exc = base_exc
 
 
-def _load_config(command_line_substitutions: dict[str, Any]) -> Config:
+def _load_config(
+    command_line_substitutions: dict[str, Any], skip_external_update: bool = False
+) -> Config:
     """Load the configuration file."""
     try:
         config = yaml_util.load_yaml(CORE.config_path)
@@ -1028,7 +1032,7 @@ def _load_config(command_line_substitutions: dict[str, Any]) -> Config:
         raise InvalidYAMLError(e) from e
 
     try:
-        return validate_config(config, command_line_substitutions)
+        return validate_config(config, command_line_substitutions, skip_external_update)
     except EsphomeError:
         raise
     except Exception:
@@ -1036,9 +1040,11 @@ def _load_config(command_line_substitutions: dict[str, Any]) -> Config:
         raise
 
 
-def load_config(command_line_substitutions: dict[str, Any]) -> Config:
+def load_config(
+    command_line_substitutions: dict[str, Any], skip_external_update: bool = False
+) -> Config:
     try:
-        return _load_config(command_line_substitutions)
+        return _load_config(command_line_substitutions, skip_external_update)
     except vol.Invalid as err:
         raise EsphomeError(f"Error while parsing config: {err}") from err
 
@@ -1178,10 +1184,10 @@ def strip_default_ids(config):
     return config
 
 
-def read_config(command_line_substitutions):
+def read_config(command_line_substitutions, skip_external_update=False):
     _LOGGER.info("Reading configuration %s...", CORE.config_path)
     try:
-        res = load_config(command_line_substitutions)
+        res = load_config(command_line_substitutions, skip_external_update)
     except EsphomeError as err:
         _LOGGER.error("Error while reading config: %s", err)
         return None

esphome/git.py

@@ -13,6 +13,9 @@ from esphome.core import CORE, TimePeriodSeconds
 
 _LOGGER = logging.getLogger(__name__)
 
+# Special value to indicate never refresh
+NEVER_REFRESH = TimePeriodSeconds(seconds=-1)
+
 
 def run_git_command(cmd, cwd=None) -> str:
     _LOGGER.debug("Running git command: %s", " ".join(cmd))
@@ -85,6 +88,11 @@ def clone_or_update(
 
     else:
         # Check refresh needed
+        # Skip refresh if NEVER_REFRESH is specified
+        if refresh == NEVER_REFRESH:
+            _LOGGER.debug("Skipping update for %s (refresh disabled)", key)
+            return repo_dir, None
+
         file_timestamp = Path(repo_dir / ".git" / "FETCH_HEAD")
         # On first clone, FETCH_HEAD does not exists
         if not file_timestamp.exists():

tests/component_tests/conftest.py

@@ -6,6 +6,7 @@ from collections.abc import Callable, Generator
 from pathlib import Path
 import sys
 from typing import Any
+from unittest import mock
 
 import pytest
 
@@ -17,6 +18,7 @@ from esphome.const import (
     PlatformFramework,
 )
 from esphome.types import ConfigType
+from esphome.util import OrderedDict
 
 # Add package root to python path
 here = Path(__file__).parent
@@ -135,3 +137,29 @@ def generate_main() -> Generator[Callable[[str | Path], str]]:
         return CORE.cpp_main_section
 
     yield generator
+
+
+@pytest.fixture
+def mock_clone_or_update() -> Generator[Any]:
+    """Mock git.clone_or_update for testing."""
+    with mock.patch("esphome.git.clone_or_update") as mock_func:
+        # Default return value
+        mock_func.return_value = (Path("/tmp/test"), None)
+        yield mock_func
+
+
+@pytest.fixture
+def mock_load_yaml() -> Generator[Any]:
+    """Mock yaml_util.load_yaml for testing."""
+
+    with mock.patch("esphome.yaml_util.load_yaml") as mock_func:
+        # Default return value
+        mock_func.return_value = OrderedDict({"sensor": []})
+        yield mock_func
+
+
+@pytest.fixture
+def mock_install_meta_finder() -> Generator[Any]:
+    """Mock loader.install_meta_finder for testing."""
+    with mock.patch("esphome.loader.install_meta_finder") as mock_func:
+        yield mock_func

tests/component_tests/external_components/test_init.py

@@ -0,0 +1,134 @@
+"""Tests for the external_components skip_update functionality."""
+
+from pathlib import Path
+from typing import Any
+from unittest.mock import MagicMock
+
+from esphome.components.external_components import do_external_components_pass
+from esphome.const import (
+    CONF_EXTERNAL_COMPONENTS,
+    CONF_REFRESH,
+    CONF_SOURCE,
+    CONF_URL,
+    TYPE_GIT,
+)
+
+
+def test_external_components_skip_update_true(
+    tmp_path: Path, mock_clone_or_update: MagicMock, mock_install_meta_finder: MagicMock
+) -> None:
+    """Test that external components don't update when skip_update=True."""
+    # Create a components directory structure
+    components_dir = tmp_path / "components"
+    components_dir.mkdir()
+
+    # Create a test component
+    test_component_dir = components_dir / "test_component"
+    test_component_dir.mkdir()
+    (test_component_dir / "__init__.py").write_text("# Test component")
+
+    # Set up mock to return our tmp_path
+    mock_clone_or_update.return_value = (tmp_path, None)
+
+    config: dict[str, Any] = {
+        CONF_EXTERNAL_COMPONENTS: [
+            {
+                CONF_SOURCE: {
+                    "type": TYPE_GIT,
+                    CONF_URL: "https://github.com/test/components",
+                },
+                CONF_REFRESH: "1d",
+                "components": "all",
+            }
+        ]
+    }
+
+    # Call with skip_update=True
+    do_external_components_pass(config, skip_update=True)
+
+    # Verify clone_or_update was called with NEVER_REFRESH
+    mock_clone_or_update.assert_called_once()
+    call_args = mock_clone_or_update.call_args
+    from esphome import git
+
+    assert call_args.kwargs["refresh"] == git.NEVER_REFRESH
+
+
+def test_external_components_skip_update_false(
+    tmp_path: Path, mock_clone_or_update: MagicMock, mock_install_meta_finder: MagicMock
+) -> None:
+    """Test that external components update when skip_update=False."""
+    # Create a components directory structure
+    components_dir = tmp_path / "components"
+    components_dir.mkdir()
+
+    # Create a test component
+    test_component_dir = components_dir / "test_component"
+    test_component_dir.mkdir()
+    (test_component_dir / "__init__.py").write_text("# Test component")
+
+    # Set up mock to return our tmp_path
+    mock_clone_or_update.return_value = (tmp_path, None)
+
+    config: dict[str, Any] = {
+        CONF_EXTERNAL_COMPONENTS: [
+            {
+                CONF_SOURCE: {
+                    "type": TYPE_GIT,
+                    CONF_URL: "https://github.com/test/components",
+                },
+                CONF_REFRESH: "1d",
+                "components": "all",
+            }
+        ]
+    }
+
+    # Call with skip_update=False
+    do_external_components_pass(config, skip_update=False)
+
+    # Verify clone_or_update was called with actual refresh value
+    mock_clone_or_update.assert_called_once()
+    call_args = mock_clone_or_update.call_args
+    from esphome.core import TimePeriodSeconds
+
+    assert call_args.kwargs["refresh"] == TimePeriodSeconds(days=1)
+
+
+def test_external_components_default_no_skip(
+    tmp_path: Path, mock_clone_or_update: MagicMock, mock_install_meta_finder: MagicMock
+) -> None:
+    """Test that external components update by default when skip_update not specified."""
+    # Create a components directory structure
+    components_dir = tmp_path / "components"
+    components_dir.mkdir()
+
+    # Create a test component
+    test_component_dir = components_dir / "test_component"
+    test_component_dir.mkdir()
+    (test_component_dir / "__init__.py").write_text("# Test component")
+
+    # Set up mock to return our tmp_path
+    mock_clone_or_update.return_value = (tmp_path, None)
+
+    config: dict[str, Any] = {
+        CONF_EXTERNAL_COMPONENTS: [
+            {
+                CONF_SOURCE: {
+                    "type": TYPE_GIT,
+                    CONF_URL: "https://github.com/test/components",
+                },
+                CONF_REFRESH: "1d",
+                "components": "all",
+            }
+        ]
+    }
+
+    # Call without skip_update parameter
+    do_external_components_pass(config)
+
+    # Verify clone_or_update was called with actual refresh value
+    mock_clone_or_update.assert_called_once()
+    call_args = mock_clone_or_update.call_args
+    from esphome.core import TimePeriodSeconds
+
+    assert call_args.kwargs["refresh"] == TimePeriodSeconds(days=1)

tests/component_tests/packages/test_init.py

@@ -0,0 +1,114 @@
+"""Tests for the packages component skip_update functionality."""
+
+from pathlib import Path
+from typing import Any
+from unittest.mock import MagicMock
+
+from esphome.components.packages import do_packages_pass
+from esphome.const import CONF_FILES, CONF_PACKAGES, CONF_REFRESH, CONF_URL
+from esphome.util import OrderedDict
+
+
+def test_packages_skip_update_true(
+    tmp_path: Path, mock_clone_or_update: MagicMock, mock_load_yaml: MagicMock
+) -> None:
+    """Test that packages don't update when skip_update=True."""
+    # Set up mock to return our tmp_path
+    mock_clone_or_update.return_value = (tmp_path, None)
+
+    # Create the test yaml file
+    test_file = tmp_path / "test.yaml"
+    test_file.write_text("sensor: []")
+
+    # Set mock_load_yaml to return some valid config
+    mock_load_yaml.return_value = OrderedDict({"sensor": []})
+
+    config: dict[str, Any] = {
+        CONF_PACKAGES: {
+            "test_package": {
+                CONF_URL: "https://github.com/test/repo",
+                CONF_FILES: ["test.yaml"],
+                CONF_REFRESH: "1d",
+            }
+        }
+    }
+
+    # Call with skip_update=True
+    do_packages_pass(config, skip_update=True)
+
+    # Verify clone_or_update was called with NEVER_REFRESH
+    mock_clone_or_update.assert_called_once()
+    call_args = mock_clone_or_update.call_args
+    from esphome import git
+
+    assert call_args.kwargs["refresh"] == git.NEVER_REFRESH
+
+
+def test_packages_skip_update_false(
+    tmp_path: Path, mock_clone_or_update: MagicMock, mock_load_yaml: MagicMock
+) -> None:
+    """Test that packages update when skip_update=False."""
+    # Set up mock to return our tmp_path
+    mock_clone_or_update.return_value = (tmp_path, None)
+
+    # Create the test yaml file
+    test_file = tmp_path / "test.yaml"
+    test_file.write_text("sensor: []")
+
+    # Set mock_load_yaml to return some valid config
+    mock_load_yaml.return_value = OrderedDict({"sensor": []})
+
+    config: dict[str, Any] = {
+        CONF_PACKAGES: {
+            "test_package": {
+                CONF_URL: "https://github.com/test/repo",
+                CONF_FILES: ["test.yaml"],
+                CONF_REFRESH: "1d",
+            }
+        }
+    }
+
+    # Call with skip_update=False (default)
+    do_packages_pass(config, skip_update=False)
+
+    # Verify clone_or_update was called with actual refresh value
+    mock_clone_or_update.assert_called_once()
+    call_args = mock_clone_or_update.call_args
+    from esphome.core import TimePeriodSeconds
+
+    assert call_args.kwargs["refresh"] == TimePeriodSeconds(days=1)
+
+
+def test_packages_default_no_skip(
+    tmp_path: Path, mock_clone_or_update: MagicMock, mock_load_yaml: MagicMock
+) -> None:
+    """Test that packages update by default when skip_update not specified."""
+    # Set up mock to return our tmp_path
+    mock_clone_or_update.return_value = (tmp_path, None)
+
+    # Create the test yaml file
+    test_file = tmp_path / "test.yaml"
+    test_file.write_text("sensor: []")
+
+    # Set mock_load_yaml to return some valid config
+    mock_load_yaml.return_value = OrderedDict({"sensor": []})
+
+    config: dict[str, Any] = {
+        CONF_PACKAGES: {
+            "test_package": {
+                CONF_URL: "https://github.com/test/repo",
+                CONF_FILES: ["test.yaml"],
+                CONF_REFRESH: "1d",
+            }
+        }
+    }
+
+    # Call without skip_update parameter
+    do_packages_pass(config)
+
+    # Verify clone_or_update was called with actual refresh value
+    mock_clone_or_update.assert_called_once()
+    call_args = mock_clone_or_update.call_args
+    from esphome.core import TimePeriodSeconds
+
+    assert call_args.kwargs["refresh"] == TimePeriodSeconds(days=1)

tests/unit_tests/conftest.py

@@ -89,6 +89,13 @@ def mock_run_external_command() -> Generator[Mock, None, None]:
         yield mock
 
 
+@pytest.fixture
+def mock_run_git_command() -> Generator[Mock, None, None]:
+    """Mock run_git_command for git module."""
+    with patch("esphome.git.run_git_command") as mock:
+        yield mock
+
+
 @pytest.fixture
 def mock_get_idedata() -> Generator[Mock, None, None]:
     """Mock get_idedata for platformio_api."""

tests/unit_tests/test_git.py

@@ -0,0 +1,246 @@
+"""Tests for git.py module."""
+
+from datetime import datetime, timedelta
+import hashlib
+import os
+from pathlib import Path
+from unittest.mock import Mock
+
+from esphome import git
+from esphome.core import CORE, TimePeriodSeconds
+
+
+def test_clone_or_update_with_never_refresh(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """Test that NEVER_REFRESH skips updates for existing repos."""
+    # Set up CORE.config_path so data_dir uses tmp_path
+    CORE.config_path = tmp_path / "test.yaml"
+
+    # Compute the expected repo directory path
+    url = "https://github.com/test/repo"
+    ref = None
+    key = f"{url}@{ref}"
+    domain = "test"
+
+    # Compute hash-based directory name (matching _compute_destination_path logic)
+    h = hashlib.new("sha256")
+    h.update(key.encode())
+    repo_dir = tmp_path / ".esphome" / domain / h.hexdigest()[:8]
+
+    # Create the git repo directory structure
+    repo_dir.mkdir(parents=True)
+    git_dir = repo_dir / ".git"
+    git_dir.mkdir()
+
+    # Create FETCH_HEAD file with current timestamp
+    fetch_head = git_dir / "FETCH_HEAD"
+    fetch_head.write_text("test")
+
+    # Call with NEVER_REFRESH
+    result_dir, revert = git.clone_or_update(
+        url=url,
+        ref=ref,
+        refresh=git.NEVER_REFRESH,
+        domain=domain,
+    )
+
+    # Should NOT call git commands since NEVER_REFRESH and repo exists
+    mock_run_git_command.assert_not_called()
+    assert result_dir == repo_dir
+    assert revert is None
+
+
+def test_clone_or_update_with_refresh_updates_old_repo(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """Test that refresh triggers update for old repos."""
+    # Set up CORE.config_path so data_dir uses tmp_path
+    CORE.config_path = tmp_path / "test.yaml"
+
+    # Compute the expected repo directory path
+    url = "https://github.com/test/repo"
+    ref = None
+    key = f"{url}@{ref}"
+    domain = "test"
+
+    # Compute hash-based directory name (matching _compute_destination_path logic)
+    h = hashlib.new("sha256")
+    h.update(key.encode())
+    repo_dir = tmp_path / ".esphome" / domain / h.hexdigest()[:8]
+
+    # Create the git repo directory structure
+    repo_dir.mkdir(parents=True)
+    git_dir = repo_dir / ".git"
+    git_dir.mkdir()
+
+    # 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)
+    fetch_head.touch()  # Create the file
+    # Set modification time to 2 days ago
+    os.utime(fetch_head, (old_time.timestamp(), old_time.timestamp()))
+
+    # Mock git command responses
+    mock_run_git_command.return_value = "abc123"  # SHA for rev-parse
+
+    # Call with refresh=1d (1 day)
+    refresh = TimePeriodSeconds(days=1)
+    result_dir, revert = git.clone_or_update(
+        url=url,
+        ref=ref,
+        refresh=refresh,
+        domain=domain,
+    )
+
+    # Should call git fetch and update commands since repo is older than refresh
+    assert mock_run_git_command.called
+    # Check for fetch command
+    fetch_calls = [
+        call
+        for call in mock_run_git_command.call_args_list
+        if len(call[0]) > 0 and "fetch" in call[0][0]
+    ]
+    assert len(fetch_calls) > 0
+
+
+def test_clone_or_update_with_refresh_skips_fresh_repo(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """Test that refresh doesn't update fresh repos."""
+    # Set up CORE.config_path so data_dir uses tmp_path
+    CORE.config_path = tmp_path / "test.yaml"
+
+    # Compute the expected repo directory path
+    url = "https://github.com/test/repo"
+    ref = None
+    key = f"{url}@{ref}"
+    domain = "test"
+
+    # Compute hash-based directory name (matching _compute_destination_path logic)
+    h = hashlib.new("sha256")
+    h.update(key.encode())
+    repo_dir = tmp_path / ".esphome" / domain / h.hexdigest()[:8]
+
+    # Create the git repo directory structure
+    repo_dir.mkdir(parents=True)
+    git_dir = repo_dir / ".git"
+    git_dir.mkdir()
+
+    # 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)
+    fetch_head.touch()  # Create the file
+    # Set modification time to 1 hour ago
+    os.utime(fetch_head, (recent_time.timestamp(), recent_time.timestamp()))
+
+    # Call with refresh=1d (1 day)
+    refresh = TimePeriodSeconds(days=1)
+    result_dir, revert = git.clone_or_update(
+        url=url,
+        ref=ref,
+        refresh=refresh,
+        domain=domain,
+    )
+
+    # Should NOT call git fetch since repo is fresh
+    mock_run_git_command.assert_not_called()
+    assert result_dir == repo_dir
+    assert revert is None
+
+
+def test_clone_or_update_clones_missing_repo(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """Test that missing repos are cloned regardless of refresh setting."""
+    # Set up CORE.config_path so data_dir uses tmp_path
+    CORE.config_path = tmp_path / "test.yaml"
+
+    # Compute the expected repo directory path
+    url = "https://github.com/test/repo"
+    ref = None
+    key = f"{url}@{ref}"
+    domain = "test"
+
+    # Compute hash-based directory name (matching _compute_destination_path logic)
+    h = hashlib.new("sha256")
+    h.update(key.encode())
+    repo_dir = tmp_path / ".esphome" / domain / h.hexdigest()[:8]
+
+    # Create base directory but NOT the repo itself
+    base_dir = tmp_path / ".esphome" / domain
+    base_dir.mkdir(parents=True)
+    # repo_dir should NOT exist
+    assert not repo_dir.exists()
+
+    # Test with NEVER_REFRESH - should still clone since repo doesn't exist
+    result_dir, revert = git.clone_or_update(
+        url=url,
+        ref=ref,
+        refresh=git.NEVER_REFRESH,
+        domain=domain,
+    )
+
+    # Should call git clone
+    assert mock_run_git_command.called
+    clone_calls = [
+        call
+        for call in mock_run_git_command.call_args_list
+        if len(call[0]) > 0 and "clone" in call[0][0]
+    ]
+    assert len(clone_calls) > 0
+
+
+def test_clone_or_update_with_none_refresh_always_updates(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """Test that refresh=None always updates existing repos."""
+    # Set up CORE.config_path so data_dir uses tmp_path
+    CORE.config_path = tmp_path / "test.yaml"
+
+    # Compute the expected repo directory path
+    url = "https://github.com/test/repo"
+    ref = None
+    key = f"{url}@{ref}"
+    domain = "test"
+
+    # Compute hash-based directory name (matching _compute_destination_path logic)
+    h = hashlib.new("sha256")
+    h.update(key.encode())
+    repo_dir = tmp_path / ".esphome" / domain / h.hexdigest()[:8]
+
+    # Create the git repo directory structure
+    repo_dir.mkdir(parents=True)
+    git_dir = repo_dir / ".git"
+    git_dir.mkdir()
+
+    # 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)
+    fetch_head.touch()  # Create the file
+    # Set modification time to 1 second ago
+    os.utime(fetch_head, (recent_time.timestamp(), recent_time.timestamp()))
+
+    # Mock git command responses
+    mock_run_git_command.return_value = "abc123"  # SHA for rev-parse
+
+    # Call with refresh=None (default behavior)
+    result_dir, revert = git.clone_or_update(
+        url=url,
+        ref=ref,
+        refresh=None,
+        domain=domain,
+    )
+
+    # Should call git fetch and update commands since refresh=None means always update
+    assert mock_run_git_command.called
+    # Check for fetch command
+    fetch_calls = [
+        call
+        for call in mock_run_git_command.call_args_list
+        if len(call[0]) > 0 and "fetch" in call[0][0]
+    ]
+    assert len(fetch_calls) > 0