alex
a5d64d1859
Fixes NameError: ToolInstance not defined at runtime because type annotations are evaluated at class definition time. Deferring annotation evaluation with __future__ annotations keeps TYPE_CHECKING imports from causing runtime crashes. Also includes ruff formatting cleanup on workspace-related files.
156 lines
4.8 KiB
Python
156 lines
4.8 KiB
Python
"""Unit tests for GitService."""
|
|
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from src.services.git_service import GitService
|
|
|
|
|
|
class TestGitServiceClone:
|
|
"""Tests for GitService.clone."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_clone_success(self):
|
|
"""Clone succeeds when git returns 0."""
|
|
mock_proc = AsyncMock()
|
|
mock_proc.returncode = 0
|
|
mock_proc.communicate.return_value = (b"", b"")
|
|
|
|
with patch(
|
|
"asyncio.create_subprocess_exec", return_value=mock_proc
|
|
) as mock_exec:
|
|
await GitService.clone(
|
|
"https://github.com/test/repo.git", "main", "/tmp/ws"
|
|
)
|
|
|
|
mock_exec.assert_called_once_with(
|
|
"git",
|
|
"clone",
|
|
"--branch",
|
|
"main",
|
|
"--single-branch",
|
|
"https://github.com/test/repo.git",
|
|
"/tmp/ws",
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_clone_failure(self):
|
|
"""Clone raises RuntimeError when git fails."""
|
|
mock_proc = AsyncMock()
|
|
mock_proc.returncode = 1
|
|
mock_proc.communicate.return_value = (b"", b"fatal: repository not found")
|
|
|
|
with patch("asyncio.create_subprocess_exec", return_value=mock_proc):
|
|
with pytest.raises(RuntimeError, match="Git clone failed"):
|
|
await GitService.clone("https://bad/url.git", "main", "/tmp/ws")
|
|
|
|
|
|
class TestGitServiceFetch:
|
|
"""Tests for GitService.fetch."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_success(self):
|
|
"""Fetch succeeds when git returns 0."""
|
|
mock_proc = AsyncMock()
|
|
mock_proc.returncode = 0
|
|
mock_proc.communicate.return_value = (b"", b"")
|
|
|
|
with patch(
|
|
"asyncio.create_subprocess_exec", return_value=mock_proc
|
|
) as mock_exec:
|
|
await GitService.fetch("/tmp/repo")
|
|
|
|
mock_exec.assert_called_once_with(
|
|
"git",
|
|
"-C",
|
|
"/tmp/repo",
|
|
"fetch",
|
|
"origin",
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_failure(self):
|
|
"""Fetch raises RuntimeError when git fails."""
|
|
mock_proc = AsyncMock()
|
|
mock_proc.returncode = 128
|
|
mock_proc.communicate.return_value = (b"", b"fatal: not a git repository")
|
|
|
|
with patch("asyncio.create_subprocess_exec", return_value=mock_proc):
|
|
with pytest.raises(RuntimeError, match="Git fetch failed"):
|
|
await GitService.fetch("/not/a/repo")
|
|
|
|
|
|
class TestGitServicePull:
|
|
"""Tests for GitService.pull."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pull_success(self):
|
|
"""Pull succeeds when git returns 0."""
|
|
mock_proc = AsyncMock()
|
|
mock_proc.returncode = 0
|
|
mock_proc.communicate.return_value = (b"Already up to date.", b"")
|
|
|
|
with patch(
|
|
"asyncio.create_subprocess_exec", return_value=mock_proc
|
|
) as mock_exec:
|
|
await GitService.pull("/tmp/repo", "feature-branch")
|
|
|
|
mock_exec.assert_called_once_with(
|
|
"git",
|
|
"-C",
|
|
"/tmp/repo",
|
|
"pull",
|
|
"origin",
|
|
"feature-branch",
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
)
|
|
|
|
|
|
class TestGitServiceBranchExistsRemotely:
|
|
"""Tests for GitService.branch_exists_remotely."""
|
|
|
|
def test_branch_exists(self):
|
|
"""Returns True when branch exists on remote."""
|
|
mock_result = MagicMock()
|
|
mock_result.returncode = 0
|
|
mock_result.stdout = "abc123 refs/heads/main\n"
|
|
|
|
with patch("subprocess.run", return_value=mock_result) as mock_run:
|
|
result = GitService.branch_exists_remotely("/tmp/repo", "main")
|
|
|
|
assert result is True
|
|
mock_run.assert_called_once_with(
|
|
["git", "-C", "/tmp/repo", "ls-remote", "--heads", "origin", "main"],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
def test_branch_not_exists(self):
|
|
"""Returns False when branch does not exist on remote."""
|
|
mock_result = MagicMock()
|
|
mock_result.returncode = 0
|
|
mock_result.stdout = ""
|
|
|
|
with patch("subprocess.run", return_value=mock_result):
|
|
result = GitService.branch_exists_remotely("/tmp/repo", "deleted-branch")
|
|
|
|
assert result is False
|
|
|
|
def test_ls_remote_fails(self):
|
|
"""Returns False when ls-remote fails."""
|
|
mock_result = MagicMock()
|
|
mock_result.returncode = 128
|
|
mock_result.stdout = ""
|
|
|
|
with patch("subprocess.run", return_value=mock_result):
|
|
result = GitService.branch_exists_remotely("/tmp/repo", "main")
|
|
|
|
assert result is False
|