diff --git a/esphome/__main__.py b/esphome/__main__.py index 0147a82530..885eaafe15 100644 --- a/esphome/__main__.py +++ b/esphome/__main__.py @@ -1244,7 +1244,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..5362a2269f 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, set refresh to None to prevent updates + actual_refresh = None 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..2f964984cc 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, set refresh to None to prevent updates + actual_refresh = None 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..c60b928d7c 100644 --- a/esphome/git.py +++ b/esphome/git.py @@ -90,7 +90,7 @@ def clone_or_update( if not file_timestamp.exists(): file_timestamp = Path(repo_dir / ".git" / "HEAD") age = datetime.now() - datetime.fromtimestamp(file_timestamp.stat().st_mtime) - if refresh is None or age.total_seconds() > refresh.total_seconds: + if refresh is not None and age.total_seconds() > refresh.total_seconds: old_sha = run_git_command(["git", "rev-parse", "HEAD"], str(repo_dir)) _LOGGER.info("Updating %s", key) _LOGGER.debug("Location: %s", repo_dir) 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..bdec13fe0f --- /dev/null +++ b/tests/component_tests/external_components/test_init.py @@ -0,0 +1,113 @@ +"""Tests for the external_components skip_update functionality.""" + +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, patch + +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, +) + + +@patch("esphome.git.clone_or_update") +@patch("esphome.loader.install_meta_finder") +def test_external_components_skip_update_true( + mock_install_meta: MagicMock, mock_clone_or_update: MagicMock +) -> None: + """Test that external components don't update when skip_update=True.""" + # Setup mocks + test_path = Path("/tmp/test/components") + test_path.mkdir(parents=True, exist_ok=True) + mock_clone_or_update.return_value = (test_path.parent, 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 refresh=None + mock_clone_or_update.assert_called_once() + call_args = mock_clone_or_update.call_args + assert call_args.kwargs["refresh"] is None + + +@patch("esphome.git.clone_or_update") +@patch("esphome.loader.install_meta_finder") +def test_external_components_skip_update_false( + mock_install_meta: MagicMock, mock_clone_or_update: MagicMock +) -> None: + """Test that external components update when skip_update=False.""" + # Setup mocks + test_path = Path("/tmp/test/components") + test_path.mkdir(parents=True, exist_ok=True) + mock_clone_or_update.return_value = (test_path.parent, 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 + assert call_args.kwargs["refresh"] == "1d" + + +@patch("esphome.git.clone_or_update") +@patch("esphome.loader.install_meta_finder") +def test_external_components_default_no_skip( + mock_install_meta: MagicMock, mock_clone_or_update: MagicMock +) -> None: + """Test that external components update by default when skip_update not specified.""" + # Setup mocks + test_path = Path("/tmp/test/components") + test_path.mkdir(parents=True, exist_ok=True) + mock_clone_or_update.return_value = (test_path.parent, 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 + assert call_args.kwargs["refresh"] == "1d" diff --git a/tests/component_tests/packages/test_init.py b/tests/component_tests/packages/test_init.py new file mode 100644 index 0000000000..fbf12829ef --- /dev/null +++ b/tests/component_tests/packages/test_init.py @@ -0,0 +1,102 @@ +"""Tests for the packages component skip_update functionality.""" + +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, patch + +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 + + +@patch("esphome.git.clone_or_update") +@patch("esphome.yaml_util.load_yaml") +@patch("pathlib.Path.is_file") +def test_packages_skip_update_true( + mock_is_file: MagicMock, mock_load_yaml: MagicMock, mock_clone_or_update: MagicMock +) -> None: + """Test that packages don't update when skip_update=True.""" + # Setup mocks + mock_clone_or_update.return_value = (Path("/tmp/test"), None) + mock_is_file.return_value = True + 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 refresh=None + mock_clone_or_update.assert_called_once() + call_args = mock_clone_or_update.call_args + assert call_args.kwargs["refresh"] is None + + +@patch("esphome.git.clone_or_update") +@patch("esphome.yaml_util.load_yaml") +@patch("pathlib.Path.is_file") +def test_packages_skip_update_false( + mock_is_file: MagicMock, mock_load_yaml: MagicMock, mock_clone_or_update: MagicMock +) -> None: + """Test that packages update when skip_update=False.""" + # Setup mocks + mock_clone_or_update.return_value = (Path("/tmp/test"), None) + mock_is_file.return_value = True + 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 + assert call_args.kwargs["refresh"] == "1d" + + +@patch("esphome.git.clone_or_update") +@patch("esphome.yaml_util.load_yaml") +@patch("pathlib.Path.is_file") +def test_packages_default_no_skip( + mock_is_file: MagicMock, mock_load_yaml: MagicMock, mock_clone_or_update: MagicMock +) -> None: + """Test that packages update by default when skip_update not specified.""" + # Setup mocks + mock_clone_or_update.return_value = (Path("/tmp/test"), None) + mock_is_file.return_value = True + 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 + assert call_args.kwargs["refresh"] == "1d" diff --git a/tests/component_tests/packages/test_packages.py b/tests/component_tests/packages/test_packages.py index 4712daad0d..99ed661649 100644 --- a/tests/component_tests/packages/test_packages.py +++ b/tests/component_tests/packages/test_packages.py @@ -732,3 +732,96 @@ def test_remote_packages_with_files_and_vars( actual = do_packages_pass(config) assert actual == expected + + +@patch("esphome.git.clone_or_update") +@patch("esphome.yaml_util.load_yaml") +@patch("pathlib.Path.is_file") +def test_packages_skip_update_true( + mock_is_file: MagicMock, mock_load_yaml: MagicMock, mock_clone_or_update: MagicMock +) -> None: + """Test that packages don't update when skip_update=True.""" + # Setup mocks + mock_clone_or_update.return_value = (Path("/tmp/test"), None) + mock_is_file.return_value = True + mock_load_yaml.return_value = OrderedDict({"sensor": []}) + + config = { + 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 refresh=None + mock_clone_or_update.assert_called_once() + call_args = mock_clone_or_update.call_args + assert call_args.kwargs["refresh"] is None + + +@patch("esphome.git.clone_or_update") +@patch("esphome.yaml_util.load_yaml") +@patch("pathlib.Path.is_file") +def test_packages_skip_update_false( + mock_is_file: MagicMock, mock_load_yaml: MagicMock, mock_clone_or_update: MagicMock +) -> None: + """Test that packages update when skip_update=False.""" + # Setup mocks + mock_clone_or_update.return_value = (Path("/tmp/test"), None) + mock_is_file.return_value = True + mock_load_yaml.return_value = OrderedDict({"sensor": []}) + + config = { + 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 + assert call_args.kwargs["refresh"] == "1d" + + +@patch("esphome.git.clone_or_update") +@patch("esphome.yaml_util.load_yaml") +@patch("pathlib.Path.is_file") +def test_packages_default_no_skip( + mock_is_file: MagicMock, mock_load_yaml: MagicMock, mock_clone_or_update: MagicMock +) -> None: + """Test that packages update by default when skip_update not specified.""" + # Setup mocks + mock_clone_or_update.return_value = (Path("/tmp/test"), None) + mock_is_file.return_value = True + mock_load_yaml.return_value = OrderedDict({"sensor": []}) + + config = { + 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 + assert call_args.kwargs["refresh"] == "1d" diff --git a/tests/unit_tests/core/test_config.py b/tests/unit_tests/core/test_config.py index 7d3b90794b..921863e2bc 100644 --- a/tests/unit_tests/core/test_config.py +++ b/tests/unit_tests/core/test_config.py @@ -852,3 +852,118 @@ async def test_add_includes_overwrites_existing_files( mock_copy_file_if_changed.assert_called_once_with( str(include_file), str(Path(CORE.build_path) / "src" / "header.h") ) + + +# Tests for skip_external_update functionality + + +@patch("esphome.yaml_util.load_yaml") +@patch("esphome.components.packages.do_packages_pass") +@patch("esphome.components.external_components.do_external_components_pass") +def test_validate_config_skip_update_true( + mock_ext_pass: MagicMock, mock_pkg_pass: MagicMock, mock_load_yaml: MagicMock +) -> None: + """Test that validate_config propagates skip_update=True.""" + from esphome.config import validate_config + from esphome.const import CONF_EXTERNAL_COMPONENTS, CONF_PACKAGES + + config_dict: dict[str, Any] = { + CONF_ESPHOME: {CONF_NAME: "test"}, + CONF_PACKAGES: {"test": {}}, + CONF_EXTERNAL_COMPONENTS: [{}], + } + + # Mock do_packages_pass to return config unchanged + mock_pkg_pass.side_effect = lambda c, **kwargs: c + + # Call validate_config with skip_external_update=True + validate_config(config_dict, {}, skip_external_update=True) + + # Verify both were called with skip_update=True + mock_pkg_pass.assert_called_once() + assert mock_pkg_pass.call_args.kwargs.get("skip_update") is True + + mock_ext_pass.assert_called_once() + assert mock_ext_pass.call_args.kwargs.get("skip_update") is True + + +@patch("esphome.yaml_util.load_yaml") +@patch("esphome.components.packages.do_packages_pass") +@patch("esphome.components.external_components.do_external_components_pass") +def test_validate_config_skip_update_false( + mock_ext_pass: MagicMock, mock_pkg_pass: MagicMock, mock_load_yaml: MagicMock +) -> None: + """Test that validate_config propagates skip_update=False.""" + from esphome.config import validate_config + from esphome.const import CONF_EXTERNAL_COMPONENTS, CONF_PACKAGES + + config_dict: dict[str, Any] = { + CONF_ESPHOME: {CONF_NAME: "test"}, + CONF_PACKAGES: {"test": {}}, + CONF_EXTERNAL_COMPONENTS: [{}], + } + + # Mock do_packages_pass to return config unchanged + mock_pkg_pass.side_effect = lambda c, **kwargs: c + + # Call validate_config with skip_external_update=False + validate_config(config_dict, {}, skip_external_update=False) + + # Verify both were called with skip_update=False + mock_pkg_pass.assert_called_once() + assert mock_pkg_pass.call_args.kwargs.get("skip_update") is False + + mock_ext_pass.assert_called_once() + assert mock_ext_pass.call_args.kwargs.get("skip_update") is False + + +@patch("esphome.yaml_util.load_yaml") +@patch("esphome.components.packages.do_packages_pass") +@patch("esphome.components.external_components.do_external_components_pass") +def test_validate_config_default_false( + mock_ext_pass: MagicMock, mock_pkg_pass: MagicMock, mock_load_yaml: MagicMock +) -> None: + """Test that validate_config defaults to skip_update=False.""" + from esphome.config import validate_config + from esphome.const import CONF_EXTERNAL_COMPONENTS, CONF_PACKAGES + + config_dict: dict[str, Any] = { + CONF_ESPHOME: {CONF_NAME: "test"}, + CONF_PACKAGES: {"test": {}}, + CONF_EXTERNAL_COMPONENTS: [{}], + } + + # Mock do_packages_pass to return config unchanged + mock_pkg_pass.side_effect = lambda c, **kwargs: c + + # Call validate_config without skip_external_update parameter + validate_config(config_dict, {}) + + # Verify both were called with skip_update=False (default) + mock_pkg_pass.assert_called_once() + assert mock_pkg_pass.call_args.kwargs.get("skip_update") is False + + mock_ext_pass.assert_called_once() + assert mock_ext_pass.call_args.kwargs.get("skip_update") is False + + +@patch("esphome.config.load_config") +def test_read_config_skip_update_parameter(mock_load_config: MagicMock) -> None: + """Test that read_config passes skip_external_update correctly.""" + from esphome.config import read_config + + # Setup + CORE.config_path = "test.yaml" + mock_load_config.return_value = MagicMock(errors=[]) + + # Test with skip_external_update=True + read_config({}, skip_external_update=True) + mock_load_config.assert_called_with({}, True) + + # Test with skip_external_update=False + read_config({}, skip_external_update=False) + mock_load_config.assert_called_with({}, False) + + # Test default (should be False) + read_config({}) + mock_load_config.assert_called_with({}, False) diff --git a/tests/unit_tests/test_git.py b/tests/unit_tests/test_git.py new file mode 100644 index 0000000000..99b0f12441 --- /dev/null +++ b/tests/unit_tests/test_git.py @@ -0,0 +1,147 @@ +"""Tests for git.py module.""" + +from datetime import datetime, timedelta +from unittest.mock import MagicMock, Mock, patch + +from esphome import git +from esphome.core import TimePeriodSeconds + + +@patch("esphome.git.run_git_command") +@patch("pathlib.Path.is_dir") +def test_clone_or_update_with_none_refresh_no_update( + mock_is_dir: MagicMock, mock_run_git: MagicMock +) -> None: + """Test that refresh=None skips updates for existing repos.""" + # Setup - repo already exists + mock_is_dir.return_value = True + + # Mock file timestamps + with patch("pathlib.Path.exists") as mock_exists: + mock_exists.return_value = True + with patch("pathlib.Path.stat") as mock_stat: + mock_stat_result = Mock() + mock_stat_result.st_mtime = datetime.now().timestamp() + mock_stat.return_value = mock_stat_result + + # Call with refresh=None + repo_dir, revert = git.clone_or_update( + url="https://github.com/test/repo", + ref=None, + refresh=None, + domain="test", + ) + + # Should NOT call git fetch or any update commands + mock_run_git.assert_not_called() + assert revert is None + + +@patch("esphome.git.run_git_command") +@patch("pathlib.Path.is_dir") +def test_clone_or_update_with_refresh_updates_old_repo( + mock_is_dir: MagicMock, mock_run_git: MagicMock +) -> None: + """Test that refresh triggers update for old repos.""" + # Setup - repo already exists + mock_is_dir.return_value = True + mock_run_git.return_value = "abc123" # mock SHA + + # Mock file timestamps - 2 days old + with patch("pathlib.Path.exists") as mock_exists: + mock_exists.return_value = True + with patch("pathlib.Path.stat") as mock_stat: + mock_stat_result = Mock() + old_time = datetime.now() - timedelta(days=2) + mock_stat_result.st_mtime = old_time.timestamp() + mock_stat.return_value = mock_stat_result + + # Call with refresh=1d (1 day) + refresh = TimePeriodSeconds(days=1) + repo_dir, revert = git.clone_or_update( + url="https://github.com/test/repo", + ref=None, + refresh=refresh, + domain="test", + ) + + # Should call git fetch and update commands + assert mock_run_git.called + # Check for fetch command + fetch_calls = [ + call for call in mock_run_git.call_args_list if "fetch" in str(call) + ] + assert len(fetch_calls) > 0 + + +@patch("esphome.git.run_git_command") +@patch("pathlib.Path.is_dir") +def test_clone_or_update_with_refresh_skips_fresh_repo( + mock_is_dir: MagicMock, mock_run_git: MagicMock +) -> None: + """Test that refresh doesn't update fresh repos.""" + # Setup - repo already exists + mock_is_dir.return_value = True + + # Mock file timestamps - 1 hour old + with patch("pathlib.Path.exists") as mock_exists: + mock_exists.return_value = True + with patch("pathlib.Path.stat") as mock_stat: + mock_stat_result = Mock() + recent_time = datetime.now() - timedelta(hours=1) + mock_stat_result.st_mtime = recent_time.timestamp() + mock_stat.return_value = mock_stat_result + + # Call with refresh=1d (1 day) + refresh = TimePeriodSeconds(days=1) + repo_dir, revert = git.clone_or_update( + url="https://github.com/test/repo", + ref=None, + refresh=refresh, + domain="test", + ) + + # Should NOT call git fetch since repo is fresh + mock_run_git.assert_not_called() + assert revert is None + + +@patch("esphome.git.run_git_command") +@patch("pathlib.Path.is_dir") +def test_clone_or_update_clones_missing_repo( + mock_is_dir: MagicMock, mock_run_git: MagicMock +) -> None: + """Test that missing repos are cloned regardless of refresh setting.""" + # Setup - repo doesn't exist + mock_is_dir.return_value = False + + # Test with refresh=None + repo_dir, revert = git.clone_or_update( + url="https://github.com/test/repo", + ref=None, + refresh=None, + domain="test", + ) + + # Should call git clone + assert mock_run_git.called + clone_calls = [call for call in mock_run_git.call_args_list if "clone" in str(call)] + assert len(clone_calls) > 0 + + # Reset mock + mock_run_git.reset_mock() + + # Test with refresh=1d + mock_is_dir.return_value = False + refresh = TimePeriodSeconds(days=1) + repo_dir, revert = git.clone_or_update( + url="https://github.com/test/repo2", + ref=None, + refresh=refresh, + domain="test", + ) + + # Should still call git clone + assert mock_run_git.called + clone_calls = [call for call in mock_run_git.call_args_list if "clone" in str(call)] + assert len(clone_calls) > 0