From edd8fa8d6f1e4d773d2592d6ebba0cfc3c818830 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Tue, 16 Sep 2025 12:02:57 -0500 Subject: [PATCH] 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