1
0
mirror of https://github.com/esphome/esphome.git synced 2025-09-22 21:22:22 +01:00
This commit is contained in:
J. Nick Koston
2025-09-16 11:57:10 -05:00
parent 586f24e02d
commit c39320c515
5 changed files with 212 additions and 261 deletions

View File

@@ -156,3 +156,10 @@ def mock_load_yaml() -> Generator[Any]:
# Default return value # Default return value
mock_func.return_value = OrderedDict({"sensor": []}) mock_func.return_value = OrderedDict({"sensor": []})
yield mock_func yield mock_func
@pytest.fixture
def mock_install_meta_finder() -> Generator[Any]:
"""Mock loader.install_meta_finder for testing."""
with mock.patch("esphome.loader.install_meta_finder") as mock_func:
yield mock_func

View File

@@ -2,7 +2,7 @@
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock
from esphome.components.external_components import do_external_components_pass from esphome.components.external_components import do_external_components_pass
from esphome.const import ( from esphome.const import (
@@ -15,7 +15,7 @@ from esphome.const import (
def test_external_components_skip_update_true( def test_external_components_skip_update_true(
tmp_path: Path, mock_clone_or_update: MagicMock tmp_path: Path, mock_clone_or_update: MagicMock, mock_install_meta_finder: MagicMock
) -> None: ) -> None:
"""Test that external components don't update when skip_update=True.""" """Test that external components don't update when skip_update=True."""
# Create a components directory structure # Create a components directory structure
@@ -30,31 +30,30 @@ def test_external_components_skip_update_true(
# Set up mock to return our tmp_path # Set up mock to return our tmp_path
mock_clone_or_update.return_value = (tmp_path, None) mock_clone_or_update.return_value = (tmp_path, None)
with patch("esphome.loader.install_meta_finder"): config: dict[str, Any] = {
config: dict[str, Any] = { CONF_EXTERNAL_COMPONENTS: [
CONF_EXTERNAL_COMPONENTS: [ {
{ CONF_SOURCE: {
CONF_SOURCE: { "type": TYPE_GIT,
"type": TYPE_GIT, CONF_URL: "https://github.com/test/components",
CONF_URL: "https://github.com/test/components", },
}, CONF_REFRESH: "1d",
CONF_REFRESH: "1d", "components": "all",
"components": "all", }
} ]
] }
}
# Call with skip_update=True # Call with skip_update=True
do_external_components_pass(config, skip_update=True) do_external_components_pass(config, skip_update=True)
# Verify clone_or_update was called with refresh=None # Verify clone_or_update was called with refresh=None
mock_clone_or_update.assert_called_once() mock_clone_or_update.assert_called_once()
call_args = mock_clone_or_update.call_args call_args = mock_clone_or_update.call_args
assert call_args.kwargs["refresh"] is None assert call_args.kwargs["refresh"] is None
def test_external_components_skip_update_false( def test_external_components_skip_update_false(
tmp_path: Path, mock_clone_or_update: MagicMock tmp_path: Path, mock_clone_or_update: MagicMock, mock_install_meta_finder: MagicMock
) -> None: ) -> None:
"""Test that external components update when skip_update=False.""" """Test that external components update when skip_update=False."""
# Create a components directory structure # Create a components directory structure
@@ -69,31 +68,30 @@ def test_external_components_skip_update_false(
# Set up mock to return our tmp_path # Set up mock to return our tmp_path
mock_clone_or_update.return_value = (tmp_path, None) mock_clone_or_update.return_value = (tmp_path, None)
with patch("esphome.loader.install_meta_finder"): config: dict[str, Any] = {
config: dict[str, Any] = { CONF_EXTERNAL_COMPONENTS: [
CONF_EXTERNAL_COMPONENTS: [ {
{ CONF_SOURCE: {
CONF_SOURCE: { "type": TYPE_GIT,
"type": TYPE_GIT, CONF_URL: "https://github.com/test/components",
CONF_URL: "https://github.com/test/components", },
}, CONF_REFRESH: "1d",
CONF_REFRESH: "1d", "components": "all",
"components": "all", }
} ]
] }
}
# Call with skip_update=False # Call with skip_update=False
do_external_components_pass(config, skip_update=False) do_external_components_pass(config, skip_update=False)
# Verify clone_or_update was called with actual refresh value # Verify clone_or_update was called with actual refresh value
mock_clone_or_update.assert_called_once() mock_clone_or_update.assert_called_once()
call_args = mock_clone_or_update.call_args call_args = mock_clone_or_update.call_args
assert call_args.kwargs["refresh"] == "1d" assert call_args.kwargs["refresh"] == "1d"
def test_external_components_default_no_skip( def test_external_components_default_no_skip(
tmp_path: Path, mock_clone_or_update: MagicMock tmp_path: Path, mock_clone_or_update: MagicMock, mock_install_meta_finder: MagicMock
) -> None: ) -> None:
"""Test that external components update by default when skip_update not specified.""" """Test that external components update by default when skip_update not specified."""
# Create a components directory structure # Create a components directory structure
@@ -108,24 +106,23 @@ def test_external_components_default_no_skip(
# Set up mock to return our tmp_path # Set up mock to return our tmp_path
mock_clone_or_update.return_value = (tmp_path, None) mock_clone_or_update.return_value = (tmp_path, None)
with patch("esphome.loader.install_meta_finder"): config: dict[str, Any] = {
config: dict[str, Any] = { CONF_EXTERNAL_COMPONENTS: [
CONF_EXTERNAL_COMPONENTS: [ {
{ CONF_SOURCE: {
CONF_SOURCE: { "type": TYPE_GIT,
"type": TYPE_GIT, CONF_URL: "https://github.com/test/components",
CONF_URL: "https://github.com/test/components", },
}, CONF_REFRESH: "1d",
CONF_REFRESH: "1d", "components": "all",
"components": "all", }
} ]
] }
}
# Call without skip_update parameter # Call without skip_update parameter
do_external_components_pass(config) do_external_components_pass(config)
# Verify clone_or_update was called with actual refresh value # Verify clone_or_update was called with actual refresh value
mock_clone_or_update.assert_called_once() mock_clone_or_update.assert_called_once()
call_args = mock_clone_or_update.call_args call_args = mock_clone_or_update.call_args
assert call_args.kwargs["refresh"] == "1d" assert call_args.kwargs["refresh"] == "1d"

View File

@@ -732,96 +732,3 @@ def test_remote_packages_with_files_and_vars(
actual = do_packages_pass(config) actual = do_packages_pass(config)
assert actual == expected 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"

View File

@@ -87,3 +87,10 @@ def mock_run_external_command() -> Generator[Mock, None, None]:
"""Mock run_external_command for platformio_api.""" """Mock run_external_command for platformio_api."""
with patch("esphome.platformio_api.run_external_command") as mock: with patch("esphome.platformio_api.run_external_command") as mock:
yield mock yield mock
@pytest.fixture
def mock_run_git_command() -> Generator[Mock, None, None]:
"""Mock run_git_command for git module."""
with patch("esphome.git.run_git_command") as mock:
yield mock

View File

@@ -1,147 +1,180 @@
"""Tests for git.py module.""" """Tests for git.py module."""
from datetime import datetime, timedelta from datetime import datetime, timedelta
from unittest.mock import MagicMock, Mock, patch from pathlib import Path
from unittest.mock import Mock
from esphome import git from esphome import git
from esphome.core import TimePeriodSeconds 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( def test_clone_or_update_with_none_refresh_no_update(
mock_is_dir: MagicMock, mock_run_git: MagicMock tmp_path: Path, mock_run_git_command: Mock
) -> None: ) -> None:
"""Test that refresh=None skips updates for existing repos.""" """Test that refresh=None skips updates for existing repos."""
# Setup - repo already exists # Create a fake git repo directory
mock_is_dir.return_value = True repo_dir = tmp_path / ".esphome" / "external_components" / "test" / "test_repo"
repo_dir.mkdir(parents=True)
git_dir = repo_dir / ".git"
git_dir.mkdir()
# Mock file timestamps # Create FETCH_HEAD file with current timestamp
with patch("pathlib.Path.exists") as mock_exists: fetch_head = git_dir / "FETCH_HEAD"
mock_exists.return_value = True fetch_head.write_text("test")
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 # Mock _compute_destination_path to return our test directory
repo_dir, revert = git.clone_or_update( with Mock() as mock_compute:
url="https://github.com/test/repo", mock_compute.return_value = repo_dir
ref=None, git._compute_destination_path = mock_compute
refresh=None,
domain="test",
)
# Should NOT call git fetch or any update commands # Call with refresh=None
mock_run_git.assert_not_called() result_dir, revert = git.clone_or_update(
assert revert is None url="https://github.com/test/repo",
ref=None,
refresh=None,
domain="test",
)
# Should NOT call git commands since refresh=None and repo exists
mock_run_git_command.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( def test_clone_or_update_with_refresh_updates_old_repo(
mock_is_dir: MagicMock, mock_run_git: MagicMock tmp_path: Path, mock_run_git_command: Mock
) -> None: ) -> None:
"""Test that refresh triggers update for old repos.""" """Test that refresh triggers update for old repos."""
# Setup - repo already exists # Create a fake git repo directory
mock_is_dir.return_value = True repo_dir = tmp_path / ".esphome" / "external_components" / "test" / "test_repo"
mock_run_git.return_value = "abc123" # mock SHA repo_dir.mkdir(parents=True)
git_dir = repo_dir / ".git"
git_dir.mkdir()
# Mock file timestamps - 2 days old # Create FETCH_HEAD file with old timestamp (2 days ago)
with patch("pathlib.Path.exists") as mock_exists: fetch_head = git_dir / "FETCH_HEAD"
mock_exists.return_value = True fetch_head.write_text("test")
with patch("pathlib.Path.stat") as mock_stat: old_time = datetime.now() - timedelta(days=2)
mock_stat_result = Mock() fetch_head.touch() # Create the file
old_time = datetime.now() - timedelta(days=2) # Set modification time to 2 days ago
mock_stat_result.st_mtime = old_time.timestamp() import os
mock_stat.return_value = mock_stat_result
# Call with refresh=1d (1 day) os.utime(fetch_head, (old_time.timestamp(), old_time.timestamp()))
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 # Mock _compute_destination_path to return our test directory
assert mock_run_git.called with Mock() as mock_compute:
# Check for fetch command mock_compute.return_value = repo_dir
fetch_calls = [ git._compute_destination_path = mock_compute
call for call in mock_run_git.call_args_list if "fetch" in str(call)
] # Mock git command responses
assert len(fetch_calls) > 0 mock_run_git_command.return_value = "abc123" # SHA for rev-parse
# Call with refresh=1d (1 day)
refresh = TimePeriodSeconds(days=1)
result_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 since repo is older than refresh
assert mock_run_git_command.called
# Check for fetch command
fetch_calls = [
call
for call in mock_run_git_command.call_args_list
if len(call[0]) > 0 and "fetch" in call[0][0]
]
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( def test_clone_or_update_with_refresh_skips_fresh_repo(
mock_is_dir: MagicMock, mock_run_git: MagicMock tmp_path: Path, mock_run_git_command: Mock
) -> None: ) -> None:
"""Test that refresh doesn't update fresh repos.""" """Test that refresh doesn't update fresh repos."""
# Setup - repo already exists # Create a fake git repo directory
mock_is_dir.return_value = True repo_dir = tmp_path / ".esphome" / "external_components" / "test" / "test_repo"
repo_dir.mkdir(parents=True)
git_dir = repo_dir / ".git"
git_dir.mkdir()
# Mock file timestamps - 1 hour old # Create FETCH_HEAD file with recent timestamp (1 hour ago)
with patch("pathlib.Path.exists") as mock_exists: fetch_head = git_dir / "FETCH_HEAD"
mock_exists.return_value = True fetch_head.write_text("test")
with patch("pathlib.Path.stat") as mock_stat: recent_time = datetime.now() - timedelta(hours=1)
mock_stat_result = Mock() fetch_head.touch() # Create the file
recent_time = datetime.now() - timedelta(hours=1) # Set modification time to 1 hour ago
mock_stat_result.st_mtime = recent_time.timestamp() import os
mock_stat.return_value = mock_stat_result
# Call with refresh=1d (1 day) os.utime(fetch_head, (recent_time.timestamp(), recent_time.timestamp()))
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 _compute_destination_path to return our test directory
mock_run_git.assert_not_called() with Mock() as mock_compute:
assert revert is None mock_compute.return_value = repo_dir
git._compute_destination_path = mock_compute
# Call with refresh=1d (1 day)
refresh = TimePeriodSeconds(days=1)
result_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_command.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( def test_clone_or_update_clones_missing_repo(
mock_is_dir: MagicMock, mock_run_git: MagicMock tmp_path: Path, mock_run_git_command: Mock
) -> None: ) -> None:
"""Test that missing repos are cloned regardless of refresh setting.""" """Test that missing repos are cloned regardless of refresh setting."""
# Setup - repo doesn't exist # Create base directory but not the repo itself
mock_is_dir.return_value = False base_dir = tmp_path / ".esphome" / "external_components" / "test"
base_dir.mkdir(parents=True)
repo_dir = base_dir / "test_repo"
# Test with refresh=None # Mock _compute_destination_path to return our test directory
repo_dir, revert = git.clone_or_update( with Mock() as mock_compute:
url="https://github.com/test/repo", mock_compute.return_value = repo_dir
ref=None, git._compute_destination_path = mock_compute
refresh=None,
domain="test",
)
# Should call git clone # Test with refresh=None
assert mock_run_git.called result_dir, revert = git.clone_or_update(
clone_calls = [call for call in mock_run_git.call_args_list if "clone" in str(call)] url="https://github.com/test/repo",
assert len(clone_calls) > 0 ref=None,
refresh=None,
domain="test",
)
# Reset mock # Should call git clone
mock_run_git.reset_mock() assert mock_run_git_command.called
clone_calls = [
call
for call in mock_run_git_command.call_args_list
if len(call[0]) > 0 and "clone" in call[0][0]
]
assert len(clone_calls) > 0
# Test with refresh=1d # Reset mock
mock_is_dir.return_value = False mock_run_git_command.reset_mock()
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 # Test with refresh=1d - should still clone since repo doesn't exist
assert mock_run_git.called refresh = TimePeriodSeconds(days=1)
clone_calls = [call for call in mock_run_git.call_args_list if "clone" in str(call)] result_dir2, revert2 = git.clone_or_update(
assert len(clone_calls) > 0 url="https://github.com/test/repo2",
ref=None,
refresh=refresh,
domain="test",
)
# Should still call git clone
assert mock_run_git_command.called
clone_calls = [
call
for call in mock_run_git_command.call_args_list
if len(call[0]) > 0 and "clone" in call[0][0]
]
assert len(clone_calls) > 0