mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-02 20:22:15 +02:00
461 lines
18 KiB
Python
461 lines
18 KiB
Python
"""Unit tests for the git_providers abstraction package."""
|
|
|
|
import hashlib
|
|
import json
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from backend.app.services.git_providers.factory import get_provider_backend
|
|
from backend.app.services.git_providers.forgejo import ForgejoBackend
|
|
from backend.app.services.git_providers.gitea import GiteaBackend
|
|
from backend.app.services.git_providers.github import GitHubBackend
|
|
from backend.app.services.git_providers.gitlab import GitLabBackend
|
|
|
|
|
|
class TestFactory:
|
|
def test_known_providers_return_correct_class(self):
|
|
assert isinstance(get_provider_backend("github"), GitHubBackend)
|
|
assert isinstance(get_provider_backend("gitea"), GiteaBackend)
|
|
assert isinstance(get_provider_backend("forgejo"), ForgejoBackend)
|
|
assert isinstance(get_provider_backend("gitlab"), GitLabBackend)
|
|
|
|
def test_unknown_provider_raises_value_error(self):
|
|
with pytest.raises(ValueError, match="Unknown Git provider"):
|
|
get_provider_backend("bitbucket")
|
|
|
|
|
|
class TestGitHubBackendParseUrl:
|
|
def setup_method(self):
|
|
self.backend = GitHubBackend()
|
|
|
|
def test_https_url(self):
|
|
owner, repo = self.backend.parse_repo_url("https://github.com/owner/repo")
|
|
assert owner == "owner"
|
|
assert repo == "repo"
|
|
|
|
def test_https_url_with_git_suffix(self):
|
|
owner, repo = self.backend.parse_repo_url("https://github.com/owner/repo.git")
|
|
assert owner == "owner"
|
|
assert repo == "repo"
|
|
|
|
def test_ssh_url(self):
|
|
owner, repo = self.backend.parse_repo_url("git@github.com:owner/repo")
|
|
assert owner == "owner"
|
|
assert repo == "repo"
|
|
|
|
def test_ssh_url_with_git_suffix(self):
|
|
owner, repo = self.backend.parse_repo_url("git@github.com:owner/repo.git")
|
|
assert owner == "owner"
|
|
assert repo == "repo"
|
|
|
|
def test_invalid_url_raises_value_error(self):
|
|
with pytest.raises(ValueError, match="Cannot parse repository URL"):
|
|
self.backend.parse_repo_url("https://example.com/not-a-repo")
|
|
|
|
def test_empty_url_raises_value_error(self):
|
|
with pytest.raises(ValueError):
|
|
self.backend.parse_repo_url("")
|
|
|
|
|
|
class TestGitHubBackendApiBase:
|
|
def setup_method(self):
|
|
self.backend = GitHubBackend()
|
|
|
|
def test_github_com_returns_api_github_com(self):
|
|
assert self.backend.get_api_base("https://github.com/owner/repo") == "https://api.github.com"
|
|
|
|
def test_ghe_host_returns_v3_endpoint(self):
|
|
assert self.backend.get_api_base("https://github.example.com/owner/repo") == "https://github.example.com/api/v3"
|
|
|
|
def test_ghe_host_with_port(self):
|
|
assert self.backend.get_api_base("https://github.example.com:8443/owner/repo") == "https://github.example.com:8443/api/v3"
|
|
|
|
def test_ssh_github_com_returns_api_github_com(self):
|
|
assert self.backend.get_api_base("git@github.com:owner/repo.git") == "https://api.github.com"
|
|
|
|
def test_ssh_ghe_host_returns_v3_endpoint(self):
|
|
assert self.backend.get_api_base("git@github.example.com:owner/repo.git") == "https://github.example.com/api/v3"
|
|
|
|
|
|
class TestGiteaBackendApiBase:
|
|
def setup_method(self):
|
|
self.backend = GiteaBackend()
|
|
|
|
def test_derives_api_base_from_repo_url(self):
|
|
result = self.backend.get_api_base("https://git.example.com/owner/repo")
|
|
assert result == "https://git.example.com/api/v1"
|
|
|
|
def test_derives_api_base_with_port(self):
|
|
result = self.backend.get_api_base("https://git.example.com:3000/owner/repo")
|
|
assert result == "https://git.example.com:3000/api/v1"
|
|
|
|
def test_invalid_url_raises_value_error(self):
|
|
with pytest.raises(ValueError, match="Cannot derive API base"):
|
|
self.backend.get_api_base("not-a-url")
|
|
|
|
def test_parse_url_uses_instance_host(self):
|
|
owner, repo = self.backend.parse_repo_url("https://git.example.com/owner/repo")
|
|
assert owner == "owner"
|
|
assert repo == "repo"
|
|
|
|
|
|
class TestGiteaBackendPushFiles:
|
|
def setup_method(self):
|
|
self.backend = GiteaBackend()
|
|
self.repo_url = "https://git.example.com/owner/repo"
|
|
self.token = "gitea-token"
|
|
self.branch = "bambuddy-backup"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_n_files_produce_single_commit(self):
|
|
"""All changed files are bundled into one commit via the Git Data API."""
|
|
files = {"a.json": {"k": "v1"}, "b.json": {"k": "v2"}}
|
|
client = AsyncMock()
|
|
client.get = AsyncMock(
|
|
side_effect=[
|
|
_make_mock_response(200, {"object": {"sha": "base-commit"}}),
|
|
_make_mock_response(200, {"tree": {"sha": "base-tree"}}),
|
|
_make_mock_response(200, {"tree": []}),
|
|
]
|
|
)
|
|
client.post = AsyncMock(
|
|
side_effect=[
|
|
_make_mock_response(201, {"sha": "blob1"}),
|
|
_make_mock_response(201, {"sha": "blob2"}),
|
|
_make_mock_response(201, {"sha": "new-tree"}),
|
|
_make_mock_response(201, {"sha": "new-commit"}),
|
|
]
|
|
)
|
|
client.patch = AsyncMock(return_value=_make_mock_response(200, {}))
|
|
|
|
result = await self.backend.push_files(self.repo_url, self.token, self.branch, files, client)
|
|
|
|
assert result["status"] == "success"
|
|
assert result["files_changed"] == 2
|
|
commit_calls = [c for c in client.post.call_args_list if "/git/commits" in c.args[0]]
|
|
assert len(commit_calls) == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_uses_gitea_api_v1_base_not_github(self):
|
|
"""Git Data API calls target the instance's /api/v1, not api.github.com."""
|
|
client = AsyncMock()
|
|
client.get = AsyncMock(
|
|
side_effect=[
|
|
_make_mock_response(200, {"object": {"sha": "base-commit"}}),
|
|
_make_mock_response(200, {"tree": {"sha": "base-tree"}}),
|
|
_make_mock_response(200, {"tree": []}),
|
|
]
|
|
)
|
|
client.post = AsyncMock(
|
|
side_effect=[
|
|
_make_mock_response(201, {"sha": "blob1"}),
|
|
_make_mock_response(201, {"sha": "new-tree"}),
|
|
_make_mock_response(201, {"sha": "new-commit"}),
|
|
]
|
|
)
|
|
client.patch = AsyncMock(return_value=_make_mock_response(200, {}))
|
|
|
|
await self.backend.push_files(self.repo_url, self.token, self.branch, {"a.json": {"k": "v"}}, client)
|
|
|
|
first_get_url = client.get.call_args_list[0].args[0]
|
|
assert "git.example.com/api/v1" in first_get_url
|
|
assert "api.github.com" not in first_get_url
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skips_unchanged_files(self):
|
|
"""Files whose blob SHA matches the existing tree entry are excluded from the commit."""
|
|
content = {"name": "my-printer"}
|
|
sha = _blob_sha(content)
|
|
|
|
client = AsyncMock()
|
|
client.get = AsyncMock(
|
|
side_effect=[
|
|
_make_mock_response(200, {"object": {"sha": "base-commit"}}),
|
|
_make_mock_response(200, {"tree": {"sha": "base-tree"}}),
|
|
_make_mock_response(200, {"tree": [{"type": "blob", "path": "config/printers.json", "sha": sha}]}),
|
|
]
|
|
)
|
|
|
|
result = await self.backend.push_files(
|
|
self.repo_url, self.token, self.branch, {"config/printers.json": content}, client
|
|
)
|
|
|
|
assert result["status"] == "skipped"
|
|
client.post.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_creates_missing_branch_via_git_refs_api(self):
|
|
"""A missing backup branch is created via the Git Data API refs endpoint."""
|
|
client = AsyncMock()
|
|
client.get = AsyncMock(
|
|
side_effect=[
|
|
# branch ref missing
|
|
_make_mock_response(404, {}),
|
|
# repo info for default branch
|
|
_make_mock_response(200, {"default_branch": "main"}),
|
|
# default branch ref
|
|
_make_mock_response(200, {"object": {"sha": "base-sha"}}),
|
|
# second push_files call: branch now exists
|
|
_make_mock_response(200, {"object": {"sha": "base-sha"}}),
|
|
_make_mock_response(200, {"tree": {"sha": "base-tree"}}),
|
|
_make_mock_response(200, {"tree": []}),
|
|
]
|
|
)
|
|
client.post = AsyncMock(
|
|
side_effect=[
|
|
_make_mock_response(201, {}), # create ref
|
|
_make_mock_response(201, {"sha": "blob1"}),
|
|
_make_mock_response(201, {"sha": "new-tree"}),
|
|
_make_mock_response(201, {"sha": "new-commit"}),
|
|
]
|
|
)
|
|
client.patch = AsyncMock(return_value=_make_mock_response(200, {}))
|
|
|
|
result = await self.backend.push_files(
|
|
self.repo_url, self.token, self.branch, {"a.json": {"k": "v"}}, client
|
|
)
|
|
|
|
assert result["status"] == "success"
|
|
ref_create_call = client.post.call_args_list[0]
|
|
assert "/git/refs" in ref_create_call.args[0]
|
|
assert ref_create_call.kwargs["json"]["ref"] == f"refs/heads/{self.branch}"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_truncates_upstream_error_body_in_failure_message(self):
|
|
client = AsyncMock()
|
|
client.get = AsyncMock(
|
|
side_effect=[
|
|
_make_mock_response(200, {"object": {"sha": "base-commit"}}),
|
|
_make_mock_response(200, {"tree": {"sha": "base-tree"}}),
|
|
_make_mock_response(200, {"tree": []}),
|
|
]
|
|
)
|
|
client.post = AsyncMock(
|
|
side_effect=[
|
|
_make_mock_response(201, {"sha": "blob1"}),
|
|
_make_mock_response(500, {}, text="x" * 500),
|
|
]
|
|
)
|
|
|
|
result = await self.backend.push_files(
|
|
self.repo_url, self.token, self.branch, {"a.json": {"k": "v"}}, client
|
|
)
|
|
|
|
assert result["status"] == "failed"
|
|
assert result["message"] == f"Failed to create tree: {'x' * 197}..."
|
|
|
|
|
|
class TestForgejoBackendApiBase:
|
|
def setup_method(self):
|
|
self.backend = ForgejoBackend()
|
|
|
|
def test_derives_api_base_from_repo_url(self):
|
|
result = self.backend.get_api_base("https://forgejo.example.com/owner/repo")
|
|
assert result == "https://forgejo.example.com/api/v1"
|
|
|
|
def test_derives_api_base_with_port(self):
|
|
result = self.backend.get_api_base("https://forgejo.example.com:3000/owner/repo")
|
|
assert result == "https://forgejo.example.com:3000/api/v1"
|
|
|
|
def test_invalid_url_raises_value_error(self):
|
|
with pytest.raises(ValueError, match="Cannot derive API base"):
|
|
self.backend.get_api_base("not-a-url")
|
|
|
|
def test_parse_url_uses_instance_host(self):
|
|
owner, repo = self.backend.parse_repo_url("https://forgejo.example.com/owner/repo")
|
|
assert owner == "owner"
|
|
assert repo == "repo"
|
|
|
|
|
|
class TestGitLabBackend:
|
|
def setup_method(self):
|
|
self.backend = GitLabBackend()
|
|
|
|
def test_parse_url_https(self):
|
|
owner, repo = self.backend.parse_repo_url("https://gitlab.com/owner/repo")
|
|
assert owner == "owner"
|
|
assert repo == "repo"
|
|
|
|
def test_parse_url_ssh(self):
|
|
owner, repo = self.backend.parse_repo_url("git@gitlab.com:owner/repo.git")
|
|
assert owner == "owner"
|
|
assert repo == "repo"
|
|
|
|
def test_parse_url_invalid_raises(self):
|
|
with pytest.raises(ValueError):
|
|
self.backend.parse_repo_url("not-a-url")
|
|
|
|
def test_get_api_base_derives_from_repo_url(self):
|
|
result = self.backend.get_api_base("https://gitlab.com/owner/repo")
|
|
assert result == "https://gitlab.com/api/v4"
|
|
|
|
def test_get_api_base_derives_from_self_hosted_url(self):
|
|
result = self.backend.get_api_base("https://my-gitlab.example.com/owner/repo")
|
|
assert result == "https://my-gitlab.example.com/api/v4"
|
|
|
|
def test_get_api_base_invalid_url_raises(self):
|
|
with pytest.raises(ValueError, match="Cannot derive API base"):
|
|
self.backend.get_api_base("not-a-url")
|
|
|
|
def test_get_headers_uses_bearer_token(self):
|
|
headers = self.backend.get_headers("mytoken")
|
|
assert headers["Authorization"] == "Bearer mytoken"
|
|
assert "Content-Type" in headers
|
|
|
|
def test_parse_url_subgroup_https(self):
|
|
namespace, repo = self.backend.parse_repo_url("https://gitlab.com/group/subgroup/project")
|
|
assert namespace == "group/subgroup"
|
|
assert repo == "project"
|
|
|
|
def test_parse_url_deep_namespace_https(self):
|
|
namespace, repo = self.backend.parse_repo_url("https://gitlab.com/myorg/team/api/backend")
|
|
assert namespace == "myorg/team/api"
|
|
assert repo == "backend"
|
|
|
|
def test_parse_url_subgroup_ssh(self):
|
|
namespace, repo = self.backend.parse_repo_url("git@gitlab.com:group/subgroup/project.git")
|
|
assert namespace == "group/subgroup"
|
|
assert repo == "project"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_push_files_encodes_subgroup_namespace_in_api_url(self):
|
|
backend = GitLabBackend()
|
|
repo_url = "https://gitlab.com/group/subgroup/project"
|
|
client = AsyncMock()
|
|
client.get = AsyncMock(
|
|
side_effect=[
|
|
_make_mock_response(200, {"name": "bambuddy-backup"}),
|
|
_make_mock_response(200, []),
|
|
]
|
|
)
|
|
client.post = AsyncMock(return_value=_make_mock_response(201, {"id": "abc123"}))
|
|
|
|
await backend.push_files(repo_url, "token", "bambuddy-backup", {"f.json": {}}, client)
|
|
|
|
called_url = client.get.call_args_list[0].args[0]
|
|
assert "group%2Fsubgroup%2Fproject" in called_url
|
|
|
|
|
|
def _blob_sha(content: dict) -> str:
|
|
content_bytes = json.dumps(content, indent=2, default=str).encode("utf-8")
|
|
return hashlib.sha1(f"blob {len(content_bytes)}\0".encode() + content_bytes, usedforsecurity=False).hexdigest()
|
|
|
|
|
|
def _make_mock_response(status_code: int, body=None, text: str = ""):
|
|
resp = MagicMock()
|
|
resp.status_code = status_code
|
|
resp.text = text
|
|
resp.json = MagicMock(return_value=body or {})
|
|
return resp
|
|
|
|
|
|
class TestGitLabBackendPushFiles:
|
|
def setup_method(self):
|
|
self.backend = GitLabBackend()
|
|
self.repo_url = "https://gitlab.com/owner/repo"
|
|
self.token = "glpat-test"
|
|
self.branch = "bambuddy-backup"
|
|
self.files = {"config/printers.json": {"name": "my-printer"}}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skips_commit_when_content_unchanged(self):
|
|
sha = _blob_sha(self.files["config/printers.json"])
|
|
|
|
client = AsyncMock()
|
|
client.get = AsyncMock(
|
|
side_effect=[
|
|
# branch check → branch exists
|
|
_make_mock_response(200, {"name": self.branch}),
|
|
# tree page 1 → one blob whose sha matches current content
|
|
_make_mock_response(200, [{"type": "blob", "path": "config/printers.json", "id": sha}]),
|
|
# tree page 2 → empty, stop pagination
|
|
_make_mock_response(200, []),
|
|
]
|
|
)
|
|
|
|
result = await self.backend.push_files(self.repo_url, self.token, self.branch, self.files, client)
|
|
|
|
assert result["status"] == "skipped"
|
|
assert result["files_changed"] == 0
|
|
client.post.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_commits_when_content_changed(self):
|
|
stale_sha = "0000000000000000000000000000000000000000"
|
|
|
|
client = AsyncMock()
|
|
client.get = AsyncMock(
|
|
side_effect=[
|
|
_make_mock_response(200, {"name": self.branch}),
|
|
_make_mock_response(200, [{"type": "blob", "path": "config/printers.json", "id": stale_sha}]),
|
|
_make_mock_response(200, []), # page 2 empty, stop pagination
|
|
]
|
|
)
|
|
client.post = AsyncMock(return_value=_make_mock_response(201, {"id": "abc123"}))
|
|
|
|
result = await self.backend.push_files(self.repo_url, self.token, self.branch, self.files, client)
|
|
|
|
assert result["status"] == "success"
|
|
assert result["files_changed"] == 1
|
|
client.post.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_truncates_upstream_error_body_in_failure_message(self):
|
|
client = AsyncMock()
|
|
client.get = AsyncMock(
|
|
side_effect=[
|
|
_make_mock_response(200, {"name": self.branch}),
|
|
_make_mock_response(200, []),
|
|
]
|
|
)
|
|
client.post = AsyncMock(return_value=_make_mock_response(500, {}, text="x" * 500))
|
|
|
|
result = await self.backend.push_files(self.repo_url, self.token, self.branch, self.files, client)
|
|
|
|
assert result["status"] == "failed"
|
|
assert result["message"] == f"Failed to create commit: {'x' * 197}..."
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_creates_new_file_not_in_existing_tree(self):
|
|
client = AsyncMock()
|
|
client.get = AsyncMock(
|
|
side_effect=[
|
|
_make_mock_response(200, {"name": self.branch}),
|
|
# tree is empty
|
|
_make_mock_response(200, []),
|
|
]
|
|
)
|
|
client.post = AsyncMock(return_value=_make_mock_response(201, {"id": "def456"}))
|
|
|
|
result = await self.backend.push_files(self.repo_url, self.token, self.branch, self.files, client)
|
|
|
|
assert result["status"] == "success"
|
|
call_kwargs = client.post.call_args.kwargs["json"]
|
|
assert call_kwargs["actions"][0]["action"] == "create"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_paginates_tree_to_find_unchanged_file_on_page_2(self):
|
|
"""Files beyond the first 100 are fetched; a file on page 2 is correctly skipped if unchanged."""
|
|
sha = _blob_sha(self.files["config/printers.json"])
|
|
page1_items = [{"type": "blob", "path": f"other{i}.json", "id": "aaa"} for i in range(100)]
|
|
page2_items = (
|
|
[{"type": "blob", "path": f"more{i}.json", "id": "bbb"} for i in range(19)]
|
|
+ [{"type": "blob", "path": "config/printers.json", "id": sha}]
|
|
) # 120 total blobs across two pages
|
|
|
|
client = AsyncMock()
|
|
client.get = AsyncMock(
|
|
side_effect=[
|
|
_make_mock_response(200, {"name": self.branch}), # branch check
|
|
_make_mock_response(200, page1_items), # tree page 1
|
|
_make_mock_response(200, page2_items), # tree page 2
|
|
_make_mock_response(200, []), # tree page 3 empty, stop
|
|
]
|
|
)
|
|
|
|
result = await self.backend.push_files(self.repo_url, self.token, self.branch, self.files, client)
|
|
|
|
assert result["status"] == "skipped"
|
|
client.post.assert_not_called()
|