"""Scenarios: organizations, teams and repositories, driven through the MCP server.""" from __future__ import annotations import asyncio import base64 import time from typing import Any from tests.integration.conftest import ( McpSession, RegularUser, RepoRef, b64, drop_org, drop_user, ) 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: await asyncio.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) async def test_repository_transfer_accept_and_reject( mcp: McpSession, gitea: GiteaInstance, regular_user: RegularUser ) -> None: """A transfer started by a non-admin stays pending until the receiving user answers it.""" owner = regular_user.login recipient, _ = gitea.create_user(gitea.unique("it-recv")) accepted = gitea.unique("it-xfer-ok") rejected = gitea.unique("it-xfer-no") gitea.create_repo(owner, accepted) gitea.create_repo(owner, rejected) try: for name in (accepted, rejected): # Started by the owner (through sudo): not an admin, so Gitea keeps the # transfer pending (answering 201 although the swagger lists 202). await mcp.ok( "gitea_repo_transfer", status=(201, 202), sudo=owner, owner=owner, repo=name, body={"new_owner": recipient}, ) # Still the owner's repository until the recipient answers. assert (await mcp.ok("gitea_repo_get", owner=owner, repo=name)).data["owner"][ "login" ] == owner took = await mcp.ok( "gitea_repo_accept_repo_transfer", status=202, sudo=recipient, owner=owner, repo=accepted, ) assert took.identity["sudo"] == recipient assert (await mcp.ok("gitea_repo_get", owner=recipient, repo=accepted)).data["owner"][ "login" ] == recipient await mcp.ok("gitea_repo_reject_repo_transfer", sudo=recipient, owner=owner, repo=rejected) assert (await mcp.ok("gitea_repo_get", owner=owner, repo=rejected)).data["owner"][ "login" ] == owner # The transfer is gone: answering it again finds nothing. await mcp.fail( "gitea_repo_accept_repo_transfer", 404, sudo=recipient, owner=owner, repo=rejected ) finally: gitea.api("DELETE", f"/repos/{owner}/{rejected}") gitea.api("DELETE", f"/repos/{owner}/{accepted}") gitea.api("DELETE", f"/repos/{recipient}/{accepted}") drop_user(gitea, recipient)