[core] Log when a git source update is skipped and when it will next refresh (#17943)

This commit is contained in:
J. Nick Koston
2026-07-30 13:06:32 -10:00
committed by GitHub
parent 4951f4fc2e
commit 963a3379a6
5 changed files with 123 additions and 9 deletions
+6 -1
View File
@@ -2515,11 +2515,16 @@ def git_ref(value):
return value
# What `refresh: never` validates to; also used to recognize a disabled
# refresh when logging (see esphome/git.py)
SOURCE_REFRESH_NEVER = "365250d"
def source_refresh(value: str):
if value.lower() == "always":
return source_refresh("0s")
if value.lower() == "never":
return source_refresh("365250d")
return source_refresh(SOURCE_REFRESH_NEVER)
return positive_time_period_seconds(value)
+22 -1
View File
@@ -16,7 +16,12 @@ import urllib.parse
import esphome.config_validation as cv
from esphome.core import CORE, EsphomeError, TimePeriodSeconds
from esphome.helpers import add_git_ceiling_directory, rmtree, write_file
from esphome.helpers import (
add_git_ceiling_directory,
format_duration,
rmtree,
write_file,
)
if TYPE_CHECKING:
from filelock import FileLock
@@ -26,6 +31,11 @@ _LOGGER = logging.getLogger(__name__)
# Special value to indicate never refresh
NEVER_REFRESH = TimePeriodSeconds(seconds=-1)
# `refresh: never` validates to a huge interval rather than the NEVER_REFRESH
# sentinel; treat any interval at least that long as refresh disabled instead
# of logging a countdown of hundreds of years
_REFRESH_DISABLED_SECONDS = cv.source_refresh(cv.SOURCE_REFRESH_NEVER).total_seconds
# revert() runs on an already-failing path; bound its wait for the cache
# entry lock so that recovery cannot hang forever behind another process.
_REVERT_LOCK_TIMEOUT_SECONDS = 60
@@ -803,6 +813,17 @@ def _clone_or_update_locked(
return True
return repo_dir, revert
if refresh.total_seconds >= _REFRESH_DISABLED_SECONDS:
# refresh: never
_LOGGER.debug("Skipping update for %s (refresh disabled)", safe_key)
else:
_LOGGER.info(
"Skipping update for %s, will refresh on the next run after %s "
"(refresh: %s); use refresh: always to update now",
safe_key,
format_duration(refresh.total_seconds - age_seconds),
format_duration(refresh.total_seconds),
)
return repo_dir, None
+15
View File
@@ -148,6 +148,21 @@ def indent(text, padding=" "):
return "\n".join(indent_list(text, padding))
def format_duration(seconds: float) -> str:
"""Format a duration in seconds as a short string like "1d 2h" or "42s".
Uses the two largest non-zero units, with unit suffixes matching the YAML
time period shorthand (d, h, min, s).
"""
remainder = max(0, int(seconds))
parts = []
for suffix, length in (("d", 86400), ("h", 3600), ("min", 60), ("s", 1)):
value, remainder = divmod(remainder, length)
if value:
parts.append(f"{value}{suffix}")
return " ".join(parts[:2]) if parts else "0s"
# From https://stackoverflow.com/a/14945195/8924614
def cpp_string_escape(string, encoding="utf-8"):
def _should_escape(byte: int) -> bool:
+62 -7
View File
@@ -15,6 +15,7 @@ from filelock import FileLock
import pytest
from esphome import git
import esphome.config_validation as cv
from esphome.core import CORE, EsphomeError, TimePeriodSeconds
from esphome.git import GitCommandError
@@ -488,7 +489,7 @@ def test_clone_or_update_with_refresh_updates_old_repo(
def test_clone_or_update_with_refresh_skips_fresh_repo(
tmp_path: Path, mock_run_git_command: Mock
tmp_path: Path, mock_run_git_command: Mock, caplog: pytest.LogCaptureFixture
) -> None:
"""Test that refresh doesn't update fresh repos."""
# Set up CORE.config_path so data_dir uses tmp_path
@@ -513,20 +514,74 @@ def test_clone_or_update_with_refresh_skips_fresh_repo(
# Set modification time to 1 hour ago
os.utime(fetch_head, (recent_time, recent_time))
# Freeze the clock at 1 hour (plus a margin larger than any filesystem
# mtime rounding) after the mtime so the logged countdown is deterministic
frozen_now = fetch_head.stat().st_mtime + 3600.5
# 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,
)
with (
patch("esphome.git.time.time", return_value=frozen_now),
caplog.at_level(logging.INFO, logger="esphome.git"),
):
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 tell the user the update was skipped and when the next refresh is
assert f"Skipping update for {url}@{ref}" in caplog.text
assert "will refresh on the next run after 22h 59min" in caplog.text
assert "(refresh: 1d)" in caplog.text
def test_clone_or_update_with_refresh_never_logs_refresh_disabled(
tmp_path: Path, mock_run_git_command: Mock, caplog: pytest.LogCaptureFixture
) -> None:
"""Test that a config-level refresh: never skips without a countdown log."""
# Set up CORE.config_path so data_dir uses tmp_path
CORE.config_path = tmp_path / "test.yaml"
url = "https://github.com/test/repo"
ref = None
domain = "test"
repo_dir = _compute_repo_dir(url, ref, domain)
# Create the git repo directory structure
repo_dir.mkdir(parents=True)
git_dir = repo_dir / ".git"
git_dir.mkdir()
_mark_clone_complete(repo_dir)
# Create FETCH_HEAD file with current timestamp
fetch_head = git_dir / "FETCH_HEAD"
fetch_head.write_text("test")
# refresh: never validates to 365250 days, not the NEVER_REFRESH sentinel
refresh = cv.source_refresh("never")
with caplog.at_level(logging.DEBUG, logger="esphome.git"):
result_dir, revert = git.clone_or_update(
url=url,
ref=ref,
refresh=refresh,
domain=domain,
)
mock_run_git_command.assert_not_called()
assert result_dir == repo_dir
assert revert is None
# Should log refresh disabled at debug level, not a countdown
assert f"Skipping update for {url}@{ref} (refresh disabled)" in caplog.text
assert "will refresh on the next run" not in caplog.text
def test_clone_or_update_clones_missing_repo(
tmp_path: Path, mock_run_git_command: Mock
+18
View File
@@ -1074,3 +1074,21 @@ def test_progressbar_enabled_on_pipe_with_dashboard(monkeypatch) -> None:
bar = ProgressBar("Uploading", stream=stream)
assert bar.enabled is True
@pytest.mark.parametrize(
("seconds", "expected"),
[
(0, "0s"),
(42, "42s"),
(60, "1min"),
(3661, "1h 1min"),
(86400, "1d"),
(90000, "1d 1h"),
(86700, "1d 5min"),
(-5, "0s"),
],
)
def test_format_duration(seconds: float, expected: str) -> None:
"""Test that durations are rendered as short human-readable strings."""
assert helpers.format_duration(seconds) == expected