pan/GitMa
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: complete Git Repo Manager MVP implementation

Browse Source 
Backend (Phase 1-6):
- Pydantic schemas for request/response validation
- Service layer (SSH Key, Server, Repo, Sync)
- API routes with authentication
- FastAPI main application with lifespan management
- ORM models (SshKey, Server, Repo, SyncLog)

Frontend (Phase 7):
- Vue 3 + Element Plus + Pinia + Vue Router
- API client with Axios and interceptors
- State management stores
- All page components (Dashboard, Servers, Repos, SyncLogs, SshKeys, Settings)

Deployment (Phase 8):
- README with quick start guide
- Startup script (start.sh)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
panw
2026-03-30 16:30:13 +08:00
parent 960056c88c
commit 44921c5646
46 changed files with 6533 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

411
backend/tests/test_services/test_repo_service.py Normal file
View File

@@ -0,0 +1,411 @@
"""
Tests for Repo Service.
"""
import base64
import pytest
import time
from pathlib import Path
from app.models.repo import Repo
from app.models.server import Server
from app.services.repo_service import RepoService
from app.services.server_service import ServerService
from app.services.ssh_key_service import SshKeyService
# Valid test SSH key
VALID_SSH_KEY = """-----BEGIN OPENSSH PRIVATE KEY-----
b3BlbnNzaC1rZXktdjEAAAAABG5vbmUAAAAEbm9uZQAAAAAAAAABAAAAMwAAAAtzc2gtZW
QyNTUxOQAAACB/pDNwjNcznNaRlLNF5G9hCQNjbqNZ7QeKyLIy/nvHAAAAJi/vqmQv6pk
AAAAAtzc2gtZWQyNTUxOQAAACB/pDNwjNcznNaRlLNF5G9hCQNjbqNZ7QeKyLIy/nvHAA
AAAEBD0cWNQnpLDUYEGNMSgVIApVJfCFuRfGG3uxJZRKLvqH+kM3CM1zOc1pGUssXkb2E
JA2uuo1ntB4rIsjL+e8cAAAADm1lc3NlbmdlckBrZW50cm9zBAgMEBQ=
-----END OPENSSH PRIVATE KEY-----
"""
def create_test_server_with_repo(db_session, server_name="test-server"):
"""Helper to create a test server."""
ssh_service = SshKeyService(db_session)
ssh_key = ssh_service.create_ssh_key(name="test-ssh-key", private_key=VALID_SSH_KEY)
server_service = ServerService(db_session)
return server_service.create_server(
name=server_name,
url="https://gitea.example.com",
api_token="test-api-token",
ssh_key_id=ssh_key.id,
sync_enabled=False,
schedule_cron=None
)
class TestCreateRepo:
"""Tests for create_repo method."""
def test_create_repo_success(self, db_session, test_env_vars):
"""Test successful repository creation."""
server = create_test_server_with_repo(db_session)
service = RepoService(db_session)
repo = service.create_repo(
server_id=server.id,
name="test-repo",
full_name="owner/test-repo",
clone_url="git@gitea.example.com:owner/test-repo.git",
local_path="/tmp/test-repo"
)
assert repo.id is not None
assert repo.server_id == server.id
assert repo.name == "test-repo"
assert repo.full_name == "owner/test-repo"
assert repo.clone_url == "git@gitea.example.com:owner/test-repo.git"
assert repo.local_path == "/tmp/test-repo"
assert repo.status == "pending"
assert repo.last_sync_at is None
assert repo.created_at is not None
def test_create_repo_with_invalid_server_id(self, db_session, test_env_vars):
"""Test that invalid server_id is rejected."""
service = RepoService(db_session)
with pytest.raises(ValueError, match="Server not found"):
service.create_repo(
server_id=99999,
name="test-repo",
full_name="owner/test-repo",
clone_url="git@gitea.example.com:owner/test-repo.git",
local_path="/tmp/test-repo"
)
def test_create_repo_generates_local_path(self, db_session, test_env_vars):
"""Test that local_path can be generated correctly."""
server = create_test_server_with_repo(db_session)
service = RepoService(db_session)
# Test with None local_path to let it generate
repo = service.create_repo(
server_id=server.id,
name="auto-path-repo",
full_name="owner/auto-path-repo",
clone_url="git@gitea.example.com:owner/auto-path-repo.git",
local_path=None
)
# Should generate path based on server and repo name
assert repo.local_path is not None
assert "auto-path-repo" in repo.local_path
def test_create_repo_duplicate_name_same_server(self, db_session, test_env_vars):
"""Test that duplicate repo names on same server are allowed (different from servers)."""
server = create_test_server_with_repo(db_session)
service = RepoService(db_session)
# Create first repo
service.create_repo(
server_id=server.id,
name="duplicate-repo",
full_name="owner1/duplicate-repo",
clone_url="git@gitea.example.com:owner1/duplicate-repo.git",
local_path="/tmp/duplicate-repo-1"
)
# Create second repo with same name but different full_name
repo2 = service.create_repo(
server_id=server.id,
name="duplicate-repo",
full_name="owner2/duplicate-repo",
clone_url="git@gitea.example.com:owner2/duplicate-repo.git",
local_path="/tmp/duplicate-repo-2"
)
assert repo2 is not None
assert repo2.full_name == "owner2/duplicate-repo"
def test_create_repo_default_status(self, db_session, test_env_vars):
"""Test that new repos have default 'pending' status."""
server = create_test_server_with_repo(db_session)
service = RepoService(db_session)
repo = service.create_repo(
server_id=server.id,
name="status-test-repo",
full_name="owner/status-test-repo",
clone_url="git@gitea.example.com:owner/status-test-repo.git",
local_path="/tmp/status-test-repo"
)
assert repo.status == "pending"
class TestListRepos:
"""Tests for list_repos method."""
def test_list_repos_empty(self, db_session, test_env_vars):
"""Test listing repos when none exist."""
server = create_test_server_with_repo(db_session)
service = RepoService(db_session)
repos = service.list_repos(server.id)
assert repos == []
def test_list_repos_multiple(self, db_session, test_env_vars):
"""Test listing multiple repos for a server."""
server = create_test_server_with_repo(db_session)
service = RepoService(db_session)
service.create_repo(
server_id=server.id,
name="repo-1",
full_name="owner/repo-1",
clone_url="git@gitea.example.com:owner/repo-1.git",
local_path="/tmp/repo-1"
)
service.create_repo(
server_id=server.id,
name="repo-2",
full_name="owner/repo-2",
clone_url="git@gitea.example.com:owner/repo-2.git",
local_path="/tmp/repo-2"
)
repos = service.list_repos(server.id)
assert len(repos) == 2
assert any(r.name == "repo-1" for r in repos)
assert any(r.name == "repo-2" for r in repos)
def test_list_repos_filters_by_server(self, db_session, test_env_vars):
"""Test that list_repos only returns repos for specified server."""
server1 = create_test_server_with_repo(db_session, "server-1")
server2 = create_test_server_with_repo(db_session, "server-2")
service = RepoService(db_session)
# Create repo for server1
service.create_repo(
server_id=server1.id,
name="server1-repo",
full_name="owner/server1-repo",
clone_url="git@gitea.example.com:owner/server1-repo.git",
local_path="/tmp/server1-repo"
)
# Create repo for server2
service.create_repo(
server_id=server2.id,
name="server2-repo",
full_name="owner/server2-repo",
clone_url="git@gitea.example.com:owner/server2-repo.git",
local_path="/tmp/server2-repo"
)
# List repos for server1 should only return server1's repo
server1_repos = service.list_repos(server1.id)
assert len(server1_repos) == 1
assert server1_repos[0].name == "server1-repo"
# List repos for server2 should only return server2's repo
server2_repos = service.list_repos(server2.id)
assert len(server2_repos) == 1
assert server2_repos[0].name == "server2-repo"
def test_list_repos_ordered_by_creation(self, db_session, test_env_vars):
"""Test that repos are listed in creation order."""
server = create_test_server_with_repo(db_session)
service = RepoService(db_session)
repo1 = service.create_repo(
server_id=server.id,
name="first-repo",
full_name="owner/first-repo",
clone_url="git@gitea.example.com:owner/first-repo.git",
local_path="/tmp/first-repo"
)
time.sleep(0.1) # Small delay to ensure timestamp difference
repo2 = service.create_repo(
server_id=server.id,
name="second-repo",
full_name="owner/second-repo",
clone_url="git@gitea.example.com:owner/second-repo.git",
local_path="/tmp/second-repo"
)
repos = service.list_repos(server.id)
assert repos[0].id == repo1.id
assert repos[1].id == repo2.id
class TestGetRepo:
"""Tests for get_repo method."""
def test_get_repo_by_id(self, db_session, test_env_vars):
"""Test getting a repo by ID."""
server = create_test_server_with_repo(db_session)
service = RepoService(db_session)
created_repo = service.create_repo(
server_id=server.id,
name="get-test-repo",
full_name="owner/get-test-repo",
clone_url="git@gitea.example.com:owner/get-test-repo.git",
local_path="/tmp/get-test-repo"
)
retrieved_repo = service.get_repo(created_repo.id)
assert retrieved_repo is not None
assert retrieved_repo.id == created_repo.id
assert retrieved_repo.name == "get-test-repo"
def test_get_repo_not_found(self, db_session, test_env_vars):
"""Test getting a non-existent repo."""
service = RepoService(db_session)
repo = service.get_repo(99999)
assert repo is None
class TestUpdateRepoStatus:
"""Tests for update_repo_status method."""
def test_update_repo_status_to_syncing(self, db_session, test_env_vars):
"""Test updating repo status to syncing."""
server = create_test_server_with_repo(db_session)
service = RepoService(db_session)
repo = service.create_repo(
server_id=server.id,
name="status-repo",
full_name="owner/status-repo",
clone_url="git@gitea.example.com:owner/status-repo.git",
local_path="/tmp/status-repo"
)
updated_repo = service.update_repo_status(repo.id, "syncing")
assert updated_repo.status == "syncing"
assert updated_repo.id == repo.id
def test_update_repo_status_to_success(self, db_session, test_env_vars):
"""Test updating repo status to success."""
server = create_test_server_with_repo(db_session)
service = RepoService(db_session)
repo = service.create_repo(
server_id=server.id,
name="status-repo",
full_name="owner/status-repo",
clone_url="git@gitea.example.com:owner/status-repo.git",
local_path="/tmp/status-repo"
)
updated_repo = service.update_repo_status(repo.id, "success")
assert updated_repo.status == "success"
def test_update_repo_status_to_failed(self, db_session, test_env_vars):
"""Test updating repo status to failed."""
server = create_test_server_with_repo(db_session)
service = RepoService(db_session)
repo = service.create_repo(
server_id=server.id,
name="status-repo",
full_name="owner/status-repo",
clone_url="git@gitea.example.com:owner/status-repo.git",
local_path="/tmp/status-repo"
)
updated_repo = service.update_repo_status(repo.id, "failed")
assert updated_repo.status == "failed"
def test_update_repo_status_not_found(self, db_session, test_env_vars):
"""Test updating status for non-existent repo."""
service = RepoService(db_session)
with pytest.raises(ValueError, match="Repo not found"):
service.update_repo_status(99999, "syncing")
class TestDeleteRepo:
"""Tests for delete_repo method."""
def test_delete_repo_success(self, db_session, test_env_vars):
"""Test successful repo deletion."""
server = create_test_server_with_repo(db_session)
service = RepoService(db_session)
repo = service.create_repo(
server_id=server.id,
name="delete-test-repo",
full_name="owner/delete-test-repo",
clone_url="git@gitea.example.com:owner/delete-test-repo.git",
local_path="/tmp/delete-test-repo"
)
result = service.delete_repo(repo.id)
assert result is True
# Verify the repo is deleted
retrieved_repo = service.get_repo(repo.id)
assert retrieved_repo is None
def test_delete_repo_not_found(self, db_session, test_env_vars):
"""Test deleting a non-existent repo."""
service = RepoService(db_session)
result = service.delete_repo(99999)
assert result is False
class TestIntegration:
"""Integration tests for Repo service."""
def test_repo_lifecycle(self, db_session, test_env_vars):
"""Test complete lifecycle of a repo."""
server = create_test_server_with_repo(db_session)
service = RepoService(db_session)
# Create repo
repo = service.create_repo(
server_id=server.id,
name="lifecycle-repo",
full_name="owner/lifecycle-repo",
clone_url="git@gitea.example.com:owner/lifecycle-repo.git",
local_path="/tmp/lifecycle-repo"
)
assert repo.status == "pending"
# Update status to syncing
repo = service.update_repo_status(repo.id, "syncing")
assert repo.status == "syncing"
# Update status to success
repo = service.update_repo_status(repo.id, "success")
assert repo.status == "success"
# Verify we can retrieve it
retrieved = service.get_repo(repo.id)
assert retrieved is not None
assert retrieved.status == "success"
# List repos for server
repos = service.list_repos(server.id)
assert len(repos) == 1
assert repos[0].id == repo.id
# Delete repo
result = service.delete_repo(repo.id)
assert result is True
# Verify it's gone
repos = service.list_repos(server.id)
assert len(repos) == 0