|
@@ -1,5 +1,6 @@
|
|
|
"""Unit tests for the git_providers abstraction package."""
|
|
"""Unit tests for the git_providers abstraction package."""
|
|
|
|
|
|
|
|
|
|
+import base64
|
|
|
import hashlib
|
|
import hashlib
|
|
|
import json
|
|
import json
|
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
@@ -69,7 +70,10 @@ class TestGitHubBackendApiBase:
|
|
|
assert self.backend.get_api_base("https://github.example.com/owner/repo") == "https://github.example.com/api/v3"
|
|
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):
|
|
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"
|
|
|
|
|
|
|
+ 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):
|
|
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"
|
|
assert self.backend.get_api_base("git@github.com:owner/repo.git") == "https://api.github.com"
|
|
@@ -204,7 +208,7 @@ class TestGiteaBackendPushFiles:
|
|
|
)
|
|
)
|
|
|
client.post = AsyncMock(
|
|
client.post = AsyncMock(
|
|
|
side_effect=[
|
|
side_effect=[
|
|
|
- _make_mock_response(201, {}), # create ref
|
|
|
|
|
|
|
+ _make_mock_response(201, {}), # create ref
|
|
|
_make_mock_response(201, {"sha": "blob1"}),
|
|
_make_mock_response(201, {"sha": "blob1"}),
|
|
|
_make_mock_response(201, {"sha": "new-tree"}),
|
|
_make_mock_response(201, {"sha": "new-tree"}),
|
|
|
_make_mock_response(201, {"sha": "new-commit"}),
|
|
_make_mock_response(201, {"sha": "new-commit"}),
|
|
@@ -212,9 +216,7 @@ class TestGiteaBackendPushFiles:
|
|
|
)
|
|
)
|
|
|
client.patch = AsyncMock(return_value=_make_mock_response(200, {}))
|
|
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
|
|
|
|
|
- )
|
|
|
|
|
|
|
+ result = await self.backend.push_files(self.repo_url, self.token, self.branch, {"a.json": {"k": "v"}}, client)
|
|
|
|
|
|
|
|
assert result["status"] == "success"
|
|
assert result["status"] == "success"
|
|
|
ref_create_call = client.post.call_args_list[0]
|
|
ref_create_call = client.post.call_args_list[0]
|
|
@@ -238,14 +240,262 @@ class TestGiteaBackendPushFiles:
|
|
|
]
|
|
]
|
|
|
)
|
|
)
|
|
|
|
|
|
|
|
- result = await self.backend.push_files(
|
|
|
|
|
- self.repo_url, self.token, self.branch, {"a.json": {"k": "v"}}, client
|
|
|
|
|
- )
|
|
|
|
|
|
|
+ result = await self.backend.push_files(self.repo_url, self.token, self.branch, {"a.json": {"k": "v"}}, client)
|
|
|
|
|
|
|
|
assert result["status"] == "failed"
|
|
assert result["status"] == "failed"
|
|
|
assert result["message"] == f"Failed to create tree: {'x' * 197}..."
|
|
assert result["message"] == f"Failed to create tree: {'x' * 197}..."
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
+class TestGiteaBackendListShapeRefResponse:
|
|
|
|
|
+ """#1224, #1225 regression: Gitea/Forgejo return refs as a *list*, not a dict.
|
|
|
|
|
+
|
|
|
|
|
+ GitHub: ``GET /git/refs/heads/{branch}`` -> ``{"ref": ..., "object": {...}}``.
|
|
|
|
|
+ Gitea/Forgejo: same endpoint -> ``[{"ref": ..., "object": {...}}]``.
|
|
|
|
|
+
|
|
|
|
|
+ The pre-fix code did ``response.json()["object"]["sha"]`` on the Gitea path
|
|
|
|
|
+ and crashed with ``list indices must be integers or slices, not str``.
|
|
|
|
|
+ """
|
|
|
|
|
+
|
|
|
|
|
+ def setup_method(self):
|
|
|
|
|
+ self.backend = GiteaBackend()
|
|
|
|
|
+ self.repo_url = "https://git.example.com/owner/repo"
|
|
|
|
|
+ self.token = "gitea-token"
|
|
|
|
|
+ self.branch = "bambuddy-backup"
|
|
|
|
|
+
|
|
|
|
|
+ def test_ref_sha_extracts_from_list(self):
|
|
|
|
|
+ assert self.backend._ref_sha([{"object": {"sha": "abc"}}]) == "abc"
|
|
|
|
|
+
|
|
|
|
|
+ def test_ref_sha_still_accepts_dict_shape(self):
|
|
|
|
|
+ # Defensive — if Gitea ever returns a dict (older versions, future change),
|
|
|
|
|
+ # we don't want to break.
|
|
|
|
|
+ assert self.backend._ref_sha({"object": {"sha": "abc"}}) == "abc"
|
|
|
|
|
+
|
|
|
|
|
+ def test_ref_sha_raises_on_empty_list(self):
|
|
|
|
|
+ with pytest.raises(ValueError):
|
|
|
|
|
+ self.backend._ref_sha([])
|
|
|
|
|
+
|
|
|
|
|
+ @pytest.mark.asyncio
|
|
|
|
|
+ async def test_push_files_handles_list_shape_branch_ref(self):
|
|
|
|
|
+ """The configured backup branch already exists — ref endpoint returns a list."""
|
|
|
|
|
+ client = AsyncMock()
|
|
|
|
|
+ client.get = AsyncMock(
|
|
|
|
|
+ side_effect=[
|
|
|
|
|
+ _make_mock_response(200, [{"object": {"sha": "base-commit"}}]), # list shape
|
|
|
|
|
+ _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, {}))
|
|
|
|
|
+
|
|
|
|
|
+ result = await self.backend.push_files(self.repo_url, self.token, self.branch, {"a.json": {"k": "v"}}, client)
|
|
|
|
|
+
|
|
|
|
|
+ assert result["status"] == "success"
|
|
|
|
|
+ assert result["commit_sha"] == "new-commit"
|
|
|
|
|
+
|
|
|
|
|
+ @pytest.mark.asyncio
|
|
|
|
|
+ async def test_create_branch_handles_list_shape_default_branch_ref(self):
|
|
|
|
|
+ """Backup branch missing — must read default branch's ref, also list-shaped."""
|
|
|
|
|
+ client = AsyncMock()
|
|
|
|
|
+ client.get = AsyncMock(
|
|
|
|
|
+ side_effect=[
|
|
|
|
|
+ _make_mock_response(404, {}), # missing backup branch
|
|
|
|
|
+ _make_mock_response(200, {"default_branch": "main"}), # repo info
|
|
|
|
|
+ _make_mock_response(200, [{"object": {"sha": "main-sha"}}]), # default branch ref (list)
|
|
|
|
|
+ # second push_files() call — branch now exists
|
|
|
|
|
+ _make_mock_response(200, [{"object": {"sha": "main-sha"}}]),
|
|
|
|
|
+ _make_mock_response(200, {"tree": {"sha": "main-tree"}}),
|
|
|
|
|
+ _make_mock_response(200, {"tree": []}),
|
|
|
|
|
+ ]
|
|
|
|
|
+ )
|
|
|
|
|
+ client.post = AsyncMock(
|
|
|
|
|
+ side_effect=[
|
|
|
|
|
+ _make_mock_response(201, {}), # create branch 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"
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+class TestGiteaBackendEmptyRepoInitialCommit:
|
|
|
|
|
+ """#1224 regression: Git Data API refuses writes against empty Gitea repos.
|
|
|
|
|
+
|
|
|
|
|
+ GitHub accepts ``POST /git/blobs`` against an empty repo and creates the
|
|
|
|
|
+ initial commit + branch. Gitea returns 404 on every blob/tree/commit POST
|
|
|
|
|
+ until the repo has at least one commit. The fix is to use the Contents
|
|
|
|
|
+ API (``POST /repos/.../contents``) which seeds the branch + initial
|
|
|
|
|
+ commit in a single transaction.
|
|
|
|
|
+ """
|
|
|
|
|
+
|
|
|
|
|
+ def setup_method(self):
|
|
|
|
|
+ self.backend = GiteaBackend()
|
|
|
|
|
+ self.repo_url = "https://git.example.com/owner/repo"
|
|
|
|
|
+ self.token = "gitea-token"
|
|
|
|
|
+ self.branch = "main"
|
|
|
|
|
+
|
|
|
|
|
+ @pytest.mark.asyncio
|
|
|
|
|
+ async def test_empty_repo_uses_contents_api_not_git_data_api(self):
|
|
|
|
|
+ files = {"config/printers.json": {"name": "p1"}, "config/spools.json": {"id": 1}}
|
|
|
|
|
+ client = AsyncMock()
|
|
|
|
|
+ client.get = AsyncMock(
|
|
|
|
|
+ side_effect=[
|
|
|
|
|
+ _make_mock_response(404, {}), # backup branch missing
|
|
|
|
|
+ _make_mock_response(200, {"default_branch": "main"}), # repo info
|
|
|
|
|
+ _make_mock_response(404, {}), # default branch missing too -> empty repo
|
|
|
|
|
+ ]
|
|
|
|
|
+ )
|
|
|
|
|
+ client.post = AsyncMock(return_value=_make_mock_response(201, {"commit": {"sha": "initial-sha"}}))
|
|
|
|
|
+
|
|
|
|
|
+ result = await self.backend.push_files(self.repo_url, self.token, self.branch, files, client)
|
|
|
|
|
+
|
|
|
|
|
+ assert result["status"] == "success"
|
|
|
|
|
+ assert result["files_changed"] == 2
|
|
|
|
|
+ assert result["commit_sha"] == "initial-sha"
|
|
|
|
|
+
|
|
|
|
|
+ contents_calls = [c for c in client.post.call_args_list if "/contents" in c.args[0]]
|
|
|
|
|
+ blob_calls = [c for c in client.post.call_args_list if "/git/blobs" in c.args[0]]
|
|
|
|
|
+ tree_calls = [c for c in client.post.call_args_list if "/git/trees" in c.args[0]]
|
|
|
|
|
+ commit_calls = [c for c in client.post.call_args_list if "/git/commits" in c.args[0]]
|
|
|
|
|
+ ref_calls = [c for c in client.post.call_args_list if "/git/refs" in c.args[0]]
|
|
|
|
|
+ # Exactly one Contents API call, no Git Data API writes
|
|
|
|
|
+ assert len(contents_calls) == 1
|
|
|
|
|
+ assert len(blob_calls) == 0
|
|
|
|
|
+ assert len(tree_calls) == 0
|
|
|
|
|
+ assert len(commit_calls) == 0
|
|
|
|
|
+ assert len(ref_calls) == 0
|
|
|
|
|
+
|
|
|
|
|
+ @pytest.mark.asyncio
|
|
|
|
|
+ async def test_contents_api_payload_shape(self):
|
|
|
|
|
+ """The Contents API call must carry branch+new_branch+files in the documented shape."""
|
|
|
|
|
+ files = {"a.json": {"k": "v"}, "nested/b.json": {"x": 1}}
|
|
|
|
|
+ client = AsyncMock()
|
|
|
|
|
+ client.get = AsyncMock(
|
|
|
|
|
+ side_effect=[
|
|
|
|
|
+ _make_mock_response(404, {}),
|
|
|
|
|
+ _make_mock_response(200, {"default_branch": "main"}),
|
|
|
|
|
+ _make_mock_response(404, {}),
|
|
|
|
|
+ ]
|
|
|
|
|
+ )
|
|
|
|
|
+ client.post = AsyncMock(return_value=_make_mock_response(201, {"commit": {"sha": "abc"}}))
|
|
|
|
|
+
|
|
|
|
|
+ await self.backend.push_files(self.repo_url, self.token, self.branch, files, client)
|
|
|
|
|
+
|
|
|
|
|
+ body = client.post.call_args.kwargs["json"]
|
|
|
|
|
+ assert body["branch"] == "main"
|
|
|
|
|
+ assert body["new_branch"] == "main"
|
|
|
|
|
+ assert body["message"].startswith("Initial Bambuddy backup")
|
|
|
|
|
+ assert len(body["files"]) == 2
|
|
|
|
|
+ paths = {f["path"] for f in body["files"]}
|
|
|
|
|
+ assert paths == {"a.json", "nested/b.json"}
|
|
|
|
|
+ for f in body["files"]:
|
|
|
|
|
+ assert f["operation"] == "create"
|
|
|
|
|
+ # Content is base64-encoded JSON of the original dict
|
|
|
|
|
+ decoded = base64.b64decode(f["content"]).decode("utf-8")
|
|
|
|
|
+ assert json.loads(decoded) == files[f["path"]]
|
|
|
|
|
+
|
|
|
|
|
+ @pytest.mark.asyncio
|
|
|
|
|
+ async def test_contents_api_failure_truncates_error_body(self):
|
|
|
|
|
+ client = AsyncMock()
|
|
|
|
|
+ client.get = AsyncMock(
|
|
|
|
|
+ side_effect=[
|
|
|
|
|
+ _make_mock_response(404, {}),
|
|
|
|
|
+ _make_mock_response(200, {"default_branch": "main"}),
|
|
|
|
|
+ _make_mock_response(404, {}),
|
|
|
|
|
+ ]
|
|
|
|
|
+ )
|
|
|
|
|
+ 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, {"a.json": {"k": "v"}}, client)
|
|
|
|
|
+
|
|
|
|
|
+ assert result["status"] == "failed"
|
|
|
|
|
+ assert result["message"] == f"Failed to create initial commit: {'x' * 197}..."
|
|
|
|
|
+
|
|
|
|
|
+ @pytest.mark.asyncio
|
|
|
|
|
+ async def test_empty_files_skips_contents_api_call(self):
|
|
|
|
|
+ # Edge: nothing to commit -> don't make a useless Contents API call.
|
|
|
|
|
+ client = AsyncMock()
|
|
|
|
|
+ client.post = AsyncMock()
|
|
|
|
|
+
|
|
|
|
|
+ result = await self.backend._create_initial_commit(
|
|
|
|
|
+ client, {}, "https://git.example.com/api/v1", "owner", "repo", "main", {}
|
|
|
|
|
+ )
|
|
|
|
|
+
|
|
|
|
|
+ assert result["status"] == "skipped"
|
|
|
|
|
+ assert result["files_changed"] == 0
|
|
|
|
|
+ client.post.assert_not_called()
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+class TestForgejoInheritsGiteaFixes:
|
|
|
|
|
+ """ForgejoBackend extends GiteaBackend with no overrides — must inherit both fixes."""
|
|
|
|
|
+
|
|
|
|
|
+ @pytest.mark.asyncio
|
|
|
|
|
+ async def test_forgejo_handles_list_shape_ref_response(self):
|
|
|
|
|
+ backend = ForgejoBackend()
|
|
|
|
|
+ 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, {}))
|
|
|
|
|
+
|
|
|
|
|
+ result = await backend.push_files(
|
|
|
|
|
+ "https://forgejo.example.com/owner/repo",
|
|
|
|
|
+ "token",
|
|
|
|
|
+ "bambuddy-backup",
|
|
|
|
|
+ {"a.json": {"k": "v"}},
|
|
|
|
|
+ client,
|
|
|
|
|
+ )
|
|
|
|
|
+ assert result["status"] == "success"
|
|
|
|
|
+
|
|
|
|
|
+ @pytest.mark.asyncio
|
|
|
|
|
+ async def test_forgejo_empty_repo_uses_contents_api(self):
|
|
|
|
|
+ backend = ForgejoBackend()
|
|
|
|
|
+ client = AsyncMock()
|
|
|
|
|
+ client.get = AsyncMock(
|
|
|
|
|
+ side_effect=[
|
|
|
|
|
+ _make_mock_response(404, {}),
|
|
|
|
|
+ _make_mock_response(200, {"default_branch": "main"}),
|
|
|
|
|
+ _make_mock_response(404, {}),
|
|
|
|
|
+ ]
|
|
|
|
|
+ )
|
|
|
|
|
+ client.post = AsyncMock(return_value=_make_mock_response(201, {"commit": {"sha": "fj-sha"}}))
|
|
|
|
|
+
|
|
|
|
|
+ result = await backend.push_files(
|
|
|
|
|
+ "https://forgejo.example.com/owner/repo",
|
|
|
|
|
+ "token",
|
|
|
|
|
+ "main",
|
|
|
|
|
+ {"a.json": {"k": "v"}},
|
|
|
|
|
+ client,
|
|
|
|
|
+ )
|
|
|
|
|
+ assert result["status"] == "success"
|
|
|
|
|
+ contents_calls = [c for c in client.post.call_args_list if "/contents" in c.args[0]]
|
|
|
|
|
+ blob_calls = [c for c in client.post.call_args_list if "/git/blobs" in c.args[0]]
|
|
|
|
|
+ assert len(contents_calls) == 1
|
|
|
|
|
+ assert len(blob_calls) == 0
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
class TestForgejoBackendApiBase:
|
|
class TestForgejoBackendApiBase:
|
|
|
def setup_method(self):
|
|
def setup_method(self):
|
|
|
self.backend = ForgejoBackend()
|
|
self.backend = ForgejoBackend()
|
|
@@ -439,18 +689,17 @@ class TestGitLabBackendPushFiles:
|
|
|
"""Files beyond the first 100 are fetched; a file on page 2 is correctly skipped if unchanged."""
|
|
"""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"])
|
|
sha = _blob_sha(self.files["config/printers.json"])
|
|
|
page1_items = [{"type": "blob", "path": f"other{i}.json", "id": "aaa"} for i in range(100)]
|
|
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
|
|
|
|
|
|
|
+ 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 = AsyncMock()
|
|
|
client.get = AsyncMock(
|
|
client.get = AsyncMock(
|
|
|
side_effect=[
|
|
side_effect=[
|
|
|
_make_mock_response(200, {"name": self.branch}), # branch check
|
|
_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
|
|
|
|
|
|
|
+ _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
|
|
|
]
|
|
]
|
|
|
)
|
|
)
|
|
|
|
|
|