mirror of
https://github.com/esphome/esphome.git
synced 2025-09-22 13:12:22 +01:00
cleaner
This commit is contained in:
@@ -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),
|
||||
|
@@ -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),
|
||||
|
@@ -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():
|
||||
|
@@ -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
|
||||
|
Reference in New Issue
Block a user