diff --git a/tests/component_tests/conftest.py b/tests/component_tests/conftest.py index fe765118ff..189549bcd8 100644 --- a/tests/component_tests/conftest.py +++ b/tests/component_tests/conftest.py @@ -156,3 +156,10 @@ def mock_load_yaml() -> Generator[Any]: # Default return value mock_func.return_value = OrderedDict({"sensor": []}) 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 diff --git a/tests/component_tests/external_components/test_init.py b/tests/component_tests/external_components/test_init.py index da1bdc3d90..efc81c5475 100644 --- a/tests/component_tests/external_components/test_init.py +++ b/tests/component_tests/external_components/test_init.py @@ -2,7 +2,7 @@ from pathlib import Path 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.const import ( @@ -15,7 +15,7 @@ from esphome.const import ( 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: """Test that external components don't update when skip_update=True.""" # Create a components directory structure @@ -30,31 +30,30 @@ def test_external_components_skip_update_true( # Set up mock to return our tmp_path mock_clone_or_update.return_value = (tmp_path, None) - with patch("esphome.loader.install_meta_finder"): - config: dict[str, Any] = { - CONF_EXTERNAL_COMPONENTS: [ - { - CONF_SOURCE: { - "type": TYPE_GIT, - CONF_URL: "https://github.com/test/components", - }, - CONF_REFRESH: "1d", - "components": "all", - } - ] - } + 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) + # 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 + # 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 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: """Test that external components update when skip_update=False.""" # Create a components directory structure @@ -69,31 +68,30 @@ def test_external_components_skip_update_false( # Set up mock to return our tmp_path mock_clone_or_update.return_value = (tmp_path, None) - with patch("esphome.loader.install_meta_finder"): - config: dict[str, Any] = { - CONF_EXTERNAL_COMPONENTS: [ - { - CONF_SOURCE: { - "type": TYPE_GIT, - CONF_URL: "https://github.com/test/components", - }, - CONF_REFRESH: "1d", - "components": "all", - } - ] - } + 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) + # 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" + # 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" 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: """Test that external components update by default when skip_update not specified.""" # Create a components directory structure @@ -108,24 +106,23 @@ def test_external_components_default_no_skip( # Set up mock to return our tmp_path mock_clone_or_update.return_value = (tmp_path, None) - with patch("esphome.loader.install_meta_finder"): - config: dict[str, Any] = { - CONF_EXTERNAL_COMPONENTS: [ - { - CONF_SOURCE: { - "type": TYPE_GIT, - CONF_URL: "https://github.com/test/components", - }, - CONF_REFRESH: "1d", - "components": "all", - } - ] - } + 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) + # 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" + # 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" diff --git a/tests/component_tests/packages/test_packages.py b/tests/component_tests/packages/test_packages.py index 99ed661649..4712daad0d 100644 --- a/tests/component_tests/packages/test_packages.py +++ b/tests/component_tests/packages/test_packages.py @@ -732,96 +732,3 @@ 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" diff --git a/tests/unit_tests/conftest.py b/tests/unit_tests/conftest.py index 06d06d0506..d2ba831b56 100644 --- a/tests/unit_tests/conftest.py +++ b/tests/unit_tests/conftest.py @@ -87,3 +87,10 @@ def mock_run_external_command() -> Generator[Mock, None, None]: """Mock run_external_command for platformio_api.""" with patch("esphome.platformio_api.run_external_command") as 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 diff --git a/tests/unit_tests/test_git.py b/tests/unit_tests/test_git.py index 99b0f12441..a0364ea7bc 100644 --- a/tests/unit_tests/test_git.py +++ b/tests/unit_tests/test_git.py @@ -1,147 +1,180 @@ """Tests for git.py module.""" 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.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 + tmp_path: Path, mock_run_git_command: Mock ) -> None: """Test that refresh=None skips updates for existing repos.""" - # Setup - repo already exists - mock_is_dir.return_value = True + # 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() - # 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 + # Create FETCH_HEAD file with current timestamp + fetch_head = git_dir / "FETCH_HEAD" + fetch_head.write_text("test") - # Call with refresh=None - repo_dir, revert = git.clone_or_update( - url="https://github.com/test/repo", - ref=None, - refresh=None, - domain="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 - # Should NOT call git fetch or any update commands - mock_run_git.assert_not_called() - assert revert is None + # Call with refresh=None + result_dir, revert = git.clone_or_update( + 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( - mock_is_dir: MagicMock, mock_run_git: MagicMock + tmp_path: Path, mock_run_git_command: Mock ) -> 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 + # 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() - # 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 + # Create FETCH_HEAD file with old timestamp (2 days ago) + fetch_head = git_dir / "FETCH_HEAD" + fetch_head.write_text("test") + old_time = datetime.now() - timedelta(days=2) + fetch_head.touch() # Create the file + # Set modification time to 2 days ago + import os - # 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", - ) + os.utime(fetch_head, (old_time.timestamp(), old_time.timestamp())) - # 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 + # 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 + + # Mock git command responses + 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( - mock_is_dir: MagicMock, mock_run_git: MagicMock + tmp_path: Path, mock_run_git_command: Mock ) -> None: """Test that refresh doesn't update fresh repos.""" - # Setup - repo already exists - mock_is_dir.return_value = True + # 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() - # 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 + # Create FETCH_HEAD file with recent timestamp (1 hour ago) + fetch_head = git_dir / "FETCH_HEAD" + fetch_head.write_text("test") + recent_time = datetime.now() - timedelta(hours=1) + fetch_head.touch() # Create the file + # Set modification time to 1 hour ago + import os - # 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", - ) + os.utime(fetch_head, (recent_time.timestamp(), recent_time.timestamp())) - # Should NOT call git fetch since repo is fresh - mock_run_git.assert_not_called() - assert revert is None + # 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=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( - mock_is_dir: MagicMock, mock_run_git: MagicMock + tmp_path: Path, mock_run_git_command: Mock ) -> None: """Test that missing repos are cloned regardless of refresh setting.""" - # Setup - repo doesn't exist - mock_is_dir.return_value = False + # Create base directory but not the repo itself + base_dir = tmp_path / ".esphome" / "external_components" / "test" + base_dir.mkdir(parents=True) + repo_dir = base_dir / "test_repo" - # Test with refresh=None - repo_dir, revert = git.clone_or_update( - url="https://github.com/test/repo", - ref=None, - refresh=None, - domain="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 - # 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 + # Test with refresh=None + result_dir, revert = git.clone_or_update( + url="https://github.com/test/repo", + ref=None, + refresh=None, + domain="test", + ) - # Reset mock - mock_run_git.reset_mock() + # Should 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 - # 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", - ) + # Reset mock + mock_run_git_command.reset_mock() - # 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 + # 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", + 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