test: add 106 unit + smoke tests, 93% coverage
- pytest + respx + pytest-cov added as dev dependencies - tests/smoke/test_registration.py: verify all 66 tools registered and unique - tests/unit/: 12 files covering every tool — HTTP method, URL, params/body, errors - tests/config.test.json + conftest.py: zero-network test setup via GITEA_CONFIG - CLAUDE.md: developer guide with testing philosophy, pyramid, and commands - server/gitea_mcp.py: get_commit, list_pr_files tools added - README.md: tool count corrected to 66
This commit is contained in:
@@ -0,0 +1,89 @@
|
||||
import json
|
||||
import pytest
|
||||
import httpx
|
||||
import respx
|
||||
from server.gitea_mcp import list_branches, get_branch, create_branch, delete_branch, list_tags
|
||||
|
||||
BASE = "https://gitea.test/api/v1"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_branches():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/branches").mock(
|
||||
return_value=httpx.Response(200, json=[{"name": "main"}])
|
||||
)
|
||||
result = list_branches("owner", "repo")
|
||||
assert route.called
|
||||
assert result[0]["name"] == "main"
|
||||
assert dict(route.calls[0].request.url.params)["limit"] == "50"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_branch():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/branches/main").mock(
|
||||
return_value=httpx.Response(200, json={"name": "main"})
|
||||
)
|
||||
result = get_branch("owner", "repo", "main")
|
||||
assert route.called
|
||||
assert result["name"] == "main"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_branch_with_slash_in_name():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/branches/feature%2Ffoo").mock(
|
||||
return_value=httpx.Response(200, json={"name": "feature/foo"})
|
||||
)
|
||||
result = get_branch("owner", "repo", "feature/foo")
|
||||
assert route.called
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_branch_not_found():
|
||||
respx.get(f"{BASE}/repos/owner/repo/branches/missing").mock(
|
||||
return_value=httpx.Response(404, json={"message": "not found"})
|
||||
)
|
||||
with pytest.raises(RuntimeError, match="404"):
|
||||
get_branch("owner", "repo", "missing")
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_create_branch():
|
||||
route = respx.post(f"{BASE}/repos/owner/repo/branches").mock(
|
||||
return_value=httpx.Response(201, json={"name": "feature/new"})
|
||||
)
|
||||
result = create_branch("owner", "repo", "feature/new")
|
||||
assert route.called
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["new_branch_name"] == "feature/new"
|
||||
assert "old_branch_name" not in body
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_create_branch_from_specific():
|
||||
route = respx.post(f"{BASE}/repos/owner/repo/branches").mock(
|
||||
return_value=httpx.Response(201, json={"name": "hotfix"})
|
||||
)
|
||||
create_branch("owner", "repo", "hotfix", old_branch_name="main")
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["old_branch_name"] == "main"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_delete_branch():
|
||||
route = respx.delete(f"{BASE}/repos/owner/repo/branches/old-branch").mock(
|
||||
return_value=httpx.Response(204)
|
||||
)
|
||||
result = delete_branch("owner", "repo", "old-branch")
|
||||
assert route.called
|
||||
assert result == {"success": True}
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_tags():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/tags").mock(
|
||||
return_value=httpx.Response(200, json=[{"name": "v1.0.0"}])
|
||||
)
|
||||
result = list_tags("owner", "repo")
|
||||
assert route.called
|
||||
assert result[0]["name"] == "v1.0.0"
|
||||
assert dict(route.calls[0].request.url.params)["limit"] == "20"
|
||||
@@ -0,0 +1,94 @@
|
||||
import pytest
|
||||
import httpx
|
||||
import respx
|
||||
from server.gitea_mcp import list_commits, get_commit, compare_branches, list_repo_contributors
|
||||
|
||||
BASE = "https://gitea.test/api/v1"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_commits_defaults():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/commits").mock(
|
||||
return_value=httpx.Response(200, json=[{"sha": "abc123"}])
|
||||
)
|
||||
result = list_commits("owner", "repo")
|
||||
assert route.called
|
||||
params = dict(route.calls[0].request.url.params)
|
||||
assert params["limit"] == "20"
|
||||
assert params["page"] == "1"
|
||||
assert "sha" not in params
|
||||
assert "path" not in params
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_commits_on_branch_and_path():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/commits").mock(
|
||||
return_value=httpx.Response(200, json=[])
|
||||
)
|
||||
list_commits("owner", "repo", sha="develop", path="server/gitea_mcp.py")
|
||||
params = dict(route.calls[0].request.url.params)
|
||||
assert params["sha"] == "develop"
|
||||
assert params["path"] == "server/gitea_mcp.py"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_commit():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/git/commits/abc123").mock(
|
||||
return_value=httpx.Response(200, json={"sha": "abc123", "commit": {"message": "fix bug"}})
|
||||
)
|
||||
result = get_commit("owner", "repo", "abc123")
|
||||
assert route.called
|
||||
assert result["sha"] == "abc123"
|
||||
params = dict(route.calls[0].request.url.params)
|
||||
assert params["stat"] == "true"
|
||||
assert params["files"] == "true"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_commit_without_stat():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/git/commits/deadbeef").mock(
|
||||
return_value=httpx.Response(200, json={"sha": "deadbeef"})
|
||||
)
|
||||
get_commit("owner", "repo", "deadbeef", stat=False, files=False)
|
||||
params = dict(route.calls[0].request.url.params)
|
||||
assert params["stat"] == "false"
|
||||
assert params["files"] == "false"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_commit_not_found():
|
||||
respx.get(f"{BASE}/repos/owner/repo/git/commits/notacommit").mock(
|
||||
return_value=httpx.Response(404, json={"message": "not found"})
|
||||
)
|
||||
with pytest.raises(RuntimeError, match="404"):
|
||||
get_commit("owner", "repo", "notacommit")
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_compare_branches():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/compare/main...develop").mock(
|
||||
return_value=httpx.Response(200, json={"total_commits": 3, "commits": []})
|
||||
)
|
||||
result = compare_branches("owner", "repo", "main", "develop")
|
||||
assert route.called
|
||||
assert result["total_commits"] == 3
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_compare_branches_with_slashes():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/compare/release%2F1.0...feature%2Fnew").mock(
|
||||
return_value=httpx.Response(200, json={"total_commits": 1})
|
||||
)
|
||||
result = compare_branches("owner", "repo", "release/1.0", "feature/new")
|
||||
assert route.called
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_repo_contributors():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/contributors").mock(
|
||||
return_value=httpx.Response(200, json=[{"login": "alice", "contributions": 42}])
|
||||
)
|
||||
result = list_repo_contributors("owner", "repo")
|
||||
assert route.called
|
||||
assert result[0]["login"] == "alice"
|
||||
assert dict(route.calls[0].request.url.params)["limit"] == "20"
|
||||
@@ -0,0 +1,108 @@
|
||||
import base64
|
||||
import json
|
||||
import pytest
|
||||
import httpx
|
||||
import respx
|
||||
from server.gitea_mcp import list_directory, get_file, create_file, update_file, delete_file
|
||||
|
||||
BASE = "https://gitea.test/api/v1"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_directory_root():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/contents/").mock(
|
||||
return_value=httpx.Response(200, json=[{"name": "README.md", "type": "file"}])
|
||||
)
|
||||
result = list_directory("owner", "repo")
|
||||
assert route.called
|
||||
assert result[0]["name"] == "README.md"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_directory_with_ref():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/contents/src").mock(
|
||||
return_value=httpx.Response(200, json=[])
|
||||
)
|
||||
list_directory("owner", "repo", path="src", ref="develop")
|
||||
params = dict(route.calls[0].request.url.params)
|
||||
assert params["ref"] == "develop"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_file():
|
||||
encoded = base64.b64encode(b"hello world").decode()
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/contents/README.md").mock(
|
||||
return_value=httpx.Response(200, json={"name": "README.md", "content": encoded, "sha": "abc"})
|
||||
)
|
||||
result = get_file("owner", "repo", "README.md")
|
||||
assert route.called
|
||||
assert result["content"] == encoded
|
||||
assert result["sha"] == "abc"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_file_with_ref():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/contents/README.md").mock(
|
||||
return_value=httpx.Response(200, json={"name": "README.md"})
|
||||
)
|
||||
get_file("owner", "repo", "README.md", ref="v1.0")
|
||||
params = dict(route.calls[0].request.url.params)
|
||||
assert params["ref"] == "v1.0"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_file_not_found():
|
||||
respx.get(f"{BASE}/repos/owner/repo/contents/missing.txt").mock(
|
||||
return_value=httpx.Response(404, json={"message": "not found"})
|
||||
)
|
||||
with pytest.raises(RuntimeError, match="404"):
|
||||
get_file("owner", "repo", "missing.txt")
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_create_file():
|
||||
route = respx.post(f"{BASE}/repos/owner/repo/contents/hello.txt").mock(
|
||||
return_value=httpx.Response(201, json={"content": {"name": "hello.txt"}})
|
||||
)
|
||||
result = create_file("owner", "repo", "hello.txt", "hello world", "add hello")
|
||||
assert route.called
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["content"] == base64.b64encode(b"hello world").decode()
|
||||
assert body["message"] == "add hello"
|
||||
assert "branch" not in body
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_create_file_on_branch():
|
||||
route = respx.post(f"{BASE}/repos/owner/repo/contents/file.py").mock(
|
||||
return_value=httpx.Response(201, json={"content": {"name": "file.py"}})
|
||||
)
|
||||
create_file("owner", "repo", "file.py", "print('hi')", "add file", branch="develop")
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["branch"] == "develop"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_update_file():
|
||||
new_content = "updated content"
|
||||
route = respx.put(f"{BASE}/repos/owner/repo/contents/README.md").mock(
|
||||
return_value=httpx.Response(200, json={"content": {"name": "README.md"}})
|
||||
)
|
||||
update_file("owner", "repo", "README.md", new_content, "update readme", sha="deadbeef")
|
||||
assert route.called
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["content"] == base64.b64encode(new_content.encode()).decode()
|
||||
assert body["sha"] == "deadbeef"
|
||||
assert body["message"] == "update readme"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_delete_file():
|
||||
route = respx.delete(f"{BASE}/repos/owner/repo/contents/old.txt").mock(
|
||||
return_value=httpx.Response(200, json={"commit": {"sha": "abc"}})
|
||||
)
|
||||
result = delete_file("owner", "repo", "old.txt", "remove old file", sha="abc123")
|
||||
assert route.called
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["message"] == "remove old file"
|
||||
assert body["sha"] == "abc123"
|
||||
@@ -0,0 +1,149 @@
|
||||
import json
|
||||
import pytest
|
||||
import httpx
|
||||
import respx
|
||||
from server.gitea_mcp import (
|
||||
list_issues, get_issue, create_issue, update_issue, close_issue, reopen_issue,
|
||||
list_issue_comments, add_issue_comment, edit_issue_comment, delete_issue_comment,
|
||||
)
|
||||
|
||||
BASE = "https://gitea.test/api/v1"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_issues_defaults():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/issues").mock(
|
||||
return_value=httpx.Response(200, json=[])
|
||||
)
|
||||
list_issues("owner", "repo")
|
||||
params = dict(route.calls[0].request.url.params)
|
||||
assert params["state"] == "open"
|
||||
assert params["type"] == "issues"
|
||||
assert params["limit"] == "20"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_issues_with_filters():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/issues").mock(
|
||||
return_value=httpx.Response(200, json=[])
|
||||
)
|
||||
list_issues("owner", "repo", state="closed", labels="bug", assignee="alice")
|
||||
params = dict(route.calls[0].request.url.params)
|
||||
assert params["state"] == "closed"
|
||||
assert params["labels"] == "bug"
|
||||
assert params["assignee"] == "alice"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_issue():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/issues/42").mock(
|
||||
return_value=httpx.Response(200, json={"number": 42, "title": "Bug"})
|
||||
)
|
||||
result = get_issue("owner", "repo", 42)
|
||||
assert route.called
|
||||
assert result["number"] == 42
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_issue_not_found():
|
||||
respx.get(f"{BASE}/repos/owner/repo/issues/999").mock(
|
||||
return_value=httpx.Response(404, json={"message": "not found"})
|
||||
)
|
||||
with pytest.raises(RuntimeError, match="404"):
|
||||
get_issue("owner", "repo", 999)
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_create_issue():
|
||||
route = respx.post(f"{BASE}/repos/owner/repo/issues").mock(
|
||||
return_value=httpx.Response(201, json={"number": 1, "title": "New bug"})
|
||||
)
|
||||
result = create_issue("owner", "repo", "New bug", body="description")
|
||||
assert route.called
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["title"] == "New bug"
|
||||
assert body["body"] == "description"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_create_issue_with_assignees_and_labels():
|
||||
route = respx.post(f"{BASE}/repos/owner/repo/issues").mock(
|
||||
return_value=httpx.Response(201, json={"number": 2})
|
||||
)
|
||||
create_issue("owner", "repo", "Task", assignees=["alice"], labels=[1, 2])
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["assignees"] == ["alice"]
|
||||
assert body["labels"] == [1, 2]
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_update_issue():
|
||||
route = respx.patch(f"{BASE}/repos/owner/repo/issues/5").mock(
|
||||
return_value=httpx.Response(200, json={"number": 5})
|
||||
)
|
||||
update_issue("owner", "repo", 5, title="Updated title")
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["title"] == "Updated title"
|
||||
assert "body" not in body
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_close_issue():
|
||||
route = respx.patch(f"{BASE}/repos/owner/repo/issues/3").mock(
|
||||
return_value=httpx.Response(200, json={"state": "closed"})
|
||||
)
|
||||
result = close_issue("owner", "repo", 3)
|
||||
assert route.called
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["state"] == "closed"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_reopen_issue():
|
||||
route = respx.patch(f"{BASE}/repos/owner/repo/issues/3").mock(
|
||||
return_value=httpx.Response(200, json={"state": "open"})
|
||||
)
|
||||
reopen_issue("owner", "repo", 3)
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["state"] == "open"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_issue_comments():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/issues/7/comments").mock(
|
||||
return_value=httpx.Response(200, json=[{"id": 1, "body": "hi"}])
|
||||
)
|
||||
result = list_issue_comments("owner", "repo", 7)
|
||||
assert route.called
|
||||
assert result[0]["id"] == 1
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_add_issue_comment():
|
||||
route = respx.post(f"{BASE}/repos/owner/repo/issues/7/comments").mock(
|
||||
return_value=httpx.Response(201, json={"id": 10, "body": "LGTM"})
|
||||
)
|
||||
result = add_issue_comment("owner", "repo", 7, "LGTM")
|
||||
assert route.called
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["body"] == "LGTM"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_edit_issue_comment():
|
||||
route = respx.patch(f"{BASE}/repos/owner/repo/issues/comments/42").mock(
|
||||
return_value=httpx.Response(200, json={"id": 42})
|
||||
)
|
||||
edit_issue_comment("owner", "repo", 42, "edited body")
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["body"] == "edited body"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_delete_issue_comment():
|
||||
route = respx.delete(f"{BASE}/repos/owner/repo/issues/comments/42").mock(
|
||||
return_value=httpx.Response(204)
|
||||
)
|
||||
result = delete_issue_comment("owner", "repo", 42)
|
||||
assert route.called
|
||||
assert result == {"success": True}
|
||||
@@ -0,0 +1,105 @@
|
||||
import json
|
||||
import pytest
|
||||
import httpx
|
||||
import respx
|
||||
from server.gitea_mcp import (
|
||||
list_labels, create_label, add_issue_labels, list_milestones, create_milestone,
|
||||
)
|
||||
|
||||
BASE = "https://gitea.test/api/v1"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_labels():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/labels").mock(
|
||||
return_value=httpx.Response(200, json=[{"id": 1, "name": "bug", "color": "#ee0701"}])
|
||||
)
|
||||
result = list_labels("owner", "repo")
|
||||
assert route.called
|
||||
assert result[0]["name"] == "bug"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_create_label():
|
||||
route = respx.post(f"{BASE}/repos/owner/repo/labels").mock(
|
||||
return_value=httpx.Response(201, json={"id": 5, "name": "enhancement"})
|
||||
)
|
||||
result = create_label("owner", "repo", "enhancement", "#84b6eb")
|
||||
assert route.called
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["name"] == "enhancement"
|
||||
assert body["color"] == "#84b6eb"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_create_label_with_description():
|
||||
route = respx.post(f"{BASE}/repos/owner/repo/labels").mock(
|
||||
return_value=httpx.Response(201, json={"id": 6})
|
||||
)
|
||||
create_label("owner", "repo", "wontfix", "#ffffff", description="Not going to fix")
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["description"] == "Not going to fix"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_create_label_error():
|
||||
respx.post(f"{BASE}/repos/owner/repo/labels").mock(
|
||||
return_value=httpx.Response(422, json={"message": "validation failed"})
|
||||
)
|
||||
with pytest.raises(RuntimeError, match="422"):
|
||||
create_label("owner", "repo", "bad", "notacolor")
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_add_issue_labels():
|
||||
route = respx.post(f"{BASE}/repos/owner/repo/issues/3/labels").mock(
|
||||
return_value=httpx.Response(200, json=[{"id": 1}])
|
||||
)
|
||||
result = add_issue_labels("owner", "repo", 3, [1, 2])
|
||||
assert route.called
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["labels"] == [1, 2]
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_milestones():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/milestones").mock(
|
||||
return_value=httpx.Response(200, json=[{"id": 1, "title": "v1.0"}])
|
||||
)
|
||||
result = list_milestones("owner", "repo")
|
||||
assert route.called
|
||||
assert result[0]["title"] == "v1.0"
|
||||
assert dict(route.calls[0].request.url.params)["state"] == "open"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_milestones_closed():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/milestones").mock(
|
||||
return_value=httpx.Response(200, json=[])
|
||||
)
|
||||
list_milestones("owner", "repo", state="closed")
|
||||
assert dict(route.calls[0].request.url.params)["state"] == "closed"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_create_milestone():
|
||||
route = respx.post(f"{BASE}/repos/owner/repo/milestones").mock(
|
||||
return_value=httpx.Response(201, json={"id": 2, "title": "v2.0"})
|
||||
)
|
||||
result = create_milestone("owner", "repo", "v2.0")
|
||||
assert route.called
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["title"] == "v2.0"
|
||||
assert "description" not in body
|
||||
assert "due_on" not in body
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_create_milestone_with_due_date():
|
||||
route = respx.post(f"{BASE}/repos/owner/repo/milestones").mock(
|
||||
return_value=httpx.Response(201, json={"id": 3})
|
||||
)
|
||||
create_milestone("owner", "repo", "v3.0", description="Major release", due_on="2025-12-31T00:00:00Z")
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["description"] == "Major release"
|
||||
assert body["due_on"] == "2025-12-31T00:00:00Z"
|
||||
@@ -0,0 +1,46 @@
|
||||
import pytest
|
||||
import httpx
|
||||
import respx
|
||||
from server.gitea_mcp import list_instances, get_server_info, get_current_user
|
||||
|
||||
BASE = "https://gitea.test/api/v1"
|
||||
|
||||
|
||||
def test_list_instances():
|
||||
result = list_instances()
|
||||
assert result["default"] == "default"
|
||||
assert "default" in result["instances"]
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_server_info():
|
||||
route = respx.get(f"{BASE}/version").mock(
|
||||
return_value=httpx.Response(200, json={"version": "1.22.0"})
|
||||
)
|
||||
result = get_server_info()
|
||||
assert route.called
|
||||
assert result == {"version": "1.22.0"}
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_server_info_error():
|
||||
respx.get(f"{BASE}/version").mock(return_value=httpx.Response(500, json={"message": "internal error"}))
|
||||
with pytest.raises(RuntimeError, match="500"):
|
||||
get_server_info()
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_current_user():
|
||||
route = respx.get(f"{BASE}/user").mock(
|
||||
return_value=httpx.Response(200, json={"login": "alice", "id": 1})
|
||||
)
|
||||
result = get_current_user()
|
||||
assert route.called
|
||||
assert result["login"] == "alice"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_current_user_error():
|
||||
respx.get(f"{BASE}/user").mock(return_value=httpx.Response(401, json={"message": "unauthorized"}))
|
||||
with pytest.raises(RuntimeError, match="401"):
|
||||
get_current_user()
|
||||
@@ -0,0 +1,47 @@
|
||||
import json
|
||||
import pytest
|
||||
import httpx
|
||||
import respx
|
||||
from server.gitea_mcp import list_notifications, mark_all_notifications_read
|
||||
|
||||
BASE = "https://gitea.test/api/v1"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_notifications_defaults():
|
||||
route = respx.get(f"{BASE}/notifications").mock(
|
||||
return_value=httpx.Response(200, json=[{"id": 1, "unread": True}])
|
||||
)
|
||||
result = list_notifications()
|
||||
assert route.called
|
||||
params = dict(route.calls[0].request.url.params)
|
||||
assert params["limit"] == "20"
|
||||
assert params["all"] == "false"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_notifications_include_read():
|
||||
route = respx.get(f"{BASE}/notifications").mock(
|
||||
return_value=httpx.Response(200, json=[])
|
||||
)
|
||||
list_notifications(include_read=True)
|
||||
params = dict(route.calls[0].request.url.params)
|
||||
assert params["all"] == "true"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_notifications_error():
|
||||
respx.get(f"{BASE}/notifications").mock(
|
||||
return_value=httpx.Response(401, json={"message": "unauthorized"})
|
||||
)
|
||||
with pytest.raises(RuntimeError, match="401"):
|
||||
list_notifications()
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_mark_all_notifications_read():
|
||||
route = respx.put(f"{BASE}/notifications").mock(
|
||||
return_value=httpx.Response(205)
|
||||
)
|
||||
result = mark_all_notifications_read()
|
||||
assert route.called
|
||||
@@ -0,0 +1,180 @@
|
||||
import json
|
||||
import pytest
|
||||
import httpx
|
||||
import respx
|
||||
from server.gitea_mcp import (
|
||||
list_pull_requests, get_pull_request, create_pull_request, update_pull_request,
|
||||
merge_pull_request, get_pr_diff, list_pr_files, list_pr_reviews, create_pr_review,
|
||||
)
|
||||
|
||||
BASE = "https://gitea.test/api/v1"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_pull_requests():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/pulls").mock(
|
||||
return_value=httpx.Response(200, json=[])
|
||||
)
|
||||
list_pull_requests("owner", "repo")
|
||||
params = dict(route.calls[0].request.url.params)
|
||||
assert params["state"] == "open"
|
||||
assert params["limit"] == "20"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_pull_requests_closed():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/pulls").mock(
|
||||
return_value=httpx.Response(200, json=[])
|
||||
)
|
||||
list_pull_requests("owner", "repo", state="closed", limit=5)
|
||||
params = dict(route.calls[0].request.url.params)
|
||||
assert params["state"] == "closed"
|
||||
assert params["limit"] == "5"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_pull_request():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/pulls/7").mock(
|
||||
return_value=httpx.Response(200, json={"number": 7, "title": "Add feature"})
|
||||
)
|
||||
result = get_pull_request("owner", "repo", 7)
|
||||
assert route.called
|
||||
assert result["number"] == 7
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_create_pull_request():
|
||||
route = respx.post(f"{BASE}/repos/owner/repo/pulls").mock(
|
||||
return_value=httpx.Response(201, json={"number": 10})
|
||||
)
|
||||
result = create_pull_request("owner", "repo", "Add feature", "feature/x", "main")
|
||||
assert route.called
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["title"] == "Add feature"
|
||||
assert body["head"] == "feature/x"
|
||||
assert body["base"] == "main"
|
||||
assert "body" not in body
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_create_pull_request_with_body_and_labels():
|
||||
route = respx.post(f"{BASE}/repos/owner/repo/pulls").mock(
|
||||
return_value=httpx.Response(201, json={"number": 11})
|
||||
)
|
||||
create_pull_request(
|
||||
"owner", "repo", "Fix bug", "fix/crash", "main",
|
||||
body="Fixes crash on startup", labels=[3],
|
||||
)
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["body"] == "Fixes crash on startup"
|
||||
assert body["labels"] == [3]
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_update_pull_request():
|
||||
route = respx.patch(f"{BASE}/repos/owner/repo/pulls/5").mock(
|
||||
return_value=httpx.Response(200, json={"number": 5})
|
||||
)
|
||||
update_pull_request("owner", "repo", 5, title="New title")
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["title"] == "New title"
|
||||
assert "body" not in body
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_merge_pull_request_defaults():
|
||||
route = respx.post(f"{BASE}/repos/owner/repo/pulls/3/merge").mock(
|
||||
return_value=httpx.Response(200, json={"success": True})
|
||||
)
|
||||
merge_pull_request("owner", "repo", 3)
|
||||
assert route.called
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["Do"] == "merge"
|
||||
assert body["delete_branch_after_merge"] is False
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_merge_pull_request_squash_with_delete():
|
||||
route = respx.post(f"{BASE}/repos/owner/repo/pulls/4/merge").mock(
|
||||
return_value=httpx.Response(200, json={"success": True})
|
||||
)
|
||||
merge_pull_request("owner", "repo", 4, merge_style="squash", delete_branch_after_merge=True)
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["Do"] == "squash"
|
||||
assert body["delete_branch_after_merge"] is True
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_pr_diff():
|
||||
diff_text = "diff --git a/file.py b/file.py\n+added line\n"
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/pulls/2.diff").mock(
|
||||
return_value=httpx.Response(200, text=diff_text)
|
||||
)
|
||||
result = get_pr_diff("owner", "repo", 2)
|
||||
assert route.called
|
||||
assert result["diff"] == diff_text
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_pr_diff_error():
|
||||
respx.get(f"{BASE}/repos/owner/repo/pulls/99.diff").mock(
|
||||
return_value=httpx.Response(404, text="not found")
|
||||
)
|
||||
with pytest.raises(RuntimeError, match="404"):
|
||||
get_pr_diff("owner", "repo", 99)
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_pr_files():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/pulls/5/files").mock(
|
||||
return_value=httpx.Response(200, json=[{"filename": "server/gitea_mcp.py", "status": "modified"}])
|
||||
)
|
||||
result = list_pr_files("owner", "repo", 5)
|
||||
assert route.called
|
||||
assert result[0]["filename"] == "server/gitea_mcp.py"
|
||||
params = dict(route.calls[0].request.url.params)
|
||||
assert params["limit"] == "50"
|
||||
assert params["page"] == "1"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_pr_files_with_whitespace():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/pulls/6/files").mock(
|
||||
return_value=httpx.Response(200, json=[])
|
||||
)
|
||||
list_pr_files("owner", "repo", 6, whitespace="ignore-all")
|
||||
params = dict(route.calls[0].request.url.params)
|
||||
assert params["whitespace"] == "ignore-all"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_pr_reviews():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/pulls/8/reviews").mock(
|
||||
return_value=httpx.Response(200, json=[{"id": 1, "state": "APPROVED"}])
|
||||
)
|
||||
result = list_pr_reviews("owner", "repo", 8)
|
||||
assert route.called
|
||||
assert result[0]["state"] == "APPROVED"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_create_pr_review_approved():
|
||||
route = respx.post(f"{BASE}/repos/owner/repo/pulls/9/reviews").mock(
|
||||
return_value=httpx.Response(200, json={"id": 5, "state": "APPROVED"})
|
||||
)
|
||||
result = create_pr_review("owner", "repo", 9, "APPROVED", body="Looks good!")
|
||||
assert route.called
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["event"] == "APPROVED"
|
||||
assert body["body"] == "Looks good!"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_create_pr_review_request_changes():
|
||||
route = respx.post(f"{BASE}/repos/owner/repo/pulls/10/reviews").mock(
|
||||
return_value=httpx.Response(200, json={"id": 6, "state": "REQUEST_CHANGES"})
|
||||
)
|
||||
create_pr_review("owner", "repo", 10, "REQUEST_CHANGES")
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["event"] == "REQUEST_CHANGES"
|
||||
assert "body" not in body
|
||||
@@ -0,0 +1,79 @@
|
||||
import json
|
||||
import pytest
|
||||
import httpx
|
||||
import respx
|
||||
from server.gitea_mcp import list_releases, get_latest_release, create_release, delete_release
|
||||
|
||||
BASE = "https://gitea.test/api/v1"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_releases():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/releases").mock(
|
||||
return_value=httpx.Response(200, json=[{"id": 1, "tag_name": "v1.0.0"}])
|
||||
)
|
||||
result = list_releases("owner", "repo")
|
||||
assert route.called
|
||||
assert result[0]["tag_name"] == "v1.0.0"
|
||||
assert dict(route.calls[0].request.url.params)["limit"] == "10"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_latest_release():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/releases/latest").mock(
|
||||
return_value=httpx.Response(200, json={"id": 3, "tag_name": "v2.0.0"})
|
||||
)
|
||||
result = get_latest_release("owner", "repo")
|
||||
assert route.called
|
||||
assert result["tag_name"] == "v2.0.0"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_latest_release_not_found():
|
||||
respx.get(f"{BASE}/repos/owner/repo/releases/latest").mock(
|
||||
return_value=httpx.Response(404, json={"message": "not found"})
|
||||
)
|
||||
with pytest.raises(RuntimeError, match="404"):
|
||||
get_latest_release("owner", "repo")
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_create_release():
|
||||
route = respx.post(f"{BASE}/repos/owner/repo/releases").mock(
|
||||
return_value=httpx.Response(201, json={"id": 5, "tag_name": "v3.0.0"})
|
||||
)
|
||||
result = create_release("owner", "repo", "v3.0.0", "Release 3.0.0")
|
||||
assert route.called
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["tag_name"] == "v3.0.0"
|
||||
assert body["name"] == "Release 3.0.0"
|
||||
assert body["draft"] is False
|
||||
assert body["prerelease"] is False
|
||||
assert "body" not in body
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_create_release_draft_with_notes():
|
||||
route = respx.post(f"{BASE}/repos/owner/repo/releases").mock(
|
||||
return_value=httpx.Response(201, json={"id": 6})
|
||||
)
|
||||
create_release(
|
||||
"owner", "repo", "v4.0.0-beta", "Beta release",
|
||||
body="## Changes\n- new feature", draft=True, prerelease=True,
|
||||
target_commitish="develop",
|
||||
)
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["body"] == "## Changes\n- new feature"
|
||||
assert body["draft"] is True
|
||||
assert body["prerelease"] is True
|
||||
assert body["target_commitish"] == "develop"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_delete_release():
|
||||
route = respx.delete(f"{BASE}/repos/owner/repo/releases/5").mock(
|
||||
return_value=httpx.Response(204)
|
||||
)
|
||||
result = delete_release("owner", "repo", 5)
|
||||
assert route.called
|
||||
assert result == {"success": True}
|
||||
@@ -0,0 +1,151 @@
|
||||
import json
|
||||
import pytest
|
||||
import httpx
|
||||
import respx
|
||||
from server.gitea_mcp import (
|
||||
list_my_repos, search_repos, get_repo, create_repo, create_org_repo,
|
||||
update_repo, delete_repo, fork_repo, set_repo_topics,
|
||||
)
|
||||
|
||||
BASE = "https://gitea.test/api/v1"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_my_repos_defaults():
|
||||
route = respx.get(f"{BASE}/user/repos").mock(return_value=httpx.Response(200, json=[]))
|
||||
list_my_repos()
|
||||
assert route.called
|
||||
params = dict(route.calls[0].request.url.params)
|
||||
assert params["limit"] == "50"
|
||||
assert params["page"] == "1"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_my_repos_custom_page():
|
||||
route = respx.get(f"{BASE}/user/repos").mock(return_value=httpx.Response(200, json=[]))
|
||||
list_my_repos(limit=10, page=3)
|
||||
assert route.called
|
||||
params = dict(route.calls[0].request.url.params)
|
||||
assert params["limit"] == "10"
|
||||
assert params["page"] == "3"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_search_repos():
|
||||
route = respx.get(f"{BASE}/repos/search").mock(
|
||||
return_value=httpx.Response(200, json={"data": []})
|
||||
)
|
||||
result = search_repos("myquery")
|
||||
assert route.called
|
||||
params = dict(route.calls[0].request.url.params)
|
||||
assert params["q"] == "myquery"
|
||||
assert params["limit"] == "20"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_repo():
|
||||
route = respx.get(f"{BASE}/repos/owner/myrepo").mock(
|
||||
return_value=httpx.Response(200, json={"name": "myrepo", "full_name": "owner/myrepo"})
|
||||
)
|
||||
result = get_repo("owner", "myrepo")
|
||||
assert route.called
|
||||
assert result["name"] == "myrepo"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_repo_not_found():
|
||||
respx.get(f"{BASE}/repos/owner/missing").mock(
|
||||
return_value=httpx.Response(404, json={"message": "not found"})
|
||||
)
|
||||
with pytest.raises(RuntimeError, match="404"):
|
||||
get_repo("owner", "missing")
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_create_repo_defaults():
|
||||
route = respx.post(f"{BASE}/user/repos").mock(
|
||||
return_value=httpx.Response(201, json={"name": "newrepo"})
|
||||
)
|
||||
result = create_repo("newrepo")
|
||||
assert route.called
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["name"] == "newrepo"
|
||||
assert body["private"] is True
|
||||
assert body["auto_init"] is True
|
||||
assert "description" not in body
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_create_repo_with_options():
|
||||
route = respx.post(f"{BASE}/user/repos").mock(
|
||||
return_value=httpx.Response(201, json={"name": "pub"})
|
||||
)
|
||||
create_repo("pub", description="A public repo", private=False, gitignores="Python")
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["description"] == "A public repo"
|
||||
assert body["private"] is False
|
||||
assert body["gitignores"] == "Python"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_create_org_repo():
|
||||
route = respx.post(f"{BASE}/orgs/myorg/repos").mock(
|
||||
return_value=httpx.Response(201, json={"name": "orgrepo"})
|
||||
)
|
||||
result = create_org_repo("myorg", "orgrepo")
|
||||
assert route.called
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["name"] == "orgrepo"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_update_repo():
|
||||
route = respx.patch(f"{BASE}/repos/owner/repo").mock(
|
||||
return_value=httpx.Response(200, json={"name": "repo"})
|
||||
)
|
||||
update_repo("owner", "repo", description="updated")
|
||||
assert route.called
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["description"] == "updated"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_delete_repo():
|
||||
route = respx.delete(f"{BASE}/repos/owner/repo").mock(
|
||||
return_value=httpx.Response(204)
|
||||
)
|
||||
result = delete_repo("owner", "repo")
|
||||
assert route.called
|
||||
assert result == {"success": True}
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_fork_repo():
|
||||
route = respx.post(f"{BASE}/repos/upstream/repo/forks").mock(
|
||||
return_value=httpx.Response(202, json={"name": "repo"})
|
||||
)
|
||||
fork_repo("upstream", "repo")
|
||||
assert route.called
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_fork_repo_to_org():
|
||||
route = respx.post(f"{BASE}/repos/upstream/repo/forks").mock(
|
||||
return_value=httpx.Response(202, json={"name": "myfork"})
|
||||
)
|
||||
fork_repo("upstream", "repo", organization="myorg", new_repo_name="myfork")
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["organization"] == "myorg"
|
||||
assert body["name"] == "myfork"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_set_repo_topics():
|
||||
route = respx.put(f"{BASE}/repos/owner/repo/topics").mock(
|
||||
return_value=httpx.Response(204)
|
||||
)
|
||||
result = set_repo_topics("owner", "repo", ["python", "mcp"])
|
||||
assert route.called
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["topics"] == ["python", "mcp"]
|
||||
assert result == {"success": True}
|
||||
@@ -0,0 +1,90 @@
|
||||
import pytest
|
||||
import httpx
|
||||
import respx
|
||||
from server.gitea_mcp import (
|
||||
get_user, search_users, list_my_orgs, get_org,
|
||||
list_org_repos, list_org_members, list_org_teams,
|
||||
)
|
||||
|
||||
BASE = "https://gitea.test/api/v1"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_user():
|
||||
route = respx.get(f"{BASE}/users/alice").mock(
|
||||
return_value=httpx.Response(200, json={"login": "alice", "id": 42})
|
||||
)
|
||||
result = get_user("alice")
|
||||
assert route.called
|
||||
assert result["login"] == "alice"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_user_not_found():
|
||||
respx.get(f"{BASE}/users/nobody").mock(
|
||||
return_value=httpx.Response(404, json={"message": "not found"})
|
||||
)
|
||||
with pytest.raises(RuntimeError, match="404"):
|
||||
get_user("nobody")
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_search_users():
|
||||
route = respx.get(f"{BASE}/users/search").mock(
|
||||
return_value=httpx.Response(200, json={"data": [{"login": "alice"}]})
|
||||
)
|
||||
result = search_users("ali")
|
||||
assert route.called
|
||||
params = dict(route.calls[0].request.url.params)
|
||||
assert params["q"] == "ali"
|
||||
assert params["limit"] == "10"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_my_orgs():
|
||||
route = respx.get(f"{BASE}/user/orgs").mock(
|
||||
return_value=httpx.Response(200, json=[{"name": "myorg"}])
|
||||
)
|
||||
result = list_my_orgs()
|
||||
assert route.called
|
||||
assert result[0]["name"] == "myorg"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_get_org():
|
||||
route = respx.get(f"{BASE}/orgs/myorg").mock(
|
||||
return_value=httpx.Response(200, json={"name": "myorg", "full_name": "My Org"})
|
||||
)
|
||||
result = get_org("myorg")
|
||||
assert route.called
|
||||
assert result["name"] == "myorg"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_org_repos():
|
||||
route = respx.get(f"{BASE}/orgs/myorg/repos").mock(
|
||||
return_value=httpx.Response(200, json=[{"name": "repo1"}])
|
||||
)
|
||||
result = list_org_repos("myorg")
|
||||
assert route.called
|
||||
assert dict(route.calls[0].request.url.params)["limit"] == "50"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_org_members():
|
||||
route = respx.get(f"{BASE}/orgs/myorg/members").mock(
|
||||
return_value=httpx.Response(200, json=[{"login": "bob"}])
|
||||
)
|
||||
result = list_org_members("myorg")
|
||||
assert route.called
|
||||
assert result[0]["login"] == "bob"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_org_teams():
|
||||
route = respx.get(f"{BASE}/orgs/myorg/teams").mock(
|
||||
return_value=httpx.Response(200, json=[{"id": 1, "name": "developers"}])
|
||||
)
|
||||
result = list_org_teams("myorg")
|
||||
assert route.called
|
||||
assert result[0]["name"] == "developers"
|
||||
@@ -0,0 +1,68 @@
|
||||
import json
|
||||
import pytest
|
||||
import httpx
|
||||
import respx
|
||||
from server.gitea_mcp import list_repo_webhooks, create_repo_webhook, delete_repo_webhook
|
||||
|
||||
BASE = "https://gitea.test/api/v1"
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_list_repo_webhooks():
|
||||
route = respx.get(f"{BASE}/repos/owner/repo/hooks").mock(
|
||||
return_value=httpx.Response(200, json=[{"id": 1, "type": "gitea"}])
|
||||
)
|
||||
result = list_repo_webhooks("owner", "repo")
|
||||
assert route.called
|
||||
assert result[0]["id"] == 1
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_create_repo_webhook_minimal():
|
||||
route = respx.post(f"{BASE}/repos/owner/repo/hooks").mock(
|
||||
return_value=httpx.Response(201, json={"id": 2})
|
||||
)
|
||||
create_repo_webhook("owner", "repo", "https://example.com/hook", ["push"])
|
||||
assert route.called
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["type"] == "gitea"
|
||||
assert body["config"]["url"] == "https://example.com/hook"
|
||||
assert body["config"]["content_type"] == "json"
|
||||
assert "secret" not in body["config"]
|
||||
assert body["events"] == ["push"]
|
||||
assert body["active"] is True
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_create_repo_webhook_with_secret():
|
||||
route = respx.post(f"{BASE}/repos/owner/repo/hooks").mock(
|
||||
return_value=httpx.Response(201, json={"id": 3})
|
||||
)
|
||||
create_repo_webhook(
|
||||
"owner", "repo", "https://ci.example.com/hook",
|
||||
["push", "pull_request", "issues"],
|
||||
secret="mysecret", active=False,
|
||||
)
|
||||
body = json.loads(route.calls[0].request.content)
|
||||
assert body["config"]["secret"] == "mysecret"
|
||||
assert body["events"] == ["push", "pull_request", "issues"]
|
||||
assert body["active"] is False
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_delete_repo_webhook():
|
||||
route = respx.delete(f"{BASE}/repos/owner/repo/hooks/7").mock(
|
||||
return_value=httpx.Response(204)
|
||||
)
|
||||
result = delete_repo_webhook("owner", "repo", 7)
|
||||
assert route.called
|
||||
assert result == {"success": True}
|
||||
|
||||
|
||||
@respx.mock
|
||||
def test_create_repo_webhook_error():
|
||||
respx.post(f"{BASE}/repos/owner/repo/hooks").mock(
|
||||
return_value=httpx.Response(422, json={"message": "invalid url"})
|
||||
)
|
||||
with pytest.raises(RuntimeError, match="422"):
|
||||
create_repo_webhook("owner", "repo", "not-a-url", ["push"])
|
||||
Reference in New Issue
Block a user