github/esphome
2
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-06-24 12:33:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

[core] Unify skip_external_update and honor it in external_files for faster esphome logs (#16016)

Browse Source
This commit is contained in:
J. Nick Koston
2026-04-26 07:24:33 -05:00
committed by GitHub
parent 8dbdcfc128
commit 0f25d91e68
11 changed files with 170 additions and 204 deletions
Download Patch File Download Diff File Expand all files Collapse all files

121
tests/component_tests/external_components/test_init.py
View File

@@ -1,4 +1,4 @@
"""Tests for the external_components skip_update functionality."""
"""Tests for the external_components skip-update behavior driven by CORE.skip_external_update."""
from pathlib import Path
from typing import Any
@@ -12,25 +12,17 @@ from esphome.const import (
CONF_URL,
TYPE_GIT,
)
from esphome.core import CORE, TimePeriodSeconds
def test_external_components_skip_update_true(
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
def _make_config(tmp_path: Path) -> dict[str, Any]:
components_dir = tmp_path / "components"
components_dir.mkdir()
# Create a test component
test_component_dir = components_dir / "test_component"
test_component_dir.mkdir()
(test_component_dir / "__init__.py").write_text("# Test component")
# Set up mock to return our tmp_path
mock_clone_or_update.return_value = (tmp_path, None)
config: dict[str, Any] = {
return {
CONF_EXTERNAL_COMPONENTS: [
{
CONF_SOURCE: {
@@ -43,92 +35,37 @@ 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 NEVER_REFRESH
mock_clone_or_update.assert_called_once()
call_args = mock_clone_or_update.call_args
from esphome import git
assert call_args.kwargs["refresh"] == git.NEVER_REFRESH
def test_external_components_skip_update_false(
tmp_path: Path, mock_clone_or_update: MagicMock, mock_install_meta_finder: MagicMock
def test_external_components_skip_update_via_core_flag(
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
components_dir = tmp_path / "components"
components_dir.mkdir()
# Create a test component
test_component_dir = components_dir / "test_component"
test_component_dir.mkdir()
(test_component_dir / "__init__.py").write_text("# Test component")
# Set up mock to return our tmp_path
"""When CORE.skip_external_update is True, refresh is still passed through;
git.clone_or_update itself short-circuits the actual fetch."""
mock_clone_or_update.return_value = (tmp_path, None)
config = _make_config(tmp_path)
CORE.skip_external_update = True
do_external_components_pass(config)
mock_clone_or_update.assert_called_once()
call_args = mock_clone_or_update.call_args
# Refresh is passed through verbatim — the global flag is enforced inside git.clone_or_update.
assert call_args.kwargs["refresh"] == TimePeriodSeconds(days=1)
def test_external_components_normal_refresh(
tmp_path: Path,
mock_clone_or_update: MagicMock,
mock_install_meta_finder: MagicMock,
) -> None:
"""When CORE.skip_external_update is False, the configured refresh value is used."""
mock_clone_or_update.return_value = (tmp_path, None)
config = _make_config(tmp_path)
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
from esphome.core import TimePeriodSeconds
assert call_args.kwargs["refresh"] == TimePeriodSeconds(days=1)
def test_external_components_default_no_skip(
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
components_dir = tmp_path / "components"
components_dir.mkdir()
# Create a test component
test_component_dir = components_dir / "test_component"
test_component_dir.mkdir()
(test_component_dir / "__init__.py").write_text("# Test component")
# Set up mock to return our tmp_path
mock_clone_or_update.return_value = (tmp_path, 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
from esphome.core import TimePeriodSeconds
assert call_args.kwargs["refresh"] == TimePeriodSeconds(days=1)

115
tests/component_tests/packages/test_init.py
View File

@@ -1,4 +1,4 @@
"""Tests for the packages component skip_update functionality."""
"""Tests for the packages skip-update behavior driven by CORE.skip_external_update."""
from pathlib import Path
from typing import Any
@@ -6,24 +6,12 @@ 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.core import CORE, TimePeriodSeconds
from esphome.util import OrderedDict
def test_packages_skip_update_true(
tmp_path: Path, mock_clone_or_update: MagicMock, mock_load_yaml: MagicMock
) -> None:
"""Test that packages don't update when skip_update=True."""
# 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] = {
def _make_config() -> dict[str, Any]:
return {
CONF_PACKAGES: {
"test_package": {
CONF_URL: "https://github.com/test/repo",
@@ -33,82 +21,47 @@ 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 NEVER_REFRESH
mock_clone_or_update.assert_called_once()
call_args = mock_clone_or_update.call_args
from esphome import git
assert call_args.kwargs["refresh"] == git.NEVER_REFRESH
def test_packages_skip_update_false(
tmp_path: Path, mock_clone_or_update: MagicMock, mock_load_yaml: MagicMock
def test_packages_skip_update_via_core_flag(
tmp_path: Path,
mock_clone_or_update: MagicMock,
mock_load_yaml: MagicMock,
) -> None:
"""Test that packages update when skip_update=False."""
# Set up mock to return our tmp_path
"""When CORE.skip_external_update is True, refresh is still passed through;
git.clone_or_update itself short-circuits the actual fetch."""
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: {
"test_package": {
CONF_URL: "https://github.com/test/repo",
CONF_FILES: ["test.yaml"],
CONF_REFRESH: "1d",
}
}
}
config = _make_config()
CORE.skip_external_update = True
do_packages_pass(config, command_line_substitutions={})
mock_clone_or_update.assert_called_once()
call_args = mock_clone_or_update.call_args
# Refresh is passed through verbatim — the global flag is enforced inside git.clone_or_update.
assert call_args.kwargs["refresh"] == TimePeriodSeconds(days=1)
def test_packages_normal_refresh(
tmp_path: Path,
mock_clone_or_update: MagicMock,
mock_load_yaml: MagicMock,
) -> None:
"""When CORE.skip_external_update is False, the configured refresh value is used."""
mock_clone_or_update.return_value = (tmp_path, None)
test_file = tmp_path / "test.yaml"
test_file.write_text("sensor: []")
mock_load_yaml.return_value = OrderedDict({"sensor": []})
config = _make_config()
# Call with skip_update=False (default)
do_packages_pass(config, command_line_substitutions={}, 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
from esphome.core import TimePeriodSeconds
assert call_args.kwargs["refresh"] == TimePeriodSeconds(days=1)
def test_packages_default_no_skip(
tmp_path: Path, mock_clone_or_update: MagicMock, mock_load_yaml: MagicMock
) -> None:
"""Test that packages update by default when skip_update not specified."""
# 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: {
"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, command_line_substitutions={})
# 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
from esphome.core import TimePeriodSeconds
assert call_args.kwargs["refresh"] == TimePeriodSeconds(days=1)

46
tests/unit_tests/test_external_files.py
View File

@@ -236,3 +236,49 @@ def test_download_content_with_network_error_no_cache_fails(
with pytest.raises(Invalid, match="Could not download from.*Network error"):
external_files.download_content(url, test_file)
@patch("esphome.external_files.requests.get")
@patch("esphome.external_files.has_remote_file_changed")
def test_download_content_skip_external_update_uses_cache(
mock_has_changed: MagicMock,
mock_get: MagicMock,
setup_core: Path,
) -> None:
"""Test download_content skips network checks when CORE.skip_external_update is set."""
test_file = setup_core / "cached.txt"
cached_content = b"cached content"
test_file.write_bytes(cached_content)
CORE.skip_external_update = True
url = "https://example.com/file.txt"
result = external_files.download_content(url, test_file)
assert result == cached_content
mock_has_changed.assert_not_called()
mock_get.assert_not_called()
@patch("esphome.external_files.requests.get")
@patch("esphome.external_files.has_remote_file_changed")
def test_download_content_skip_external_update_downloads_when_missing(
mock_has_changed: MagicMock,
mock_get: MagicMock,
setup_core: Path,
) -> None:
"""Test download_content still downloads when file is missing, even with skip_external_update."""
test_file = setup_core / "missing.txt"
new_content = b"fresh content"
mock_has_changed.return_value = True
mock_response = MagicMock()
mock_response.content = new_content
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
CORE.skip_external_update = True
url = "https://example.com/file.txt"
result = external_files.download_content(url, test_file)
assert result == new_content
assert test_file.read_bytes() == new_content

29
tests/unit_tests/test_git.py
View File

@@ -236,6 +236,35 @@ def test_clone_or_update_with_never_refresh(
assert revert is None
def test_clone_or_update_skips_when_core_skip_external_update(
tmp_path: Path, mock_run_git_command: Mock
) -> None:
"""CORE.skip_external_update short-circuits the refresh for existing repos."""
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)
repo_dir.mkdir(parents=True)
git_dir = repo_dir / ".git"
git_dir.mkdir()
(git_dir / "FETCH_HEAD").write_text("test")
CORE.skip_external_update = True
result_dir, revert = git.clone_or_update(
url=url,
ref=ref,
refresh=TimePeriodSeconds(days=1),
domain=domain,
)
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:

4
tests/unit_tests/test_substitutions.py
View File

@@ -654,7 +654,7 @@ def test_resolve_package_max_depth_exceeded(tmp_path: Path) -> None:
package_config = yaml_util.IncludeFile(
parent, "test.yaml", None, always_returns_include
)
processor = _PackageProcessor({}, None, False)
processor = _PackageProcessor({}, None)
with pytest.raises(
cv.Invalid,
match=f"Maximum include nesting depth \\({MAX_INCLUDE_DEPTH}\\) exceeded",
@@ -776,7 +776,7 @@ def test_resolve_package_undefined_var_in_include_filename(tmp_path: Path) -> No
package_config = yaml_util.IncludeFile(
parent, "${undefined_var}.yaml", None, loader
)
processor = _PackageProcessor({}, None, False)
processor = _PackageProcessor({}, None)
with pytest.raises(cv.Invalid, match="unresolved substitutions"):
processor.resolve_package(package_config, substitutions.ContextVars(), [])