diff --git a/esphome/__main__.py b/esphome/__main__.py index b63720d672..aa237c83a7 100644 --- a/esphome/__main__.py +++ b/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 diff --git a/esphome/components/external_components/__init__.py b/esphome/components/external_components/__init__.py index a09217fd21..ceb402c5b7 100644 --- a/esphome/components/external_components/__init__.py +++ b/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) diff --git a/esphome/components/packages/__init__.py b/esphome/components/packages/__init__.py index 2e7dc0e197..fdc75d995a 100644 --- a/esphome/components/packages/__init__.py +++ b/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" diff --git a/esphome/config.py b/esphome/config.py index 90325cbf6e..36892fcd25 100644 --- a/esphome/config.py +++ b/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 diff --git a/esphome/git.py b/esphome/git.py index 56aedd1519..62fe37a3fe 100644 --- a/esphome/git.py +++ b/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(): diff --git a/tests/component_tests/conftest.py b/tests/component_tests/conftest.py index 79b4151c55..0641e698e9 100644 --- a/tests/component_tests/conftest.py +++ b/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 diff --git a/tests/component_tests/external_components/test_init.py b/tests/component_tests/external_components/test_init.py new file mode 100644 index 0000000000..905c0afa8b --- /dev/null +++ b/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) diff --git a/tests/component_tests/packages/test_init.py b/tests/component_tests/packages/test_init.py new file mode 100644 index 0000000000..779244e2ed --- /dev/null +++ b/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) diff --git a/tests/unit_tests/conftest.py b/tests/unit_tests/conftest.py index 3f78acd10d..e8d9c02524 100644 --- a/tests/unit_tests/conftest.py +++ b/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.""" diff --git a/tests/unit_tests/test_git.py b/tests/unit_tests/test_git.py new file mode 100644 index 0000000000..6a51206ec2 --- /dev/null +++ b/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