feat: workspace backend foundation (PR-1)
- Add workspaces table migration (2026_06_01_add_workspaces) - Create Workspace model with repo_id, user_id, branch, path, status - Add workspace_id nullable FK to ToolInstance - Create GitService for clone/fetch/pull/branch_exists_remotely - Create WorkspaceManager for create/delete/sync lifecycle - Create workspace CRUD API with 409 handling for duplicates and instances - Wire workspace routes into FastAPI app - 17 tests passing (8 unit + 9 integration), 1 skipped Quality gates: ruff clean
This commit is contained in:
@@ -0,0 +1,153 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user