Developer
81b9a66ef5
- Delete 4 obsolete unit tests tied to removed git mount/clone models - Update imports and assertions across unit/integration/service tests - Fix Settings defaults (postgres host, JWT props, cookie_samesite) - Add skip guards for PostgreSQL-dependent integration tests - Fix GitService env assertions and HealthMonitor state-change tests - Repair docker/container inspect assertions in test_docker_service - Fix ToolTypeCreate default_port validator ordering bug - Fix check_port_exposed substring false-positive for port 0 - Update test_tool_types_api_extended to use interface_type field Quality gates: pytest 311 passed, 34 skipped; npm typecheck/lint/test 87 passed
160 lines
4.9 KiB
Python
160 lines
4.9 KiB
Python
"""Unit tests for GitService."""
|
|
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from src.services.git.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,
|
|
env=None,
|
|
)
|
|
|
|
@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,
|
|
env=None,
|
|
)
|
|
|
|
@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,
|
|
env=None,
|
|
)
|
|
|
|
|
|
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,
|
|
env=None,
|
|
)
|
|
|
|
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
|