mirror of
https://github.com/esphome/esphome.git
synced 2025-09-22 21:22:22 +01:00
fixes
This commit is contained in:
@@ -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
|
||||
|
@@ -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)
|
||||
|
@@ -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"
|
||||
|
@@ -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
|
||||
|
@@ -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)
|
||||
|
113
tests/component_tests/external_components/test_init.py
Normal file
113
tests/component_tests/external_components/test_init.py
Normal file
@@ -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"
|
102
tests/component_tests/packages/test_init.py
Normal file
102
tests/component_tests/packages/test_init.py
Normal file
@@ -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"
|
@@ -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"
|
||||
|
@@ -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)
|
||||
|
147
tests/unit_tests/test_git.py
Normal file
147
tests/unit_tests/test_git.py
Normal file
@@ -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
|
Reference in New Issue
Block a user