mirror of
https://github.com/esphome/esphome.git
synced 2026-08-31 18:16:03 +00:00
J. Nick Koston
e907e71101
- Add `--` terminator before submodule paths (both clone and refresh paths) so a path beginning with `-` cannot be parsed as a git option. - Reword the refresh-fetch comment: this fetch also runs when ref is None, in which case it pulls the remote default branch.
975 lines
32 KiB
Python
975 lines
32 KiB
Python
"""Tests for git.py module."""
|
|
|
|
from datetime import datetime, timedelta
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
from esphome import git
|
|
from esphome.core import CORE, TimePeriodSeconds
|
|
from esphome.git import GitCommandError
|
|
|
|
|
|
def _compute_repo_dir(url: str, ref: str | None, domain: str) -> Path:
|
|
"""Helper to compute the expected repo directory path using git module's logic."""
|
|
key = f"{url}@{ref}"
|
|
return git._compute_destination_path(key, domain)
|
|
|
|
|
|
def _setup_old_repo(repo_dir: Path, days_old: int = 2) -> None:
|
|
"""Helper to set up a git repo directory structure with an old timestamp.
|
|
|
|
Args:
|
|
repo_dir: The repository directory path to create.
|
|
days_old: Number of days old to make the FETCH_HEAD file (default: 2).
|
|
"""
|
|
# Create repo directory
|
|
repo_dir.mkdir(parents=True)
|
|
git_dir = repo_dir / ".git"
|
|
git_dir.mkdir()
|
|
|
|
# Create FETCH_HEAD file with old timestamp
|
|
fetch_head = git_dir / "FETCH_HEAD"
|
|
fetch_head.write_text("test")
|
|
old_time = datetime.now() - timedelta(days=days_old)
|
|
fetch_head.touch()
|
|
os.utime(fetch_head, (old_time.timestamp(), old_time.timestamp()))
|
|
|
|
|
|
def _get_git_command_type(cmd: list[str]) -> str | None:
|
|
"""Helper to determine the type of git command from a command list.
|
|
|
|
Args:
|
|
cmd: The git command list (e.g., ["git", "rev-parse", "HEAD"]).
|
|
|
|
Returns:
|
|
The command type ("rev-parse", "stash", "fetch", "reset", "clone") or None.
|
|
"""
|
|
# Git commands are always in format ["git", "command", ...], so check index 1
|
|
if len(cmd) > 1:
|
|
return cmd[1]
|
|
return None
|
|
|
|
|
|
def test_run_git_command_success(tmp_path: Path) -> None:
|
|
"""Test that run_git_command returns output on success."""
|
|
# Create a simple git repo to test with
|
|
repo_dir = tmp_path / "test_repo"
|
|
repo_dir.mkdir()
|
|
|
|
# Initialize a git repo
|
|
result = git.run_git_command(["git", "init"], str(repo_dir))
|
|
assert "Initialized empty Git repository" in result or result == ""
|
|
|
|
# Verify we can run a command and get output
|
|
result = git.run_git_command(["git", "status", "--porcelain"], str(repo_dir))
|
|
# Empty repo should have empty status
|
|
assert isinstance(result, str)
|
|
|
|
|
|
def test_run_git_command_with_git_dir_isolation(
|
|
tmp_path: Path, mock_subprocess_run: Mock
|
|
) -> None:
|
|
"""Test that git_dir parameter properly isolates git operations."""
|
|
repo_dir = tmp_path / "test_repo"
|
|
repo_dir.mkdir()
|
|
git_dir = repo_dir / ".git"
|
|
git_dir.mkdir()
|
|
|
|
# Configure mock to return success
|
|
mock_subprocess_run.return_value = Mock(
|
|
returncode=0,
|
|
stdout=b"test output",
|
|
stderr=b"",
|
|
)
|
|
|
|
result = git.run_git_command(
|
|
["git", "rev-parse", "HEAD"],
|
|
git_dir=repo_dir,
|
|
)
|
|
|
|
# Verify subprocess.run was called
|
|
assert mock_subprocess_run.called
|
|
call_args = mock_subprocess_run.call_args
|
|
|
|
# Verify environment was set
|
|
env = call_args[1]["env"]
|
|
assert "GIT_DIR" in env
|
|
assert "GIT_WORK_TREE" in env
|
|
assert env["GIT_DIR"] == str(repo_dir / ".git")
|
|
assert env["GIT_WORK_TREE"] == str(repo_dir)
|
|
|
|
assert result == "test output"
|
|
|
|
|
|
def test_run_git_command_raises_git_not_installed_error(
|
|
tmp_path: Path, mock_subprocess_run: Mock
|
|
) -> None:
|
|
"""Test that FileNotFoundError is converted to GitNotInstalledError."""
|
|
from esphome.git import GitNotInstalledError
|
|
|
|
repo_dir = tmp_path / "test_repo"
|
|
|
|
# Configure mock to raise FileNotFoundError
|
|
mock_subprocess_run.side_effect = FileNotFoundError("git not found")
|
|
|
|
with pytest.raises(GitNotInstalledError, match="git is not installed"):
|
|
git.run_git_command(["git", "status"], git_dir=repo_dir)
|
|
|
|
|
|
def test_run_git_command_raises_git_command_error_on_failure(
|
|
tmp_path: Path, mock_subprocess_run: Mock
|
|
) -> None:
|
|
"""Test that failed git commands raise GitCommandError."""
|
|
repo_dir = tmp_path / "test_repo"
|
|
|
|
# Configure mock to return non-zero exit code
|
|
mock_subprocess_run.return_value = Mock(
|
|
returncode=1,
|
|
stdout=b"",
|
|
stderr=b"fatal: not a git repository",
|
|
)
|
|
|
|
with pytest.raises(GitCommandError, match="not a git repository"):
|
|
git.run_git_command(["git", "status"], git_dir=repo_dir)
|
|
|
|
|
|
def test_run_git_command_strips_fatal_prefix(
|
|
tmp_path: Path, mock_subprocess_run: Mock
|
|
) -> None:
|
|
"""Test that 'fatal: ' prefix is stripped from error messages."""
|
|
repo_dir = tmp_path / "test_repo"
|
|
|
|
# Configure mock to return error with "fatal: " prefix
|
|
mock_subprocess_run.return_value = Mock(
|
|
returncode=128,
|
|
stdout=b"",
|
|
stderr=b"fatal: repository not found\n",
|
|
)
|
|
|
|
with pytest.raises(GitCommandError) as exc_info:
|
|
git.run_git_command(["git", "clone", "invalid-url"], git_dir=repo_dir)
|
|
|
|
# Error message should NOT include "fatal: " prefix
|
|
assert "fatal:" not in str(exc_info.value)
|
|
assert "repository not found" in str(exc_info.value)
|
|
|
|
|
|
def test_run_git_command_without_git_dir(mock_subprocess_run: Mock) -> None:
|
|
"""Test that run_git_command works without git_dir (clone case)."""
|
|
# Configure mock to return success
|
|
mock_subprocess_run.return_value = Mock(
|
|
returncode=0,
|
|
stdout=b"Cloning into 'test_repo'...",
|
|
stderr=b"",
|
|
)
|
|
|
|
result = git.run_git_command(["git", "clone", "https://github.com/test/repo"])
|
|
|
|
# Verify subprocess.run was called
|
|
assert mock_subprocess_run.called
|
|
call_args = mock_subprocess_run.call_args
|
|
|
|
# Verify environment does NOT have GIT_DIR or GIT_WORK_TREE set
|
|
# (it should use the default environment or None)
|
|
env = call_args[1].get("env")
|
|
if env is not None:
|
|
assert "GIT_DIR" not in env
|
|
assert "GIT_WORK_TREE" not in env
|
|
|
|
# Verify cwd is None (default)
|
|
assert call_args[1].get("cwd") is None
|
|
|
|
assert result == "Cloning into 'test_repo'..."
|
|
|
|
|
|
def test_run_git_command_without_git_dir_raises_error(
|
|
mock_subprocess_run: Mock,
|
|
) -> None:
|
|
"""Test that run_git_command without git_dir can still raise errors."""
|
|
# Configure mock to return error
|
|
mock_subprocess_run.return_value = Mock(
|
|
returncode=128,
|
|
stdout=b"",
|
|
stderr=b"fatal: repository not found\n",
|
|
)
|
|
|
|
with pytest.raises(GitCommandError, match="repository not found"):
|
|
git.run_git_command(["git", "clone", "https://invalid.url/repo.git"])
|
|
|
|
|
|
def test_clone_or_update_with_never_refresh(
|
|
tmp_path: Path, mock_run_git_command: Mock
|
|
) -> None:
|
|
"""Test that NEVER_REFRESH skips updates for existing repos."""
|
|
# Set up CORE.config_path so data_dir uses tmp_path
|
|
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)
|
|
|
|
# Create the git repo directory structure
|
|
repo_dir.mkdir(parents=True)
|
|
git_dir = repo_dir / ".git"
|
|
git_dir.mkdir()
|
|
|
|
# Create FETCH_HEAD file with current timestamp
|
|
fetch_head = git_dir / "FETCH_HEAD"
|
|
fetch_head.write_text("test")
|
|
|
|
# Call with NEVER_REFRESH
|
|
result_dir, revert = git.clone_or_update(
|
|
url=url,
|
|
ref=ref,
|
|
refresh=git.NEVER_REFRESH,
|
|
domain=domain,
|
|
)
|
|
|
|
# Should NOT call git commands since NEVER_REFRESH and repo exists
|
|
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:
|
|
"""Test that refresh triggers update for old repos."""
|
|
# Set up CORE.config_path so data_dir uses tmp_path
|
|
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)
|
|
|
|
# Create the git repo directory structure
|
|
repo_dir.mkdir(parents=True)
|
|
git_dir = repo_dir / ".git"
|
|
git_dir.mkdir()
|
|
|
|
# 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
|
|
os.utime(fetch_head, (old_time.timestamp(), old_time.timestamp()))
|
|
|
|
# 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=url,
|
|
ref=ref,
|
|
refresh=refresh,
|
|
domain=domain,
|
|
)
|
|
|
|
# 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
|
|
|
|
|
|
def test_clone_or_update_with_refresh_skips_fresh_repo(
|
|
tmp_path: Path, mock_run_git_command: Mock
|
|
) -> None:
|
|
"""Test that refresh doesn't update fresh repos."""
|
|
# Set up CORE.config_path so data_dir uses tmp_path
|
|
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)
|
|
|
|
# Create the git repo directory structure
|
|
repo_dir.mkdir(parents=True)
|
|
git_dir = repo_dir / ".git"
|
|
git_dir.mkdir()
|
|
|
|
# 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
|
|
os.utime(fetch_head, (recent_time.timestamp(), recent_time.timestamp()))
|
|
|
|
# Call with refresh=1d (1 day)
|
|
refresh = TimePeriodSeconds(days=1)
|
|
result_dir, revert = git.clone_or_update(
|
|
url=url,
|
|
ref=ref,
|
|
refresh=refresh,
|
|
domain=domain,
|
|
)
|
|
|
|
# Should NOT call git fetch since repo is fresh
|
|
mock_run_git_command.assert_not_called()
|
|
assert result_dir == repo_dir
|
|
assert revert is None
|
|
|
|
|
|
def test_clone_or_update_clones_missing_repo(
|
|
tmp_path: Path, mock_run_git_command: Mock
|
|
) -> None:
|
|
"""Test that missing repos are cloned regardless of refresh setting."""
|
|
# Set up CORE.config_path so data_dir uses tmp_path
|
|
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)
|
|
|
|
# Create base directory but NOT the repo itself
|
|
base_dir = tmp_path / ".esphome" / domain
|
|
base_dir.mkdir(parents=True)
|
|
# repo_dir should NOT exist
|
|
assert not repo_dir.exists()
|
|
|
|
# Test with NEVER_REFRESH - should still clone since repo doesn't exist
|
|
result_dir, revert = git.clone_or_update(
|
|
url=url,
|
|
ref=ref,
|
|
refresh=git.NEVER_REFRESH,
|
|
domain=domain,
|
|
)
|
|
|
|
# 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
|
|
|
|
|
|
def test_clone_or_update_with_none_refresh_always_updates(
|
|
tmp_path: Path, mock_run_git_command: Mock
|
|
) -> None:
|
|
"""Test that refresh=None always updates existing repos."""
|
|
# Set up CORE.config_path so data_dir uses tmp_path
|
|
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)
|
|
|
|
# Create the git repo directory structure
|
|
repo_dir.mkdir(parents=True)
|
|
git_dir = repo_dir / ".git"
|
|
git_dir.mkdir()
|
|
|
|
# Create FETCH_HEAD file with very recent timestamp (1 second ago)
|
|
fetch_head = git_dir / "FETCH_HEAD"
|
|
fetch_head.write_text("test")
|
|
recent_time = datetime.now() - timedelta(seconds=1)
|
|
fetch_head.touch() # Create the file
|
|
# Set modification time to 1 second ago
|
|
os.utime(fetch_head, (recent_time.timestamp(), recent_time.timestamp()))
|
|
|
|
# Mock git command responses
|
|
mock_run_git_command.return_value = "abc123" # SHA for rev-parse
|
|
|
|
# Call with refresh=None (default behavior)
|
|
result_dir, revert = git.clone_or_update(
|
|
url=url,
|
|
ref=ref,
|
|
refresh=None,
|
|
domain=domain,
|
|
)
|
|
|
|
# Should call git fetch and update commands since refresh=None means always update
|
|
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
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("fail_command", "error_message"),
|
|
[
|
|
(
|
|
"rev-parse",
|
|
"ambiguous argument 'HEAD': unknown revision or path not in the working tree.",
|
|
),
|
|
("stash", "fatal: unable to write new index file"),
|
|
(
|
|
"fetch",
|
|
"fatal: unable to access 'https://github.com/test/repo/': Could not resolve host",
|
|
),
|
|
("reset", "fatal: Could not reset index file to revision 'FETCH_HEAD'"),
|
|
],
|
|
)
|
|
def test_clone_or_update_recovers_from_git_failures(
|
|
tmp_path: Path, mock_run_git_command: Mock, fail_command: str, error_message: str
|
|
) -> None:
|
|
"""Test that repos are re-cloned when various git commands fail."""
|
|
# Set up CORE.config_path so data_dir uses tmp_path
|
|
CORE.config_path = tmp_path / "test.yaml"
|
|
|
|
url = "https://github.com/test/repo"
|
|
ref = "main"
|
|
domain = "test"
|
|
repo_dir = _compute_repo_dir(url, ref, domain)
|
|
|
|
# Use helper to set up old repo
|
|
_setup_old_repo(repo_dir)
|
|
|
|
# Track command call counts to make first call fail, subsequent calls succeed
|
|
call_counts: dict[str, int] = {}
|
|
|
|
def git_command_side_effect(
|
|
cmd: list[str], cwd: str | None = None, **kwargs: Any
|
|
) -> str:
|
|
# Determine which command this is
|
|
cmd_type = _get_git_command_type(cmd)
|
|
|
|
# Track call count for this command type
|
|
if cmd_type:
|
|
call_counts[cmd_type] = call_counts.get(cmd_type, 0) + 1
|
|
|
|
# Fail on first call to the specified command, succeed on subsequent calls
|
|
if cmd_type == fail_command and call_counts[cmd_type] == 1:
|
|
raise GitCommandError(error_message)
|
|
|
|
# Default successful responses
|
|
if cmd_type == "rev-parse":
|
|
return "abc123"
|
|
return ""
|
|
|
|
mock_run_git_command.side_effect = git_command_side_effect
|
|
|
|
refresh = TimePeriodSeconds(days=1)
|
|
result_dir, revert = git.clone_or_update(
|
|
url=url,
|
|
ref=ref,
|
|
refresh=refresh,
|
|
domain=domain,
|
|
)
|
|
|
|
# Verify recovery happened
|
|
call_list = mock_run_git_command.call_args_list
|
|
|
|
# Should have attempted the failing command
|
|
assert any(fail_command in str(c) for c in call_list)
|
|
|
|
# Should have called clone for recovery
|
|
assert any("clone" in str(c) for c in call_list)
|
|
|
|
# Verify the repo directory path is returned
|
|
assert result_dir == repo_dir
|
|
|
|
|
|
def test_clone_or_update_fails_when_recovery_also_fails(
|
|
tmp_path: Path, mock_run_git_command: Mock
|
|
) -> None:
|
|
"""Test that we don't infinitely recurse when recovery also fails."""
|
|
# Set up CORE.config_path so data_dir uses tmp_path
|
|
CORE.config_path = tmp_path / "test.yaml"
|
|
|
|
url = "https://github.com/test/repo"
|
|
ref = "main"
|
|
domain = "test"
|
|
repo_dir = _compute_repo_dir(url, ref, domain)
|
|
|
|
# Use helper to set up old repo
|
|
_setup_old_repo(repo_dir)
|
|
|
|
# Mock git command to fail on clone (simulating network failure during recovery)
|
|
def git_command_side_effect(
|
|
cmd: list[str], cwd: str | None = None, **kwargs: Any
|
|
) -> str:
|
|
cmd_type = _get_git_command_type(cmd)
|
|
if cmd_type == "rev-parse":
|
|
# First time fails (broken repo)
|
|
raise GitCommandError(
|
|
"ambiguous argument 'HEAD': unknown revision or path not in the working tree."
|
|
)
|
|
if cmd_type == "clone":
|
|
# Clone also fails (recovery fails)
|
|
raise GitCommandError("fatal: unable to access repository")
|
|
return ""
|
|
|
|
mock_run_git_command.side_effect = git_command_side_effect
|
|
|
|
refresh = TimePeriodSeconds(days=1)
|
|
|
|
# Should raise after one recovery attempt fails
|
|
with pytest.raises(GitCommandError, match="fatal: unable to access repository"):
|
|
git.clone_or_update(
|
|
url=url,
|
|
ref=ref,
|
|
refresh=refresh,
|
|
domain=domain,
|
|
)
|
|
|
|
# Verify we only tried to clone once (no infinite recursion)
|
|
call_list = mock_run_git_command.call_args_list
|
|
clone_calls = [c for c in call_list if "clone" in c[0][0]]
|
|
# Should have exactly one clone call (the recovery attempt that failed)
|
|
assert len(clone_calls) == 1
|
|
# Should have tried rev-parse once (which failed and triggered recovery)
|
|
rev_parse_calls = [c for c in call_list if "rev-parse" in c[0][0]]
|
|
assert len(rev_parse_calls) == 1
|
|
|
|
|
|
def test_clone_or_update_recover_broken_flag_prevents_second_recovery(
|
|
tmp_path: Path, mock_run_git_command: Mock
|
|
) -> None:
|
|
"""Test that _recover_broken=False prevents a second recovery attempt (tests the raise path)."""
|
|
# Set up CORE.config_path so data_dir uses tmp_path
|
|
CORE.config_path = tmp_path / "test.yaml"
|
|
|
|
url = "https://github.com/test/repo"
|
|
ref = "main"
|
|
domain = "test"
|
|
repo_dir = _compute_repo_dir(url, ref, domain)
|
|
|
|
# Use helper to set up old repo
|
|
_setup_old_repo(repo_dir)
|
|
|
|
# Track fetch calls to differentiate between first (in clone) and second (in recovery update)
|
|
call_counts: dict[str, int] = {}
|
|
|
|
# Mock git command to fail on fetch during recovery's ref checkout
|
|
def git_command_side_effect(
|
|
cmd: list[str], cwd: str | None = None, **kwargs: Any
|
|
) -> str:
|
|
cmd_type = _get_git_command_type(cmd)
|
|
|
|
if cmd_type:
|
|
call_counts[cmd_type] = call_counts.get(cmd_type, 0) + 1
|
|
|
|
# First attempt: rev-parse fails (broken repo)
|
|
if cmd_type == "rev-parse" and call_counts[cmd_type] == 1:
|
|
raise GitCommandError(
|
|
"ambiguous argument 'HEAD': unknown revision or path not in the working tree."
|
|
)
|
|
|
|
# Recovery: clone succeeds
|
|
if cmd_type == "clone":
|
|
return ""
|
|
|
|
# Recovery: fetch for ref checkout fails
|
|
# This happens in the clone path when ref is not None (line 80 in git.py)
|
|
if cmd_type == "fetch" and call_counts[cmd_type] == 1:
|
|
raise GitCommandError("fatal: couldn't find remote ref main")
|
|
|
|
# Default success
|
|
return "abc123" if cmd_type == "rev-parse" else ""
|
|
|
|
mock_run_git_command.side_effect = git_command_side_effect
|
|
|
|
refresh = TimePeriodSeconds(days=1)
|
|
|
|
# Should raise on the fetch during recovery (when _recover_broken=False)
|
|
# This tests the critical "if not _recover_broken: raise" path
|
|
with pytest.raises(GitCommandError, match="fatal: couldn't find remote ref main"):
|
|
git.clone_or_update(
|
|
url=url,
|
|
ref=ref,
|
|
refresh=refresh,
|
|
domain=domain,
|
|
)
|
|
|
|
# Verify the sequence of events
|
|
call_list = mock_run_git_command.call_args_list
|
|
|
|
# Should have: rev-parse (fail, triggers recovery), clone (success),
|
|
# fetch (fail during ref checkout, raises because _recover_broken=False)
|
|
rev_parse_calls = [c for c in call_list if "rev-parse" in c[0][0]]
|
|
# Should have exactly one rev-parse call that failed
|
|
assert len(rev_parse_calls) == 1
|
|
|
|
clone_calls = [c for c in call_list if "clone" in c[0][0]]
|
|
# Should have exactly one clone call (the recovery attempt)
|
|
assert len(clone_calls) == 1
|
|
|
|
fetch_calls = [c for c in call_list if "fetch" in c[0][0]]
|
|
# Should have exactly one fetch call that failed (during ref checkout in recovery)
|
|
assert len(fetch_calls) == 1
|
|
|
|
|
|
def test_clone_or_update_recover_broken_flag_prevents_infinite_loop(
|
|
tmp_path: Path, mock_run_git_command: Mock
|
|
) -> None:
|
|
"""Test that _recover_broken=False prevents infinite recursion when repo persists."""
|
|
# This tests the critical "if not _recover_broken: raise" path at line 124-125
|
|
# Set up CORE.config_path so data_dir uses tmp_path
|
|
CORE.config_path = tmp_path / "test.yaml"
|
|
|
|
url = "https://github.com/test/repo"
|
|
ref = "main"
|
|
domain = "test"
|
|
repo_dir = _compute_repo_dir(url, ref, domain)
|
|
|
|
# Use helper to set up old repo
|
|
_setup_old_repo(repo_dir)
|
|
|
|
# Mock shutil.rmtree to NOT actually delete the directory
|
|
# This simulates a scenario where deletion fails (permissions, etc.)
|
|
import unittest.mock
|
|
|
|
def mock_rmtree(path, *args, **kwargs):
|
|
# Don't actually delete - this causes the recursive call to still see the repo
|
|
pass
|
|
|
|
# Mock git commands to always fail on stash
|
|
def git_command_side_effect(
|
|
cmd: list[str], cwd: str | None = None, **kwargs: Any
|
|
) -> str:
|
|
cmd_type = _get_git_command_type(cmd)
|
|
if cmd_type == "rev-parse":
|
|
return "abc123"
|
|
if cmd_type == "stash":
|
|
# Always fails
|
|
raise GitCommandError("fatal: unable to write new index file")
|
|
return ""
|
|
|
|
mock_run_git_command.side_effect = git_command_side_effect
|
|
|
|
refresh = TimePeriodSeconds(days=1)
|
|
|
|
# Mock shutil.rmtree and test
|
|
# Should raise on the second attempt when _recover_broken=False
|
|
# This hits the "if not _recover_broken: raise" path
|
|
with (
|
|
unittest.mock.patch("esphome.git.rmtree", side_effect=mock_rmtree),
|
|
pytest.raises(GitCommandError, match="fatal: unable to write new index file"),
|
|
):
|
|
git.clone_or_update(
|
|
url=url,
|
|
ref=ref,
|
|
refresh=refresh,
|
|
domain=domain,
|
|
)
|
|
|
|
# Verify the sequence: stash fails twice (once triggering recovery, once raising)
|
|
call_list = mock_run_git_command.call_args_list
|
|
stash_calls = [c for c in call_list if "stash" in c[0][0]]
|
|
# Should have exactly two stash calls
|
|
assert len(stash_calls) == 2
|
|
|
|
|
|
def test_clone_or_update_cleans_up_on_failed_ref_fetch(
|
|
tmp_path: Path, mock_run_git_command: Mock
|
|
) -> None:
|
|
"""Test that a failed ref fetch removes the incomplete clone directory.
|
|
|
|
When cloning with a specific ref, if `git clone` succeeds but the
|
|
subsequent `git fetch <ref>` fails, the clone directory should be
|
|
removed so the next attempt starts fresh instead of finding a stale
|
|
clone on the default branch.
|
|
"""
|
|
CORE.config_path = tmp_path / "test.yaml"
|
|
|
|
url = "https://github.com/test/repo"
|
|
ref = "pull/123/head"
|
|
domain = "test"
|
|
repo_dir = _compute_repo_dir(url, ref, domain)
|
|
|
|
def git_command_side_effect(
|
|
cmd: list[str], cwd: str | None = None, **kwargs: Any
|
|
) -> str:
|
|
cmd_type = _get_git_command_type(cmd)
|
|
if cmd_type == "clone":
|
|
# Simulate successful clone by creating the directory
|
|
repo_dir.mkdir(parents=True, exist_ok=True)
|
|
(repo_dir / ".git").mkdir(exist_ok=True)
|
|
return ""
|
|
if cmd_type == "fetch":
|
|
raise GitCommandError("fatal: couldn't find remote ref pull/123/head")
|
|
return ""
|
|
|
|
mock_run_git_command.side_effect = git_command_side_effect
|
|
|
|
refresh = TimePeriodSeconds(days=1)
|
|
|
|
with pytest.raises(GitCommandError, match="couldn't find remote ref"):
|
|
git.clone_or_update(
|
|
url=url,
|
|
ref=ref,
|
|
refresh=refresh,
|
|
domain=domain,
|
|
)
|
|
|
|
# The incomplete clone directory should have been removed
|
|
assert not repo_dir.exists()
|
|
|
|
# Verify clone was attempted then fetch failed
|
|
call_list = mock_run_git_command.call_args_list
|
|
clone_calls = [c for c in call_list if "clone" in c[0][0]]
|
|
assert len(clone_calls) == 1
|
|
fetch_calls = [c for c in call_list if "fetch" in c[0][0]]
|
|
assert len(fetch_calls) == 1
|
|
|
|
|
|
def test_clone_or_update_stale_clone_is_retried_after_cleanup(
|
|
tmp_path: Path, mock_run_git_command: Mock
|
|
) -> None:
|
|
"""Test that after cleanup, a subsequent call does a fresh clone.
|
|
|
|
This is the full scenario: first call fails at fetch (directory cleaned up),
|
|
second call sees no directory and clones fresh.
|
|
"""
|
|
CORE.config_path = tmp_path / "test.yaml"
|
|
|
|
url = "https://github.com/test/repo"
|
|
ref = "pull/123/head"
|
|
domain = "test"
|
|
repo_dir = _compute_repo_dir(url, ref, domain)
|
|
|
|
call_count = {"clone": 0, "fetch": 0}
|
|
|
|
def git_command_side_effect(
|
|
cmd: list[str], cwd: str | None = None, **kwargs: Any
|
|
) -> str:
|
|
cmd_type = _get_git_command_type(cmd)
|
|
if cmd_type == "clone":
|
|
call_count["clone"] += 1
|
|
repo_dir.mkdir(parents=True, exist_ok=True)
|
|
(repo_dir / ".git").mkdir(exist_ok=True)
|
|
return ""
|
|
if cmd_type == "fetch":
|
|
call_count["fetch"] += 1
|
|
if call_count["fetch"] == 1:
|
|
# First fetch fails
|
|
raise GitCommandError("fatal: couldn't find remote ref pull/123/head")
|
|
# Second fetch succeeds
|
|
return ""
|
|
if cmd_type == "reset":
|
|
return ""
|
|
return ""
|
|
|
|
mock_run_git_command.side_effect = git_command_side_effect
|
|
|
|
refresh = TimePeriodSeconds(days=1)
|
|
|
|
# First call: clone succeeds, fetch fails, directory cleaned up
|
|
with pytest.raises(GitCommandError, match="couldn't find remote ref"):
|
|
git.clone_or_update(url=url, ref=ref, refresh=refresh, domain=domain)
|
|
|
|
assert not repo_dir.exists()
|
|
|
|
# Second call: fresh clone + fetch succeeds
|
|
result_dir, _ = git.clone_or_update(
|
|
url=url, ref=ref, refresh=refresh, domain=domain
|
|
)
|
|
|
|
assert result_dir == repo_dir
|
|
assert repo_dir.exists()
|
|
assert call_count["clone"] == 2
|
|
assert call_count["fetch"] == 2
|
|
|
|
|
|
def test_clone_with_ref_uses_shallow_fetch(
|
|
tmp_path: Path, mock_run_git_command: Mock
|
|
) -> None:
|
|
"""Clone with a ref should use --depth=1 on both clone and fetch."""
|
|
CORE.config_path = tmp_path / "test.yaml"
|
|
|
|
url = "https://github.com/test/repo"
|
|
ref = "pull/123/head"
|
|
domain = "test"
|
|
repo_dir = _compute_repo_dir(url, ref, domain)
|
|
|
|
def git_command_side_effect(
|
|
cmd: list[str], cwd: str | None = None, **kwargs: Any
|
|
) -> str:
|
|
if _get_git_command_type(cmd) == "clone":
|
|
repo_dir.mkdir(parents=True, exist_ok=True)
|
|
(repo_dir / ".git").mkdir(exist_ok=True)
|
|
return ""
|
|
|
|
mock_run_git_command.side_effect = git_command_side_effect
|
|
|
|
git.clone_or_update(url=url, ref=ref, refresh=None, domain=domain)
|
|
|
|
call_list = mock_run_git_command.call_args_list
|
|
|
|
clone_calls = [c for c in call_list if "clone" in c[0][0]]
|
|
assert len(clone_calls) == 1
|
|
assert "--depth=1" in clone_calls[0][0][0]
|
|
|
|
fetch_calls = [c for c in call_list if "fetch" in c[0][0]]
|
|
assert len(fetch_calls) == 1
|
|
assert "--depth=1" in fetch_calls[0][0][0]
|
|
# Ref must still be passed so the requested commit/branch is fetched.
|
|
assert ref in fetch_calls[0][0][0]
|
|
|
|
|
|
def test_clone_with_submodules_uses_shallow_submodule_update(
|
|
tmp_path: Path, mock_run_git_command: Mock
|
|
) -> None:
|
|
"""Submodule init on a fresh clone should use --depth=1."""
|
|
CORE.config_path = tmp_path / "test.yaml"
|
|
|
|
url = "https://github.com/test/repo"
|
|
domain = "test"
|
|
repo_dir = _compute_repo_dir(url, None, domain)
|
|
|
|
def git_command_side_effect(
|
|
cmd: list[str], cwd: str | None = None, **kwargs: Any
|
|
) -> str:
|
|
if _get_git_command_type(cmd) == "clone":
|
|
repo_dir.mkdir(parents=True, exist_ok=True)
|
|
(repo_dir / ".git").mkdir(exist_ok=True)
|
|
return ""
|
|
|
|
mock_run_git_command.side_effect = git_command_side_effect
|
|
|
|
git.clone_or_update(
|
|
url=url,
|
|
ref=None,
|
|
refresh=None,
|
|
domain=domain,
|
|
submodules=["components/foo"],
|
|
)
|
|
|
|
submodule_calls = [
|
|
c for c in mock_run_git_command.call_args_list if "submodule" in c[0][0]
|
|
]
|
|
assert len(submodule_calls) == 1
|
|
cmd = submodule_calls[0][0][0]
|
|
assert "--depth=1" in cmd
|
|
assert "components/foo" in cmd
|
|
# The `--` terminator must precede the submodule paths so a path
|
|
# beginning with `-` cannot be parsed as an option.
|
|
assert cmd.index("--") < cmd.index("components/foo")
|
|
|
|
|
|
def test_refresh_fetch_is_shallow(tmp_path: Path, mock_run_git_command: Mock) -> None:
|
|
"""The refresh-path fetch should use --depth=1."""
|
|
CORE.config_path = tmp_path / "test.yaml"
|
|
|
|
url = "https://github.com/test/repo"
|
|
ref = "main"
|
|
domain = "test"
|
|
repo_dir = _compute_repo_dir(url, ref, domain)
|
|
|
|
_setup_old_repo(repo_dir)
|
|
mock_run_git_command.return_value = "abc123"
|
|
|
|
git.clone_or_update(
|
|
url=url, ref=ref, refresh=TimePeriodSeconds(days=1), domain=domain
|
|
)
|
|
|
|
fetch_calls = [c for c in mock_run_git_command.call_args_list if "fetch" in c[0][0]]
|
|
assert len(fetch_calls) == 1
|
|
cmd = fetch_calls[0][0][0]
|
|
assert "--depth=1" in cmd
|
|
# Ref must still be in the refresh fetch so the right tip is updated.
|
|
assert cmd[-1] == ref
|
|
|
|
|
|
def test_refresh_submodule_update_is_shallow(
|
|
tmp_path: Path, mock_run_git_command: Mock
|
|
) -> None:
|
|
"""The refresh-path submodule update should use --depth=1."""
|
|
CORE.config_path = tmp_path / "test.yaml"
|
|
|
|
url = "https://github.com/test/repo"
|
|
domain = "test"
|
|
repo_dir = _compute_repo_dir(url, None, domain)
|
|
|
|
_setup_old_repo(repo_dir)
|
|
mock_run_git_command.return_value = "abc123"
|
|
|
|
git.clone_or_update(
|
|
url=url,
|
|
ref=None,
|
|
refresh=TimePeriodSeconds(days=1),
|
|
domain=domain,
|
|
submodules=["components/foo"],
|
|
)
|
|
|
|
submodule_calls = [
|
|
c for c in mock_run_git_command.call_args_list if "submodule" in c[0][0]
|
|
]
|
|
assert len(submodule_calls) == 1
|
|
cmd = submodule_calls[0][0][0]
|
|
assert "--depth=1" in cmd
|
|
assert "components/foo" in cmd
|
|
assert cmd.index("--") < cmd.index("components/foo")
|
|
|
|
|
|
def test_refresh_picks_up_new_remote_commits(
|
|
tmp_path: Path, mock_run_git_command: Mock
|
|
) -> None:
|
|
"""Shallow fetch must still pull new commits when the remote tip moves.
|
|
|
|
Simulates a stale local repo at SHA "old" while the remote has advanced
|
|
to SHA "new". The refresh path must run fetch (with --depth=1) followed
|
|
by reset --hard FETCH_HEAD so the working tree advances to the new tip.
|
|
"""
|
|
CORE.config_path = tmp_path / "test.yaml"
|
|
|
|
url = "https://github.com/test/repo"
|
|
ref = "main"
|
|
domain = "test"
|
|
repo_dir = _compute_repo_dir(url, ref, domain)
|
|
|
|
_setup_old_repo(repo_dir)
|
|
|
|
# rev-parse is called once before fetch to record the pre-update SHA.
|
|
rev_parse_calls = {"count": 0}
|
|
|
|
def git_command_side_effect(
|
|
cmd: list[str], cwd: str | None = None, **kwargs: Any
|
|
) -> str:
|
|
cmd_type = _get_git_command_type(cmd)
|
|
if cmd_type == "rev-parse":
|
|
rev_parse_calls["count"] += 1
|
|
return "old_sha"
|
|
return ""
|
|
|
|
mock_run_git_command.side_effect = git_command_side_effect
|
|
|
|
_, revert = git.clone_or_update(
|
|
url=url, ref=ref, refresh=TimePeriodSeconds(days=1), domain=domain
|
|
)
|
|
|
|
# Verify the refresh sequence: rev-parse -> stash -> fetch (depth=1) -> reset
|
|
call_list = mock_run_git_command.call_args_list
|
|
cmd_sequence = [_get_git_command_type(c[0][0]) for c in call_list]
|
|
assert cmd_sequence == ["rev-parse", "stash", "fetch", "reset"]
|
|
|
|
fetch_cmd = call_list[2][0][0]
|
|
assert "--depth=1" in fetch_cmd
|
|
assert fetch_cmd[-1] == ref
|
|
|
|
reset_cmd = call_list[3][0][0]
|
|
assert reset_cmd[-1] == "FETCH_HEAD"
|
|
|
|
# revert callback should reset back to the recorded pre-update SHA.
|
|
assert revert is not None
|
|
revert()
|
|
assert mock_run_git_command.call_args_list[-1][0][0] == [
|
|
"git",
|
|
"reset",
|
|
"--hard",
|
|
"old_sha",
|
|
]
|