232 lines
9.6 KiB
Python
232 lines
9.6 KiB
Python
"""Scenarios: organizations, teams and repositories, driven through the MCP server."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import time
|
|
from typing import Any
|
|
|
|
from tests.integration.conftest import McpSession, RegularUser, RepoRef, b64, drop_org
|
|
from tests.integration.harness import GiteaInstance
|
|
|
|
|
|
def names(items: Any, key: str = "name") -> list[str]:
|
|
"""The ``key`` member of every dict in a list result."""
|
|
return [item[key] for item in items]
|
|
|
|
|
|
async def test_org_and_team_lifecycle(
|
|
mcp: McpSession, gitea: GiteaInstance, regular_user: RegularUser
|
|
) -> None:
|
|
org = gitea.unique("it-org")
|
|
repo = gitea.unique("team-repo")
|
|
try:
|
|
created = await mcp.ok(
|
|
"gitea_org_create", status=201, body={"username": org, "visibility": "public"}
|
|
)
|
|
assert created.data["name"] == org
|
|
|
|
fetched = await mcp.ok("gitea_org_get", org=org)
|
|
assert fetched.data["name"] == org
|
|
|
|
team = await mcp.ok(
|
|
"gitea_team_create_team",
|
|
status=201,
|
|
org=org,
|
|
body={
|
|
"name": "devs",
|
|
"permission": "write",
|
|
"includes_all_repositories": False,
|
|
"units": ["repo.code", "repo.issues", "repo.pulls"],
|
|
},
|
|
)
|
|
team_id = team.data["id"]
|
|
assert team.data["name"] == "devs"
|
|
|
|
listed = await mcp.ok("gitea_team_list_teams", org=org)
|
|
assert {"devs", "Owners"} <= set(names(listed.data))
|
|
found = await mcp.ok("gitea_team_search", org=org, q="dev")
|
|
assert "devs" in names(found.data["data"])
|
|
|
|
await mcp.ok("gitea_team_add_team_member", id=team_id, username=regular_user.login)
|
|
members = await mcp.ok("gitea_team_list_team_members", id=team_id)
|
|
assert regular_user.login in names(members.data, "login")
|
|
await mcp.ok("gitea_team_list_team_member", id=team_id, username=regular_user.login)
|
|
assert regular_user.login in names(
|
|
(await mcp.ok("gitea_org_list_members", org=org)).data, "login"
|
|
)
|
|
|
|
await mcp.ok("gitea_org_create_org_repo", status=201, org=org, body={"name": repo})
|
|
await mcp.ok("gitea_team_add_team_repository", id=team_id, org=org, repo=repo)
|
|
repos = await mcp.ok("gitea_team_list_team_repos", id=team_id)
|
|
assert repo in names(repos.data)
|
|
await mcp.ok("gitea_team_list_team_repo", id=team_id, org=org, repo=repo)
|
|
assert (await mcp.ok("gitea_org_list_repos", org=org)).data
|
|
|
|
edited = await mcp.ok(
|
|
"gitea_team_edit_team", id=team_id, body={"name": "devs", "description": "the devs"}
|
|
)
|
|
assert edited.data["description"] == "the devs"
|
|
|
|
await mcp.ok("gitea_team_remove_team_repository", id=team_id, org=org, repo=repo)
|
|
await mcp.fail("gitea_team_list_team_repo", 404, id=team_id, org=org, repo=repo)
|
|
await mcp.ok("gitea_team_remove_team_member", id=team_id, username=regular_user.login)
|
|
await mcp.fail("gitea_team_list_team_member", 404, id=team_id, username=regular_user.login)
|
|
|
|
await mcp.ok("gitea_team_delete_team", id=team_id)
|
|
await mcp.fail("gitea_team_get_team", 404, id=team_id)
|
|
|
|
await mcp.ok("gitea_repo_delete", owner=org, repo=repo)
|
|
await mcp.ok("gitea_org_delete", org=org)
|
|
await mcp.fail("gitea_org_get", 404, org=org)
|
|
finally:
|
|
drop_org(gitea, org)
|
|
|
|
|
|
async def test_repo_files_branches_tags_and_topics(
|
|
mcp: McpSession, gitea: GiteaInstance, fresh_repo: RepoRef
|
|
) -> None:
|
|
repo = fresh_repo.args
|
|
info = await mcp.ok("gitea_repo_get", **repo)
|
|
assert info.data["full_name"] == fresh_repo.full
|
|
assert info.data["default_branch"] == "main"
|
|
|
|
# Files through the contents API: create, read, update, delete.
|
|
path = "docs/hello.txt"
|
|
created = await mcp.ok(
|
|
"gitea_repo_create_file",
|
|
status=201,
|
|
filepath=path,
|
|
body={"content": b64("hello"), "message": "add hello", "branch": "main"},
|
|
**repo,
|
|
)
|
|
assert created.data["content"]["path"] == path
|
|
got = await mcp.ok("gitea_repo_get_contents", filepath=path, **repo)
|
|
assert base64.b64decode(got.data["content"]) == b"hello"
|
|
sha = got.data["sha"]
|
|
updated = await mcp.ok(
|
|
"gitea_repo_update_file",
|
|
filepath=path,
|
|
body={"content": b64("hello again"), "sha": sha, "message": "update hello"},
|
|
**repo,
|
|
)
|
|
new_sha = updated.data["content"]["sha"]
|
|
assert new_sha != sha
|
|
raw = await mcp.ok("gitea_repo_get_raw_file", filepath=path, **repo)
|
|
body = raw.data
|
|
if isinstance(body, dict): # served as a binary type: base64 envelope
|
|
body = base64.b64decode(body["content_base64"]).decode("utf-8")
|
|
assert body == "hello again"
|
|
listing = await mcp.ok("gitea_repo_get_contents_list", **repo)
|
|
assert "README.md" in names(listing.data)
|
|
deleted = await mcp.ok(
|
|
"gitea_repo_delete_file",
|
|
filepath=path,
|
|
body={"sha": new_sha, "message": "remove hello"},
|
|
**repo,
|
|
)
|
|
assert deleted.data["commit"]["message"].startswith("remove hello")
|
|
await mcp.fail("gitea_repo_get_contents", 404, filepath=path, **repo)
|
|
|
|
# Branches.
|
|
await mcp.ok(
|
|
"gitea_repo_create_branch",
|
|
status=201,
|
|
body={"new_branch_name": "feature", "old_branch_name": "main"},
|
|
**repo,
|
|
)
|
|
assert "feature" in names((await mcp.ok("gitea_repo_list_branches", **repo)).data)
|
|
assert (await mcp.ok("gitea_repo_get_branch", branch="feature", **repo)).data["name"] == (
|
|
"feature"
|
|
)
|
|
await mcp.ok("gitea_repo_delete_branch", branch="feature", **repo)
|
|
await mcp.fail("gitea_repo_get_branch", 404, branch="feature", **repo)
|
|
|
|
# Tags.
|
|
tag = await mcp.ok(
|
|
"gitea_repo_create_tag",
|
|
status=201,
|
|
body={"tag_name": "v0.1.0", "target": "main", "message": "first"},
|
|
**repo,
|
|
)
|
|
assert tag.data["name"] == "v0.1.0"
|
|
assert "v0.1.0" in names((await mcp.ok("gitea_repo_list_tags", **repo)).data)
|
|
await mcp.ok("gitea_repo_get_tag", tag="v0.1.0", **repo)
|
|
await mcp.ok("gitea_repo_delete_tag", tag="v0.1.0", **repo)
|
|
await mcp.fail("gitea_repo_get_tag", 404, tag="v0.1.0", **repo)
|
|
|
|
# Topics.
|
|
await mcp.ok("gitea_repo_update_topics", body={"topics": ["aegis", "integration"]}, **repo)
|
|
assert set((await mcp.ok("gitea_repo_list_topics", **repo)).data["topics"]) == {
|
|
"aegis",
|
|
"integration",
|
|
}
|
|
await mcp.ok("gitea_repo_add_topic", topic="extra", **repo)
|
|
await mcp.ok("gitea_repo_delete_topic", topic="aegis", **repo)
|
|
assert set((await mcp.ok("gitea_repo_list_topics", **repo)).data["topics"]) == {
|
|
"integration",
|
|
"extra",
|
|
}
|
|
|
|
# Edit and read back.
|
|
edited = await mcp.ok("gitea_repo_edit", body={"description": "edited by it"}, **repo)
|
|
assert edited.data["description"] == "edited by it"
|
|
|
|
|
|
async def test_repo_collaborators_fork_transfer_archive_search(
|
|
mcp: McpSession, gitea: GiteaInstance, regular_user: RegularUser, fresh_repo: RepoRef
|
|
) -> None:
|
|
repo = fresh_repo.args
|
|
me = regular_user.login
|
|
|
|
# Collaborators.
|
|
await mcp.ok(
|
|
"gitea_repo_add_collaborator", collaborator=me, body={"permission": "write"}, **repo
|
|
)
|
|
await mcp.ok("gitea_repo_check_collaborator", collaborator=me, **repo)
|
|
assert me in names((await mcp.ok("gitea_repo_list_collaborators", **repo)).data, "login")
|
|
permission = await mcp.ok("gitea_repo_get_repo_permissions", collaborator=me, **repo)
|
|
assert permission.data["permission"] == "write"
|
|
await mcp.ok("gitea_repo_delete_collaborator", collaborator=me, **repo)
|
|
await mcp.fail("gitea_repo_check_collaborator", 404, collaborator=me, **repo)
|
|
|
|
# Archive: a binary envelope (base64, content type, size).
|
|
archive = await mcp.ok("gitea_repo_get_archive", archive="main.zip", **repo)
|
|
blob = base64.b64decode(archive.data["content_base64"])
|
|
assert blob.startswith(b"PK")
|
|
assert archive.data["size"] == len(blob)
|
|
assert archive.data["content_type"].startswith("application/")
|
|
|
|
# Search finds the repository by name.
|
|
deadline = time.monotonic() + 15
|
|
found: list[str] = []
|
|
while time.monotonic() < deadline and fresh_repo.name not in found:
|
|
search = await mcp.ok("gitea_repo_search", q=fresh_repo.name)
|
|
found = names(search.data["data"])
|
|
if fresh_repo.name not in found:
|
|
time.sleep(0.5)
|
|
assert fresh_repo.name in found
|
|
|
|
# Fork as another user (admin sudo), then list forks.
|
|
fork = await mcp.ok("gitea_repo_create_fork", status=(201, 202), sudo=me, body={}, **repo)
|
|
assert fork.identity["sudo"] == me
|
|
assert fork.data["owner"]["login"] == me
|
|
forks = await mcp.ok("gitea_repo_list_forks", **repo)
|
|
assert me in [item["owner"]["login"] for item in forks.data]
|
|
await mcp.ok("gitea_repo_delete", owner=me, repo=fresh_repo.name)
|
|
|
|
# Transfer to an organization created for the purpose.
|
|
org = gitea.unique("it-xfer")
|
|
try:
|
|
await mcp.ok("gitea_org_create", status=201, body={"username": org})
|
|
await mcp.ok("gitea_repo_transfer", status=(201, 202), body={"new_owner": org}, **repo)
|
|
moved = await mcp.ok("gitea_repo_get", owner=org, repo=fresh_repo.name)
|
|
assert moved.data["owner"]["login"] == org
|
|
# Gitea redirects the old path to the new owner; redirects are never followed.
|
|
redirected = await mcp.fail("gitea_repo_get", 301, **repo)
|
|
assert redirected.error_kind == "redirect"
|
|
await mcp.ok("gitea_repo_edit", owner=org, repo=fresh_repo.name, body={"archived": True})
|
|
assert (await mcp.ok("gitea_repo_get", owner=org, repo=fresh_repo.name)).data["archived"]
|
|
finally:
|
|
drop_org(gitea, org)
|