alex
3728c245d3
- Add GitHubAdapter and GitLabAdapter with URL parsing - Create provider factory in apps/api/app/git/providers/ - Implement clone, fetch, push in LocalGitOperations - Add repository_connections router with CRUD and SSH key endpoints - Create RepositoryConnection schema with validation - Update models and routers __init__.py for new components - Add comprehensive tests for git operations
148 lines
4.3 KiB
Python
148 lines
4.3 KiB
Python
"""Tests for local Git operations."""
|
|
|
|
import subprocess
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from app.git.operations import LocalGitOperations
|
|
|
|
|
|
@pytest.fixture
|
|
def local_git() -> LocalGitOperations:
|
|
return LocalGitOperations()
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_repo() -> Any:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
repo_path = Path(tmpdir) / "repo"
|
|
repo_path.mkdir()
|
|
subprocess.run(
|
|
["git", "init", "--initial-branch=main"],
|
|
cwd=repo_path,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
subprocess.run(
|
|
["git", "config", "user.email", "test@example.com"],
|
|
cwd=repo_path,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
subprocess.run(
|
|
["git", "config", "user.name", "Test User"],
|
|
cwd=repo_path,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
yield repo_path
|
|
|
|
|
|
def test_clone_invalid_repo_raises(local_git: LocalGitOperations) -> None:
|
|
with pytest.raises(RuntimeError, match="Git clone failed"):
|
|
local_git.clone("https://example.com/repo.git", Path("/tmp/dest"), "cred-id")
|
|
|
|
|
|
def test_fetch_no_remote_succeeds(local_git: LocalGitOperations, temp_repo: Path) -> None:
|
|
local_git.fetch(temp_repo, "cred-id")
|
|
|
|
|
|
def test_push_no_remote_raises(local_git: LocalGitOperations, temp_repo: Path) -> None:
|
|
with pytest.raises(RuntimeError, match="Git push failed"):
|
|
local_git.push(temp_repo, "cred-id")
|
|
|
|
|
|
def test_get_status_on_non_git_directory_raises(local_git: LocalGitOperations) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
with pytest.raises(RuntimeError, match="Not a git repository"):
|
|
local_git.get_status(Path(tmpdir))
|
|
|
|
|
|
def test_get_status_clean_repo(local_git: LocalGitOperations, temp_repo: Path) -> None:
|
|
status = local_git.get_status(temp_repo)
|
|
assert status["branch"] == "main"
|
|
assert status["clean"] is True
|
|
assert status["untracked"] == []
|
|
assert status["modified"] == []
|
|
assert status["staged"] == []
|
|
assert status["deleted"] == []
|
|
|
|
|
|
def test_get_status_untracked_file(local_git: LocalGitOperations, temp_repo: Path) -> None:
|
|
(temp_repo / "newfile.txt").write_text("hello")
|
|
status = local_git.get_status(temp_repo)
|
|
assert "newfile.txt" in status["untracked"]
|
|
assert status["clean"] is False
|
|
|
|
|
|
def test_get_status_staged_file(local_git: LocalGitOperations, temp_repo: Path) -> None:
|
|
file_path = temp_repo / "newfile.txt"
|
|
file_path.write_text("hello")
|
|
subprocess.run(
|
|
["git", "add", "newfile.txt"],
|
|
cwd=temp_repo,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
status = local_git.get_status(temp_repo)
|
|
assert "newfile.txt" in status["staged"]
|
|
assert status["clean"] is False
|
|
|
|
|
|
def test_get_status_modified_file(local_git: LocalGitOperations, temp_repo: Path) -> None:
|
|
file_path = temp_repo / "newfile.txt"
|
|
file_path.write_text("hello")
|
|
subprocess.run(
|
|
["git", "add", "newfile.txt"],
|
|
cwd=temp_repo,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
file_path.write_text("world")
|
|
status = local_git.get_status(temp_repo)
|
|
assert "newfile.txt" in status["modified"]
|
|
assert status["clean"] is False
|
|
|
|
|
|
def test_get_status_deleted_file(local_git: LocalGitOperations, temp_repo: Path) -> None:
|
|
file_path = temp_repo / "newfile.txt"
|
|
file_path.write_text("hello")
|
|
subprocess.run(
|
|
["git", "add", "newfile.txt"],
|
|
cwd=temp_repo,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
subprocess.run(
|
|
["git", "commit", "-m", "add file"],
|
|
cwd=temp_repo,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
file_path.unlink()
|
|
status = local_git.get_status(temp_repo)
|
|
assert "newfile.txt" in status["deleted"]
|
|
assert status["clean"] is False
|
|
|
|
|
|
def test_get_status_branch_name(local_git: LocalGitOperations, temp_repo: Path) -> None:
|
|
subprocess.run(
|
|
["git", "checkout", "-b", "feature-branch"],
|
|
cwd=temp_repo,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
status = local_git.get_status(temp_repo)
|
|
assert status["branch"] == "feature-branch"
|