From 8e13335ff6f7a166355a87fc6255c2ffd09f473d Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Tue, 16 Sep 2025 11:49:36 -0500 Subject: [PATCH 01/11] fixes --- esphome/__main__.py | 7 +- .../external_components/__init__.py | 14 +- esphome/components/packages/__init__.py | 18 ++- esphome/config.py | 24 +-- esphome/git.py | 2 +- .../external_components/test_init.py | 113 ++++++++++++++ tests/component_tests/packages/test_init.py | 102 ++++++++++++ .../component_tests/packages/test_packages.py | 93 +++++++++++ tests/unit_tests/core/test_config.py | 115 ++++++++++++++ tests/unit_tests/test_git.py | 147 ++++++++++++++++++ 10 files changed, 610 insertions(+), 25 deletions(-) create mode 100644 tests/component_tests/external_components/test_init.py create mode 100644 tests/component_tests/packages/test_init.py create mode 100644 tests/unit_tests/test_git.py 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 From 7d87dbe641bccd458a16966e52378800676f2e7b Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Tue, 16 Sep 2025 11:51:44 -0500 Subject: [PATCH 02/11] fixes --- tests/component_tests/conftest.py | 21 ++++++++++++ tests/component_tests/packages/test_init.py | 36 ++++++++------------- 2 files changed, 34 insertions(+), 23 deletions(-) diff --git a/tests/component_tests/conftest.py b/tests/component_tests/conftest.py index 2045b03502..1756e4de14 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 @@ -135,3 +136,23 @@ 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.""" + from esphome.util import OrderedDict + + with mock.patch("esphome.yaml_util.load_yaml") as mock_func: + # Default return value + mock_func.return_value = OrderedDict({"sensor": []}) + yield mock_func diff --git a/tests/component_tests/packages/test_init.py b/tests/component_tests/packages/test_init.py index fbf12829ef..25493c2770 100644 --- a/tests/component_tests/packages/test_init.py +++ b/tests/component_tests/packages/test_init.py @@ -2,24 +2,20 @@ from pathlib import Path from typing import Any -from unittest.mock import MagicMock, patch +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 -@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 + mock_clone_or_update: MagicMock, mock_load_yaml: 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": []}) + with MagicMock() as mock_is_file: + mock_is_file.return_value = True + Path.is_file = mock_is_file config: dict[str, Any] = { CONF_PACKAGES: { @@ -40,17 +36,14 @@ def test_packages_skip_update_true( 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 + mock_clone_or_update: MagicMock, mock_load_yaml: 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": []}) + with MagicMock() as mock_is_file: + mock_is_file.return_value = True + Path.is_file = mock_is_file config: dict[str, Any] = { CONF_PACKAGES: { @@ -71,17 +64,14 @@ def test_packages_skip_update_false( 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 + mock_clone_or_update: MagicMock, mock_load_yaml: 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": []}) + with MagicMock() as mock_is_file: + mock_is_file.return_value = True + Path.is_file = mock_is_file config: dict[str, Any] = { CONF_PACKAGES: { From 9be832a23cf18628ddd7d8dc1ea9d0465d9af4e9 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Tue, 16 Sep 2025 11:51:55 -0500 Subject: [PATCH 03/11] fixes --- tests/component_tests/conftest.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/component_tests/conftest.py b/tests/component_tests/conftest.py index 1756e4de14..fe765118ff 100644 --- a/tests/component_tests/conftest.py +++ b/tests/component_tests/conftest.py @@ -18,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 @@ -150,7 +151,6 @@ def mock_clone_or_update() -> Generator[Any]: @pytest.fixture def mock_load_yaml() -> Generator[Any]: """Mock yaml_util.load_yaml for testing.""" - from esphome.util import OrderedDict with mock.patch("esphome.yaml_util.load_yaml") as mock_func: # Default return value From 586f24e02d70086e14c14bb2fe7a600d29d2ee4c Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Tue, 16 Sep 2025 11:54:09 -0500 Subject: [PATCH 04/11] fixes --- .../external_components/test_init.py | 168 ++++++++++-------- tests/component_tests/packages/test_init.py | 46 +++-- 2 files changed, 124 insertions(+), 90 deletions(-) diff --git a/tests/component_tests/external_components/test_init.py b/tests/component_tests/external_components/test_init.py index bdec13fe0f..da1bdc3d90 100644 --- a/tests/component_tests/external_components/test_init.py +++ b/tests/component_tests/external_components/test_init.py @@ -14,100 +14,118 @@ from esphome.const import ( ) -@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 + tmp_path: Path, 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) + # Create a components directory structure + components_dir = tmp_path / "components" + components_dir.mkdir() - config: dict[str, Any] = { - CONF_EXTERNAL_COMPONENTS: [ - { - CONF_SOURCE: { - "type": TYPE_GIT, - CONF_URL: "https://github.com/test/components", - }, - CONF_REFRESH: "1d", - "components": "all", - } - ] - } + # Create a test component + test_component_dir = components_dir / "test_component" + test_component_dir.mkdir() + (test_component_dir / "__init__.py").write_text("# Test component") - # Call with skip_update=True - do_external_components_pass(config, skip_update=True) + # Set up mock to return our tmp_path + mock_clone_or_update.return_value = (tmp_path, None) - # 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 + with patch("esphome.loader.install_meta_finder"): + 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 + tmp_path: Path, 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) + # Create a components directory structure + components_dir = tmp_path / "components" + components_dir.mkdir() - config: dict[str, Any] = { - CONF_EXTERNAL_COMPONENTS: [ - { - CONF_SOURCE: { - "type": TYPE_GIT, - CONF_URL: "https://github.com/test/components", - }, - CONF_REFRESH: "1d", - "components": "all", - } - ] - } + # Create a test component + test_component_dir = components_dir / "test_component" + test_component_dir.mkdir() + (test_component_dir / "__init__.py").write_text("# Test component") - # Call with skip_update=False - do_external_components_pass(config, skip_update=False) + # Set up mock to return our tmp_path + mock_clone_or_update.return_value = (tmp_path, None) - # 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" + with patch("esphome.loader.install_meta_finder"): + 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 + tmp_path: Path, 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) + # Create a components directory structure + components_dir = tmp_path / "components" + components_dir.mkdir() - config: dict[str, Any] = { - CONF_EXTERNAL_COMPONENTS: [ - { - CONF_SOURCE: { - "type": TYPE_GIT, - CONF_URL: "https://github.com/test/components", - }, - CONF_REFRESH: "1d", - "components": "all", - } - ] - } + # Create a test component + test_component_dir = components_dir / "test_component" + test_component_dir.mkdir() + (test_component_dir / "__init__.py").write_text("# Test component") - # Call without skip_update parameter - do_external_components_pass(config) + # Set up mock to return our tmp_path + mock_clone_or_update.return_value = (tmp_path, None) - # 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" + with patch("esphome.loader.install_meta_finder"): + 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 index 25493c2770..4f66f83e84 100644 --- a/tests/component_tests/packages/test_init.py +++ b/tests/component_tests/packages/test_init.py @@ -6,16 +6,22 @@ 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( - mock_clone_or_update: MagicMock, mock_load_yaml: MagicMock + tmp_path: Path, mock_clone_or_update: MagicMock, mock_load_yaml: MagicMock ) -> None: """Test that packages don't update when skip_update=True.""" - # Setup mocks - with MagicMock() as mock_is_file: - mock_is_file.return_value = True - Path.is_file = mock_is_file + # 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: { @@ -37,13 +43,18 @@ def test_packages_skip_update_true( def test_packages_skip_update_false( - mock_clone_or_update: MagicMock, mock_load_yaml: MagicMock + tmp_path: Path, mock_clone_or_update: MagicMock, mock_load_yaml: MagicMock ) -> None: """Test that packages update when skip_update=False.""" - # Setup mocks - with MagicMock() as mock_is_file: - mock_is_file.return_value = True - Path.is_file = mock_is_file + # 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: { @@ -65,13 +76,18 @@ def test_packages_skip_update_false( def test_packages_default_no_skip( - mock_clone_or_update: MagicMock, mock_load_yaml: MagicMock + tmp_path: Path, mock_clone_or_update: MagicMock, mock_load_yaml: MagicMock ) -> None: """Test that packages update by default when skip_update not specified.""" - # Setup mocks - with MagicMock() as mock_is_file: - mock_is_file.return_value = True - Path.is_file = mock_is_file + # 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: { From c39320c5152057a83f66886e0a81e5917e62f02e Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Tue, 16 Sep 2025 11:57:10 -0500 Subject: [PATCH 05/11] fixes --- tests/component_tests/conftest.py | 7 + .../external_components/test_init.py | 119 ++++----- .../component_tests/packages/test_packages.py | 93 ------- tests/unit_tests/conftest.py | 7 + tests/unit_tests/test_git.py | 247 ++++++++++-------- 5 files changed, 212 insertions(+), 261 deletions(-) diff --git a/tests/component_tests/conftest.py b/tests/component_tests/conftest.py index fe765118ff..189549bcd8 100644 --- a/tests/component_tests/conftest.py +++ b/tests/component_tests/conftest.py @@ -156,3 +156,10 @@ def mock_load_yaml() -> Generator[Any]: # 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 index da1bdc3d90..efc81c5475 100644 --- a/tests/component_tests/external_components/test_init.py +++ b/tests/component_tests/external_components/test_init.py @@ -2,7 +2,7 @@ from pathlib import Path from typing import Any -from unittest.mock import MagicMock, patch +from unittest.mock import MagicMock from esphome.components.external_components import do_external_components_pass from esphome.const import ( @@ -15,7 +15,7 @@ from esphome.const import ( def test_external_components_skip_update_true( - tmp_path: Path, mock_clone_or_update: MagicMock + 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 @@ -30,31 +30,30 @@ def test_external_components_skip_update_true( # Set up mock to return our tmp_path mock_clone_or_update.return_value = (tmp_path, None) - with patch("esphome.loader.install_meta_finder"): - config: dict[str, Any] = { - CONF_EXTERNAL_COMPONENTS: [ - { - CONF_SOURCE: { - "type": TYPE_GIT, - CONF_URL: "https://github.com/test/components", - }, - CONF_REFRESH: "1d", - "components": "all", - } - ] - } + 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) + # 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 + # 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 def test_external_components_skip_update_false( - tmp_path: Path, mock_clone_or_update: MagicMock + 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 @@ -69,31 +68,30 @@ def test_external_components_skip_update_false( # Set up mock to return our tmp_path mock_clone_or_update.return_value = (tmp_path, None) - with patch("esphome.loader.install_meta_finder"): - config: dict[str, Any] = { - CONF_EXTERNAL_COMPONENTS: [ - { - CONF_SOURCE: { - "type": TYPE_GIT, - CONF_URL: "https://github.com/test/components", - }, - CONF_REFRESH: "1d", - "components": "all", - } - ] - } + 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) + # 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" + # 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" def test_external_components_default_no_skip( - tmp_path: Path, mock_clone_or_update: MagicMock + 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 @@ -108,24 +106,23 @@ def test_external_components_default_no_skip( # Set up mock to return our tmp_path mock_clone_or_update.return_value = (tmp_path, None) - with patch("esphome.loader.install_meta_finder"): - config: dict[str, Any] = { - CONF_EXTERNAL_COMPONENTS: [ - { - CONF_SOURCE: { - "type": TYPE_GIT, - CONF_URL: "https://github.com/test/components", - }, - CONF_REFRESH: "1d", - "components": "all", - } - ] - } + 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) + # 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" + # 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 99ed661649..4712daad0d 100644 --- a/tests/component_tests/packages/test_packages.py +++ b/tests/component_tests/packages/test_packages.py @@ -732,96 +732,3 @@ 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/conftest.py b/tests/unit_tests/conftest.py index 06d06d0506..d2ba831b56 100644 --- a/tests/unit_tests/conftest.py +++ b/tests/unit_tests/conftest.py @@ -87,3 +87,10 @@ def mock_run_external_command() -> Generator[Mock, None, None]: """Mock run_external_command for platformio_api.""" with patch("esphome.platformio_api.run_external_command") as mock: 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 diff --git a/tests/unit_tests/test_git.py b/tests/unit_tests/test_git.py index 99b0f12441..a0364ea7bc 100644 --- a/tests/unit_tests/test_git.py +++ b/tests/unit_tests/test_git.py @@ -1,147 +1,180 @@ """Tests for git.py module.""" from datetime import datetime, timedelta -from unittest.mock import MagicMock, Mock, patch +from pathlib import Path +from unittest.mock import Mock 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 + tmp_path: Path, mock_run_git_command: Mock ) -> None: """Test that refresh=None skips updates for existing repos.""" - # Setup - repo already exists - mock_is_dir.return_value = True + # Create a fake git repo directory + repo_dir = tmp_path / ".esphome" / "external_components" / "test" / "test_repo" + repo_dir.mkdir(parents=True) + git_dir = repo_dir / ".git" + git_dir.mkdir() - # 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 + # Create FETCH_HEAD file with current timestamp + fetch_head = git_dir / "FETCH_HEAD" + fetch_head.write_text("test") - # Call with refresh=None - repo_dir, revert = git.clone_or_update( - url="https://github.com/test/repo", - ref=None, - refresh=None, - domain="test", - ) + # Mock _compute_destination_path to return our test directory + with Mock() as mock_compute: + mock_compute.return_value = repo_dir + git._compute_destination_path = mock_compute - # Should NOT call git fetch or any update commands - mock_run_git.assert_not_called() - assert revert is None + # Call with refresh=None + result_dir, revert = git.clone_or_update( + url="https://github.com/test/repo", + ref=None, + refresh=None, + domain="test", + ) + + # Should NOT call git commands since refresh=None and repo exists + mock_run_git_command.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 + tmp_path: Path, mock_run_git_command: Mock ) -> 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 + # Create a fake git repo directory + repo_dir = tmp_path / ".esphome" / "external_components" / "test" / "test_repo" + repo_dir.mkdir(parents=True) + git_dir = repo_dir / ".git" + git_dir.mkdir() - # 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 + # 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 + import os - # 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", - ) + os.utime(fetch_head, (old_time.timestamp(), old_time.timestamp())) - # 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 + # Mock _compute_destination_path to return our test directory + with Mock() as mock_compute: + mock_compute.return_value = repo_dir + git._compute_destination_path = mock_compute + + # 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="https://github.com/test/repo", + ref=None, + refresh=refresh, + domain="test", + ) + + # 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 -@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 + tmp_path: Path, mock_run_git_command: Mock ) -> None: """Test that refresh doesn't update fresh repos.""" - # Setup - repo already exists - mock_is_dir.return_value = True + # Create a fake git repo directory + repo_dir = tmp_path / ".esphome" / "external_components" / "test" / "test_repo" + repo_dir.mkdir(parents=True) + git_dir = repo_dir / ".git" + git_dir.mkdir() - # 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 + # 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 + import os - # 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", - ) + os.utime(fetch_head, (recent_time.timestamp(), recent_time.timestamp())) - # Should NOT call git fetch since repo is fresh - mock_run_git.assert_not_called() - assert revert is None + # Mock _compute_destination_path to return our test directory + with Mock() as mock_compute: + mock_compute.return_value = repo_dir + git._compute_destination_path = mock_compute + + # Call with refresh=1d (1 day) + refresh = TimePeriodSeconds(days=1) + result_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_command.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 + tmp_path: Path, mock_run_git_command: Mock ) -> None: """Test that missing repos are cloned regardless of refresh setting.""" - # Setup - repo doesn't exist - mock_is_dir.return_value = False + # Create base directory but not the repo itself + base_dir = tmp_path / ".esphome" / "external_components" / "test" + base_dir.mkdir(parents=True) + repo_dir = base_dir / "test_repo" - # Test with refresh=None - repo_dir, revert = git.clone_or_update( - url="https://github.com/test/repo", - ref=None, - refresh=None, - domain="test", - ) + # Mock _compute_destination_path to return our test directory + with Mock() as mock_compute: + mock_compute.return_value = repo_dir + git._compute_destination_path = mock_compute - # 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 + # Test with refresh=None + result_dir, revert = git.clone_or_update( + url="https://github.com/test/repo", + ref=None, + refresh=None, + domain="test", + ) - # Reset mock - mock_run_git.reset_mock() + # 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 - # 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", - ) + # Reset mock + mock_run_git_command.reset_mock() - # 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 + # Test with refresh=1d - should still clone since repo doesn't exist + refresh = TimePeriodSeconds(days=1) + result_dir2, revert2 = 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_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 From 452a12892e645b6cd8d42f63904fa1dd357c77b0 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Tue, 16 Sep 2025 12:01:01 -0500 Subject: [PATCH 06/11] fix reg --- esphome/components/external_components/__init__.py | 7 ++++--- esphome/git.py | 2 +- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/esphome/components/external_components/__init__.py b/esphome/components/external_components/__init__.py index 5362a2269f..6b7754f2f4 100644 --- a/esphome/components/external_components/__init__.py +++ b/esphome/components/external_components/__init__.py @@ -17,7 +17,7 @@ from esphome.const import ( TYPE_GIT, TYPE_LOCAL, ) -from esphome.core import CORE +from esphome.core import CORE, TimePeriodSeconds _LOGGER = logging.getLogger(__name__) @@ -40,8 +40,9 @@ async def to_code(config): 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 + # When skip_update is True, set a very large refresh value to prevent updates + # Using 100 years in seconds to effectively disable refresh + actual_refresh = TimePeriodSeconds(days=36500) if skip_update else refresh repo_dir, _ = git.clone_or_update( url=config[CONF_URL], ref=config.get(CONF_REF), diff --git a/esphome/git.py b/esphome/git.py index c60b928d7c..56aedd1519 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 not None and age.total_seconds() > refresh.total_seconds: + if refresh is None or 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) From edd8fa8d6f1e4d773d2592d6ebba0cfc3c818830 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Tue, 16 Sep 2025 12:02:57 -0500 Subject: [PATCH 07/11] cleaner --- .../external_components/__init__.py | 7 +- esphome/components/packages/__init__.py | 4 +- esphome/git.py | 8 ++ tests/unit_tests/test_git.py | 83 +++++++++++-------- 4 files changed, 61 insertions(+), 41 deletions(-) diff --git a/esphome/components/external_components/__init__.py b/esphome/components/external_components/__init__.py index 6b7754f2f4..ceb402c5b7 100644 --- a/esphome/components/external_components/__init__.py +++ b/esphome/components/external_components/__init__.py @@ -17,7 +17,7 @@ from esphome.const import ( TYPE_GIT, TYPE_LOCAL, ) -from esphome.core import CORE, TimePeriodSeconds +from esphome.core import CORE _LOGGER = logging.getLogger(__name__) @@ -40,9 +40,8 @@ async def to_code(config): def _process_git_config(config: dict, refresh, skip_update: bool = False) -> str: - # When skip_update is True, set a very large refresh value to prevent updates - # Using 100 years in seconds to effectively disable refresh - actual_refresh = TimePeriodSeconds(days=36500) if skip_update else refresh + # 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), diff --git a/esphome/components/packages/__init__.py b/esphome/components/packages/__init__.py index 2f964984cc..fdc75d995a 100644 --- a/esphome/components/packages/__init__.py +++ b/esphome/components/packages/__init__.py @@ -107,8 +107,8 @@ CONFIG_SCHEMA = cv.Any( 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] + # 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), 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/unit_tests/test_git.py b/tests/unit_tests/test_git.py index a0364ea7bc..b4bf6ed827 100644 --- a/tests/unit_tests/test_git.py +++ b/tests/unit_tests/test_git.py @@ -2,16 +2,16 @@ from datetime import datetime, timedelta from pathlib import Path -from unittest.mock import Mock +from unittest.mock import Mock, patch from esphome import git from esphome.core import TimePeriodSeconds -def test_clone_or_update_with_none_refresh_no_update( +def test_clone_or_update_with_never_refresh( tmp_path: Path, mock_run_git_command: Mock ) -> None: - """Test that refresh=None skips updates for existing repos.""" + """Test that NEVER_REFRESH skips updates for existing repos.""" # Create a fake git repo directory repo_dir = tmp_path / ".esphome" / "external_components" / "test" / "test_repo" repo_dir.mkdir(parents=True) @@ -23,20 +23,18 @@ def test_clone_or_update_with_none_refresh_no_update( fetch_head.write_text("test") # Mock _compute_destination_path to return our test directory - with Mock() as mock_compute: - mock_compute.return_value = repo_dir - git._compute_destination_path = mock_compute - - # Call with refresh=None + with patch.object(git, "_compute_destination_path", return_value=repo_dir): + # Call with NEVER_REFRESH result_dir, revert = git.clone_or_update( url="https://github.com/test/repo", ref=None, - refresh=None, + refresh=git.NEVER_REFRESH, domain="test", ) - # Should NOT call git commands since refresh=None and repo exists + # 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 @@ -61,10 +59,7 @@ def test_clone_or_update_with_refresh_updates_old_repo( os.utime(fetch_head, (old_time.timestamp(), old_time.timestamp())) # Mock _compute_destination_path to return our test directory - with Mock() as mock_compute: - mock_compute.return_value = repo_dir - git._compute_destination_path = mock_compute - + with patch.object(git, "_compute_destination_path", return_value=repo_dir): # Mock git command responses mock_run_git_command.return_value = "abc123" # SHA for rev-parse @@ -109,10 +104,7 @@ def test_clone_or_update_with_refresh_skips_fresh_repo( os.utime(fetch_head, (recent_time.timestamp(), recent_time.timestamp())) # Mock _compute_destination_path to return our test directory - with Mock() as mock_compute: - mock_compute.return_value = repo_dir - git._compute_destination_path = mock_compute - + with patch.object(git, "_compute_destination_path", return_value=repo_dir): # Call with refresh=1d (1 day) refresh = TimePeriodSeconds(days=1) result_dir, revert = git.clone_or_update( @@ -124,6 +116,7 @@ def test_clone_or_update_with_refresh_skips_fresh_repo( # 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 @@ -137,15 +130,12 @@ def test_clone_or_update_clones_missing_repo( repo_dir = base_dir / "test_repo" # Mock _compute_destination_path to return our test directory - with Mock() as mock_compute: - mock_compute.return_value = repo_dir - git._compute_destination_path = mock_compute - - # Test with refresh=None + with patch.object(git, "_compute_destination_path", return_value=repo_dir): + # Test with NEVER_REFRESH - should still clone since repo doesn't exist result_dir, revert = git.clone_or_update( url="https://github.com/test/repo", ref=None, - refresh=None, + refresh=git.NEVER_REFRESH, domain="test", ) @@ -158,23 +148,46 @@ def test_clone_or_update_clones_missing_repo( ] assert len(clone_calls) > 0 - # Reset mock - mock_run_git_command.reset_mock() - # Test with refresh=1d - should still clone since repo doesn't exist - refresh = TimePeriodSeconds(days=1) - result_dir2, revert2 = git.clone_or_update( - url="https://github.com/test/repo2", +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.""" + # Create a fake git repo directory + repo_dir = tmp_path / ".esphome" / "external_components" / "test" / "test_repo" + 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 + import os + + os.utime(fetch_head, (recent_time.timestamp(), recent_time.timestamp())) + + # Mock _compute_destination_path to return our test directory + with patch.object(git, "_compute_destination_path", return_value=repo_dir): + # 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="https://github.com/test/repo", ref=None, - refresh=refresh, + refresh=None, domain="test", ) - # Should still call git clone + # Should call git fetch and update commands since refresh=None means always update assert mock_run_git_command.called - clone_calls = [ + # Check for fetch command + fetch_calls = [ call for call in mock_run_git_command.call_args_list - if len(call[0]) > 0 and "clone" in call[0][0] + if len(call[0]) > 0 and "fetch" in call[0][0] ] - assert len(clone_calls) > 0 + assert len(fetch_calls) > 0 From 81cfc30f3a4ea3c3761be309d07a026d1564213b Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Tue, 16 Sep 2025 12:03:28 -0500 Subject: [PATCH 08/11] cleaner --- tests/component_tests/external_components/test_init.py | 6 ++++-- tests/component_tests/packages/test_init.py | 6 ++++-- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/tests/component_tests/external_components/test_init.py b/tests/component_tests/external_components/test_init.py index efc81c5475..d1c328b3cb 100644 --- a/tests/component_tests/external_components/test_init.py +++ b/tests/component_tests/external_components/test_init.py @@ -46,10 +46,12 @@ def test_external_components_skip_update_true( # Call with skip_update=True do_external_components_pass(config, skip_update=True) - # Verify clone_or_update was called with refresh=None + # Verify clone_or_update was called with NEVER_REFRESH mock_clone_or_update.assert_called_once() call_args = mock_clone_or_update.call_args - assert call_args.kwargs["refresh"] is None + from esphome import git + + assert call_args.kwargs["refresh"] == git.NEVER_REFRESH def test_external_components_skip_update_false( diff --git a/tests/component_tests/packages/test_init.py b/tests/component_tests/packages/test_init.py index 4f66f83e84..dfd4147c14 100644 --- a/tests/component_tests/packages/test_init.py +++ b/tests/component_tests/packages/test_init.py @@ -36,10 +36,12 @@ def test_packages_skip_update_true( # Call with skip_update=True do_packages_pass(config, skip_update=True) - # Verify clone_or_update was called with refresh=None + # Verify clone_or_update was called with NEVER_REFRESH mock_clone_or_update.assert_called_once() call_args = mock_clone_or_update.call_args - assert call_args.kwargs["refresh"] is None + from esphome import git + + assert call_args.kwargs["refresh"] == git.NEVER_REFRESH def test_packages_skip_update_false( From 1793b6a27b6a9a41fbddaa3cf770da3f346bbb5b Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Tue, 16 Sep 2025 12:04:05 -0500 Subject: [PATCH 09/11] cleaner --- tests/component_tests/external_components/test_init.py | 8 ++++++-- tests/component_tests/packages/test_init.py | 8 ++++++-- 2 files changed, 12 insertions(+), 4 deletions(-) diff --git a/tests/component_tests/external_components/test_init.py b/tests/component_tests/external_components/test_init.py index d1c328b3cb..905c0afa8b 100644 --- a/tests/component_tests/external_components/test_init.py +++ b/tests/component_tests/external_components/test_init.py @@ -89,7 +89,9 @@ def test_external_components_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" + from esphome.core import TimePeriodSeconds + + assert call_args.kwargs["refresh"] == TimePeriodSeconds(days=1) def test_external_components_default_no_skip( @@ -127,4 +129,6 @@ def test_external_components_default_no_skip( # 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" + 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 index dfd4147c14..779244e2ed 100644 --- a/tests/component_tests/packages/test_init.py +++ b/tests/component_tests/packages/test_init.py @@ -74,7 +74,9 @@ def test_packages_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" + from esphome.core import TimePeriodSeconds + + assert call_args.kwargs["refresh"] == TimePeriodSeconds(days=1) def test_packages_default_no_skip( @@ -107,4 +109,6 @@ def test_packages_default_no_skip( # 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" + from esphome.core import TimePeriodSeconds + + assert call_args.kwargs["refresh"] == TimePeriodSeconds(days=1) From c74777098f4173f64992e3b0f77068b87c722605 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Tue, 16 Sep 2025 12:05:35 -0500 Subject: [PATCH 10/11] cleaner --- tests/unit_tests/core/test_config.py | 115 --------------------------- 1 file changed, 115 deletions(-) diff --git a/tests/unit_tests/core/test_config.py b/tests/unit_tests/core/test_config.py index 921863e2bc..7d3b90794b 100644 --- a/tests/unit_tests/core/test_config.py +++ b/tests/unit_tests/core/test_config.py @@ -852,118 +852,3 @@ 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) From d249e54e8b613d50348c3c435e276d1f97f9620d Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Tue, 16 Sep 2025 12:08:18 -0500 Subject: [PATCH 11/11] cleanup, less mocking --- tests/unit_tests/test_git.py | 261 +++++++++++++++++++++-------------- 1 file changed, 157 insertions(+), 104 deletions(-) diff --git a/tests/unit_tests/test_git.py b/tests/unit_tests/test_git.py index b4bf6ed827..ebe7177bd2 100644 --- a/tests/unit_tests/test_git.py +++ b/tests/unit_tests/test_git.py @@ -1,19 +1,34 @@ """Tests for git.py module.""" from datetime import datetime, timedelta +import hashlib +import os from pathlib import Path -from unittest.mock import Mock, patch +from unittest.mock import Mock from esphome import git -from esphome.core import TimePeriodSeconds +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.""" - # Create a fake git repo directory - repo_dir = tmp_path / ".esphome" / "external_components" / "test" / "test_repo" + # Set up CORE.config_path so data_dir uses tmp_path + CORE.config_path = str(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() @@ -22,28 +37,39 @@ def test_clone_or_update_with_never_refresh( fetch_head = git_dir / "FETCH_HEAD" fetch_head.write_text("test") - # Mock _compute_destination_path to return our test directory - with patch.object(git, "_compute_destination_path", return_value=repo_dir): - # Call with NEVER_REFRESH - result_dir, revert = git.clone_or_update( - url="https://github.com/test/repo", - ref=None, - refresh=git.NEVER_REFRESH, - domain="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 + # 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.""" - # Create a fake git repo directory - repo_dir = tmp_path / ".esphome" / "external_components" / "test" / "test_repo" + # Set up CORE.config_path so data_dir uses tmp_path + CORE.config_path = str(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() @@ -54,41 +80,50 @@ def test_clone_or_update_with_refresh_updates_old_repo( old_time = datetime.now() - timedelta(days=2) fetch_head.touch() # Create the file # Set modification time to 2 days ago - import os - os.utime(fetch_head, (old_time.timestamp(), old_time.timestamp())) - # Mock _compute_destination_path to return our test directory - with patch.object(git, "_compute_destination_path", return_value=repo_dir): - # Mock git command responses - mock_run_git_command.return_value = "abc123" # SHA for rev-parse + # 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="https://github.com/test/repo", - ref=None, - refresh=refresh, - domain="test", - ) + # 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 + # 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.""" - # Create a fake git repo directory - repo_dir = tmp_path / ".esphome" / "external_components" / "test" / "test_repo" + # Set up CORE.config_path so data_dir uses tmp_path + CORE.config_path = str(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() @@ -99,62 +134,84 @@ def test_clone_or_update_with_refresh_skips_fresh_repo( recent_time = datetime.now() - timedelta(hours=1) fetch_head.touch() # Create the file # Set modification time to 1 hour ago - import os - os.utime(fetch_head, (recent_time.timestamp(), recent_time.timestamp())) - # Mock _compute_destination_path to return our test directory - with patch.object(git, "_compute_destination_path", return_value=repo_dir): - # Call with refresh=1d (1 day) - refresh = TimePeriodSeconds(days=1) - result_dir, revert = git.clone_or_update( - url="https://github.com/test/repo", - ref=None, - refresh=refresh, - domain="test", - ) + # 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 + # 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.""" - # Create base directory but not the repo itself - base_dir = tmp_path / ".esphome" / "external_components" / "test" + # Set up CORE.config_path so data_dir uses tmp_path + CORE.config_path = str(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 = base_dir / "test_repo" + # repo_dir should NOT exist + assert not repo_dir.exists() - # Mock _compute_destination_path to return our test directory - with patch.object(git, "_compute_destination_path", return_value=repo_dir): - # Test with NEVER_REFRESH - should still clone since repo doesn't exist - result_dir, revert = git.clone_or_update( - url="https://github.com/test/repo", - ref=None, - refresh=git.NEVER_REFRESH, - domain="test", - ) + # 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 + # 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.""" - # Create a fake git repo directory - repo_dir = tmp_path / ".esphome" / "external_components" / "test" / "test_repo" + # Set up CORE.config_path so data_dir uses tmp_path + CORE.config_path = str(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() @@ -165,29 +222,25 @@ def test_clone_or_update_with_none_refresh_always_updates( recent_time = datetime.now() - timedelta(seconds=1) fetch_head.touch() # Create the file # Set modification time to 1 second ago - import os - os.utime(fetch_head, (recent_time.timestamp(), recent_time.timestamp())) - # Mock _compute_destination_path to return our test directory - with patch.object(git, "_compute_destination_path", return_value=repo_dir): - # Mock git command responses - mock_run_git_command.return_value = "abc123" # SHA for rev-parse + # 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="https://github.com/test/repo", - ref=None, - refresh=None, - domain="test", - ) + # 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 + # 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