mirror of
https://github.com/esphome/esphome.git
synced 2025-09-01 19:02:18 +01:00
353 lines
12 KiB
Python
353 lines
12 KiB
Python
"""Unit tests for script/determine-jobs.py module."""
|
|
|
|
from collections.abc import Generator
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from unittest.mock import Mock, patch
|
|
|
|
import pytest
|
|
|
|
# Add the script directory to Python path so we can import the module
|
|
script_dir = os.path.abspath(
|
|
os.path.join(os.path.dirname(__file__), "..", "..", "script")
|
|
)
|
|
sys.path.insert(0, script_dir)
|
|
|
|
spec = importlib.util.spec_from_file_location(
|
|
"determine_jobs", os.path.join(script_dir, "determine-jobs.py")
|
|
)
|
|
determine_jobs = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(determine_jobs)
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_should_run_integration_tests() -> Generator[Mock, None, None]:
|
|
"""Mock should_run_integration_tests from helpers."""
|
|
with patch.object(determine_jobs, "should_run_integration_tests") as mock:
|
|
yield mock
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_should_run_clang_tidy() -> Generator[Mock, None, None]:
|
|
"""Mock should_run_clang_tidy from helpers."""
|
|
with patch.object(determine_jobs, "should_run_clang_tidy") as mock:
|
|
yield mock
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_should_run_clang_format() -> Generator[Mock, None, None]:
|
|
"""Mock should_run_clang_format from helpers."""
|
|
with patch.object(determine_jobs, "should_run_clang_format") as mock:
|
|
yield mock
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_should_run_python_linters() -> Generator[Mock, None, None]:
|
|
"""Mock should_run_python_linters from helpers."""
|
|
with patch.object(determine_jobs, "should_run_python_linters") as mock:
|
|
yield mock
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_subprocess_run() -> Generator[Mock, None, None]:
|
|
"""Mock subprocess.run for list-components.py calls."""
|
|
with patch.object(determine_jobs.subprocess, "run") as mock:
|
|
yield mock
|
|
|
|
|
|
def test_main_all_tests_should_run(
|
|
mock_should_run_integration_tests: Mock,
|
|
mock_should_run_clang_tidy: Mock,
|
|
mock_should_run_clang_format: Mock,
|
|
mock_should_run_python_linters: Mock,
|
|
mock_subprocess_run: Mock,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
"""Test when all tests should run."""
|
|
mock_should_run_integration_tests.return_value = True
|
|
mock_should_run_clang_tidy.return_value = True
|
|
mock_should_run_clang_format.return_value = True
|
|
mock_should_run_python_linters.return_value = True
|
|
|
|
# Mock list-components.py output
|
|
mock_result = Mock()
|
|
mock_result.stdout = "wifi\napi\nsensor\n"
|
|
mock_subprocess_run.return_value = mock_result
|
|
|
|
# Run main function with mocked argv
|
|
with patch("sys.argv", ["determine-jobs.py"]):
|
|
determine_jobs.main()
|
|
|
|
# Check output
|
|
captured = capsys.readouterr()
|
|
output = json.loads(captured.out)
|
|
|
|
assert output["integration_tests"] is True
|
|
assert output["clang_tidy"] is True
|
|
assert output["clang_format"] is True
|
|
assert output["python_linters"] is True
|
|
assert output["changed_components"] == ["wifi", "api", "sensor"]
|
|
assert output["component_test_count"] == 3
|
|
|
|
|
|
def test_main_no_tests_should_run(
|
|
mock_should_run_integration_tests: Mock,
|
|
mock_should_run_clang_tidy: Mock,
|
|
mock_should_run_clang_format: Mock,
|
|
mock_should_run_python_linters: Mock,
|
|
mock_subprocess_run: Mock,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
"""Test when no tests should run."""
|
|
mock_should_run_integration_tests.return_value = False
|
|
mock_should_run_clang_tidy.return_value = False
|
|
mock_should_run_clang_format.return_value = False
|
|
mock_should_run_python_linters.return_value = False
|
|
|
|
# Mock empty list-components.py output
|
|
mock_result = Mock()
|
|
mock_result.stdout = ""
|
|
mock_subprocess_run.return_value = mock_result
|
|
|
|
# Run main function with mocked argv
|
|
with patch("sys.argv", ["determine-jobs.py"]):
|
|
determine_jobs.main()
|
|
|
|
# Check output
|
|
captured = capsys.readouterr()
|
|
output = json.loads(captured.out)
|
|
|
|
assert output["integration_tests"] is False
|
|
assert output["clang_tidy"] is False
|
|
assert output["clang_format"] is False
|
|
assert output["python_linters"] is False
|
|
assert output["changed_components"] == []
|
|
assert output["component_test_count"] == 0
|
|
|
|
|
|
def test_main_list_components_fails(
|
|
mock_should_run_integration_tests: Mock,
|
|
mock_should_run_clang_tidy: Mock,
|
|
mock_should_run_clang_format: Mock,
|
|
mock_should_run_python_linters: Mock,
|
|
mock_subprocess_run: Mock,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
"""Test when list-components.py fails."""
|
|
mock_should_run_integration_tests.return_value = True
|
|
mock_should_run_clang_tidy.return_value = True
|
|
mock_should_run_clang_format.return_value = True
|
|
mock_should_run_python_linters.return_value = True
|
|
|
|
# Mock list-components.py failure
|
|
mock_subprocess_run.side_effect = subprocess.CalledProcessError(1, "cmd")
|
|
|
|
# Run main function with mocked argv - should raise
|
|
with patch("sys.argv", ["determine-jobs.py"]):
|
|
with pytest.raises(subprocess.CalledProcessError):
|
|
determine_jobs.main()
|
|
|
|
|
|
def test_main_with_branch_argument(
|
|
mock_should_run_integration_tests: Mock,
|
|
mock_should_run_clang_tidy: Mock,
|
|
mock_should_run_clang_format: Mock,
|
|
mock_should_run_python_linters: Mock,
|
|
mock_subprocess_run: Mock,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
"""Test with branch argument."""
|
|
mock_should_run_integration_tests.return_value = False
|
|
mock_should_run_clang_tidy.return_value = True
|
|
mock_should_run_clang_format.return_value = False
|
|
mock_should_run_python_linters.return_value = True
|
|
|
|
# Mock list-components.py output
|
|
mock_result = Mock()
|
|
mock_result.stdout = "mqtt\n"
|
|
mock_subprocess_run.return_value = mock_result
|
|
|
|
with patch("sys.argv", ["script.py", "-b", "main"]):
|
|
determine_jobs.main()
|
|
|
|
# Check that functions were called with branch
|
|
mock_should_run_integration_tests.assert_called_once_with("main")
|
|
mock_should_run_clang_tidy.assert_called_once_with("main")
|
|
mock_should_run_clang_format.assert_called_once_with("main")
|
|
mock_should_run_python_linters.assert_called_once_with("main")
|
|
|
|
# Check that list-components.py was called with branch
|
|
mock_subprocess_run.assert_called_once()
|
|
call_args = mock_subprocess_run.call_args[0][0]
|
|
assert "--changed" in call_args
|
|
assert "-b" in call_args
|
|
assert "main" in call_args
|
|
|
|
# Check output
|
|
captured = capsys.readouterr()
|
|
output = json.loads(captured.out)
|
|
|
|
assert output["integration_tests"] is False
|
|
assert output["clang_tidy"] is True
|
|
assert output["clang_format"] is False
|
|
assert output["python_linters"] is True
|
|
assert output["changed_components"] == ["mqtt"]
|
|
assert output["component_test_count"] == 1
|
|
|
|
|
|
def test_should_run_integration_tests(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Test should_run_integration_tests function."""
|
|
# Core C++ files trigger tests
|
|
with patch.object(
|
|
determine_jobs, "changed_files", return_value=["esphome/core/component.cpp"]
|
|
):
|
|
result = determine_jobs.should_run_integration_tests()
|
|
assert result is True
|
|
|
|
# Core Python files trigger tests
|
|
with patch.object(
|
|
determine_jobs, "changed_files", return_value=["esphome/core/config.py"]
|
|
):
|
|
result = determine_jobs.should_run_integration_tests()
|
|
assert result is True
|
|
|
|
# Python files directly in esphome/ do NOT trigger tests
|
|
with patch.object(
|
|
determine_jobs, "changed_files", return_value=["esphome/config.py"]
|
|
):
|
|
result = determine_jobs.should_run_integration_tests()
|
|
assert result is False
|
|
|
|
# Python files in subdirectories (not core) do NOT trigger tests
|
|
with patch.object(
|
|
determine_jobs,
|
|
"changed_files",
|
|
return_value=["esphome/dashboard/web_server.py"],
|
|
):
|
|
result = determine_jobs.should_run_integration_tests()
|
|
assert result is False
|
|
|
|
|
|
def test_should_run_integration_tests_with_branch() -> None:
|
|
"""Test should_run_integration_tests with branch argument."""
|
|
with patch.object(determine_jobs, "changed_files") as mock_changed:
|
|
mock_changed.return_value = []
|
|
determine_jobs.should_run_integration_tests("release")
|
|
mock_changed.assert_called_once_with("release")
|
|
|
|
|
|
def test_should_run_integration_tests_component_dependency() -> None:
|
|
"""Test that integration tests run when components used in fixtures change."""
|
|
with patch.object(
|
|
determine_jobs, "changed_files", return_value=["esphome/components/api/api.cpp"]
|
|
):
|
|
with patch.object(
|
|
determine_jobs, "get_components_from_integration_fixtures"
|
|
) as mock_fixtures:
|
|
mock_fixtures.return_value = {"api", "sensor"}
|
|
with patch.object(determine_jobs, "get_all_dependencies") as mock_deps:
|
|
mock_deps.return_value = {"api", "sensor", "network"}
|
|
result = determine_jobs.should_run_integration_tests()
|
|
assert result is True
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("check_returncode", "changed_files", "expected_result"),
|
|
[
|
|
(0, [], True), # Hash changed - need full scan
|
|
(1, ["esphome/core.cpp"], True), # C++ file changed
|
|
(1, ["README.md"], False), # No C++ files changed
|
|
],
|
|
)
|
|
def test_should_run_clang_tidy(
|
|
check_returncode: int,
|
|
changed_files: list[str],
|
|
expected_result: bool,
|
|
) -> None:
|
|
"""Test should_run_clang_tidy function."""
|
|
with patch.object(determine_jobs, "changed_files", return_value=changed_files):
|
|
# Test with hash check returning specific code
|
|
with patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = Mock(returncode=check_returncode)
|
|
result = determine_jobs.should_run_clang_tidy()
|
|
assert result == expected_result
|
|
|
|
# Test with hash check failing (exception)
|
|
if check_returncode != 0:
|
|
with patch("subprocess.run", side_effect=Exception("Failed")):
|
|
result = determine_jobs.should_run_clang_tidy()
|
|
assert result is True # Fail safe - run clang-tidy
|
|
|
|
|
|
def test_should_run_clang_tidy_with_branch() -> None:
|
|
"""Test should_run_clang_tidy with branch argument."""
|
|
with patch.object(determine_jobs, "changed_files") as mock_changed:
|
|
mock_changed.return_value = []
|
|
with patch("subprocess.run") as mock_run:
|
|
mock_run.return_value = Mock(returncode=1) # Hash unchanged
|
|
determine_jobs.should_run_clang_tidy("release")
|
|
mock_changed.assert_called_once_with("release")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("changed_files", "expected_result"),
|
|
[
|
|
(["esphome/core.py"], True),
|
|
(["script/test.py"], True),
|
|
(["esphome/test.pyi"], True), # .pyi files should trigger
|
|
(["README.md"], False),
|
|
([], False),
|
|
],
|
|
)
|
|
def test_should_run_python_linters(
|
|
changed_files: list[str], expected_result: bool
|
|
) -> None:
|
|
"""Test should_run_python_linters function."""
|
|
with patch.object(determine_jobs, "changed_files", return_value=changed_files):
|
|
result = determine_jobs.should_run_python_linters()
|
|
assert result == expected_result
|
|
|
|
|
|
def test_should_run_python_linters_with_branch() -> None:
|
|
"""Test should_run_python_linters with branch argument."""
|
|
with patch.object(determine_jobs, "changed_files") as mock_changed:
|
|
mock_changed.return_value = []
|
|
determine_jobs.should_run_python_linters("release")
|
|
mock_changed.assert_called_once_with("release")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("changed_files", "expected_result"),
|
|
[
|
|
(["esphome/core.cpp"], True),
|
|
(["esphome/core.h"], True),
|
|
(["test.hpp"], True),
|
|
(["test.cc"], True),
|
|
(["test.cxx"], True),
|
|
(["test.c"], True),
|
|
(["test.tcc"], True),
|
|
(["README.md"], False),
|
|
([], False),
|
|
],
|
|
)
|
|
def test_should_run_clang_format(
|
|
changed_files: list[str], expected_result: bool
|
|
) -> None:
|
|
"""Test should_run_clang_format function."""
|
|
with patch.object(determine_jobs, "changed_files", return_value=changed_files):
|
|
result = determine_jobs.should_run_clang_format()
|
|
assert result == expected_result
|
|
|
|
|
|
def test_should_run_clang_format_with_branch() -> None:
|
|
"""Test should_run_clang_format with branch argument."""
|
|
with patch.object(determine_jobs, "changed_files") as mock_changed:
|
|
mock_changed.return_value = []
|
|
determine_jobs.should_run_clang_format("release")
|
|
mock_changed.assert_called_once_with("release")
|