"""Scenarios: site administration, sudo, notifications, current-user features and packages.""" from __future__ import annotations import asyncio import secrets import time from typing import Any from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from tests.integration.conftest import McpSession, RegularUser, RepoRef, drop_org, drop_user from tests.integration.harness import GiteaInstance def ssh_public_key() -> str: """A fresh ed25519 public key in OpenSSH format (throwaway, the private half is dropped).""" key = Ed25519PrivateKey.generate().public_key() raw = key.public_bytes(serialization.Encoding.OpenSSH, serialization.PublicFormat.OpenSSH) return raw.decode("ascii") def logins(items: Any, key: str = "login") -> list[str]: """The ``key`` member of every dict in a list result.""" return [item[key] for item in items] # Cron tasks that are safe to trigger on a throwaway instance. SAFE_CRON_TASKS = ("cleanup_hook_task_table", "cleanup_expired_actions_assets", "check_repo_stats") async def test_admin_manages_users_orgs_keys_and_cron( mcp: McpSession, gitea: GiteaInstance, fresh_repo: RepoRef ) -> None: # fresh_repo makes sure the repository root exists (the unadopted scan needs it). login = gitea.unique("it-adm") org = gitea.unique("it-adm-org") email = f"{login}@example.test" try: created = await mcp.ok( "gitea_admin_create_user", status=201, body={ "username": login, "email": email, "password": secrets.token_urlsafe(18), "must_change_password": False, }, ) assert created.data["login"] == login edited = await mcp.ok( "gitea_admin_edit_user", username=login, body={"source_id": 0, "login_name": login, "full_name": "Integration User"}, ) assert edited.data["full_name"] == "Integration User" found = await mcp.ok("gitea_admin_search_users", q=login) assert login in logins(found.data) assert email in logins( (await mcp.ok("gitea_admin_get_all_emails")).data, "email" ) or email in logins((await mcp.ok("gitea_admin_search_emails", q=login)).data, "email") # SSH key for the user, listed as the user, then removed. key = await mcp.ok( "gitea_admin_create_public_key", status=201, username=login, body={"title": "it key", "key": ssh_public_key()}, ) listed = await mcp.ok("gitea_user_list_keys", username=login) assert key.data["id"] in [item["id"] for item in listed.data] await mcp.ok("gitea_admin_delete_user_public_key", username=login, id=key.data["id"]) assert (await mcp.ok("gitea_user_list_keys", username=login)).data == [] # An organization created on behalf of the user. await mcp.ok("gitea_admin_create_org", status=201, username=login, body={"username": org}) assert org in logins((await mcp.ok("gitea_admin_get_all_orgs")).data, "name") assert org in logins((await mcp.ok("gitea_user_list_orgs", username=login)).data, "name") await mcp.ok("gitea_org_delete", org=org) # Cron: list the tasks and run one safe task. tasks = await mcp.ok("gitea_admin_cron_list") task_names = logins(tasks.data, "name") assert task_names chosen = next((name for name in SAFE_CRON_TASKS if name in task_names), task_names[0]) await mcp.ok("gitea_admin_cron_run", task=chosen) # Unadopted repositories and runner registration token. await mcp.ok("gitea_admin_unadopted_list") token = await mcp.ok("gitea_admin_create_runner_registration_token", status=(200, 201)) assert isinstance(token.data["token"], str) and token.data["token"] assert await mcp.ok("gitea_admin_get_runners") await mcp.ok("gitea_admin_delete_user", username=login, purge=True) await mcp.fail("gitea_user_get", 404, username=login) finally: drop_org(gitea, org) drop_user(gitea, login) async def test_sudo_acts_as_the_named_user( mcp: McpSession, gitea: GiteaInstance, regular_user: RegularUser ) -> None: me = regular_user.login plain = await mcp.ok("gitea_user_get_current") assert plain.data["login"] == gitea.admin_login assert plain.identity == {"login": gitea.admin_login, "mode": "admin-token", "sudo": None} sudoed = await mcp.ok("gitea_user_get_current", sudo=me) assert sudoed.data["login"] == me assert sudoed.identity["sudo"] == me assert sudoed.identity["login"] == gitea.admin_login repo = gitea.unique("sudo-repo") try: made = await mcp.ok( "gitea_user_current_create_repo", status=201, sudo=me, body={"name": repo} ) assert made.data["owner"]["login"] == me assert gitea.api("GET", f"/repos/{me}/{repo}").status_code == 200 finally: gitea.api("DELETE", f"/repos/{me}/{repo}") # The audit log records who acted and who they acted as. entries = [e for e in mcp.audit_entries() if e.get("sudo") == me] assert {e["tool_name"] for e in entries} == { "gitea_user_get_current", "gitea_user_current_create_repo", } assert all(e["identity_login"] == gitea.admin_login for e in entries) assert all(e["identity_mode"] == "admin-token" for e in entries) async def test_notifications_list_and_mark_read( mcp: McpSession, gitea: GiteaInstance, regular_user: RegularUser, fresh_repo: RepoRef ) -> None: repo = fresh_repo.args # A second user assigns the admin to a new issue, which notifies the admin. await mcp.ok( "gitea_repo_add_collaborator", collaborator=regular_user.login, body={"permission": "write"}, **repo, ) await mcp.ok( "gitea_issue_create_issue", status=201, sudo=regular_user.login, body={"title": "Ping", "assignees": [gitea.admin_login]}, **repo, ) deadline = time.monotonic() + 30 notices: list[dict[str, Any]] = [] while time.monotonic() < deadline and not notices: notices = (await mcp.ok("gitea_notification_get_list", **{"status-types": ["unread"]})).data if not notices: await asyncio.sleep(0.5) assert notices, "no notification arrived within 30 seconds" assert (await mcp.ok("gitea_notification_new_available")).data["new"] >= 1 repo_list = await mcp.ok("gitea_notification_get_repo_list", **repo) assert repo_list.data thread_id = str(notices[0]["id"]) thread = await mcp.ok("gitea_notification_get_thread", id=thread_id) assert thread.data["unread"] is True await mcp.ok("gitea_notification_read_thread", status=(200, 205), id=thread_id) assert (await mcp.ok("gitea_notification_get_thread", id=thread_id)).data["unread"] is False await mcp.ok("gitea_notification_read_list", status=(200, 205)) assert (await mcp.ok("gitea_notification_get_list", **{"status-types": ["unread"]})).data == [] async def test_current_user_features( mcp: McpSession, gitea: GiteaInstance, regular_user: RegularUser, fresh_repo: RepoRef ) -> None: repo = fresh_repo.args other = regular_user.login me = await mcp.ok("gitea_user_get_current") assert me.data["is_admin"] is True assert (await mcp.ok("gitea_user_get", username=other)).data["login"] == other assert (await mcp.ok("gitea_user_get_settings")).data is not None # Email addresses. address = f"extra-{secrets.token_hex(3)}@example.test" await mcp.ok("gitea_user_add_email", status=201, body={"emails": [address]}) assert address in logins((await mcp.ok("gitea_user_list_emails")).data, "email") await mcp.ok("gitea_user_delete_email", body={"emails": [address]}) assert address not in logins((await mcp.ok("gitea_user_list_emails")).data, "email") # SSH keys and GPG keys. key = await mcp.ok( "gitea_user_current_post_key", status=201, body={"title": "it key", "key": ssh_public_key()}, ) key_id = key.data["id"] assert key_id in [k["id"] for k in (await mcp.ok("gitea_user_current_list_keys")).data] assert (await mcp.ok("gitea_user_current_get_key", id=key_id)).data["title"] == "it key" await mcp.ok("gitea_user_current_delete_key", id=key_id) assert (await mcp.ok("gitea_user_current_list_gpg_keys")).data == [] assert (await mcp.ok("gitea_user_list_gpg_keys", username=other)).data == [] # Follow and unfollow. await mcp.ok("gitea_user_current_put_follow", username=other) await mcp.ok("gitea_user_current_check_following", username=other) await mcp.ok("gitea_user_check_following", username=gitea.admin_login, target=other) assert other in logins((await mcp.ok("gitea_user_current_list_following")).data) assert gitea.admin_login in logins( (await mcp.ok("gitea_user_list_followers", username=other)).data ) await mcp.ok("gitea_user_current_delete_follow", username=other) await mcp.fail("gitea_user_current_check_following", 404, username=other) # Block and unblock. await mcp.ok("gitea_user_block_user", username=other) await mcp.ok("gitea_user_check_user_block", username=other) assert other in logins((await mcp.ok("gitea_user_list_blocks")).data) await mcp.ok("gitea_user_unblock_user", username=other) await mcp.fail("gitea_user_check_user_block", 404, username=other) # Starred and watched repositories. await mcp.ok("gitea_user_current_put_star", **repo) await mcp.ok("gitea_user_current_check_starring", **repo) assert fresh_repo.name in logins((await mcp.ok("gitea_user_current_list_starred")).data, "name") await mcp.ok("gitea_user_current_delete_star", **repo) await mcp.ok("gitea_repo_put_subscription", **repo) watched = await mcp.ok("gitea_user_current_list_subscriptions") assert fresh_repo.name in logins(watched.data, "name") await mcp.ok("gitea_repo_delete_subscription", **repo) # Search. assert other in logins((await mcp.ok("gitea_user_search", q=other)).data["data"]) async def test_generic_package_roundtrip(mcp: McpSession, gitea: GiteaInstance) -> None: owner = gitea.admin_login name = gitea.unique("it-pkg") content = b"package bytes " + secrets.token_bytes(16) upload = gitea.request( "PUT", f"/api/packages/{owner}/generic/{name}/1.0.0/payload.bin", content=content ) assert upload.status_code == 201, upload.text deadline = time.monotonic() + 15 listed: list[Any] = [] while time.monotonic() < deadline and not listed: listed = ( await mcp.ok("gitea_package_list_packages", owner=owner, type="generic", q=name) ).data if not listed: await asyncio.sleep(0.5) assert [item["name"] for item in listed] == [name] assert listed[0]["version"] == "1.0.0" pkg = {"owner": owner, "type": "generic", "name": name} got = await mcp.ok("gitea_package_get_package", version="1.0.0", **pkg) assert got.data["name"] == name files = await mcp.ok("gitea_package_list_package_files", version="1.0.0", **pkg) assert [item["name"] for item in files.data] == ["payload.bin"] assert files.data[0]["size"] == len(content) latest = await mcp.ok("gitea_package_get_latest_package_version", **pkg) assert latest.data["version"] == "1.0.0" assert [ v["version"] for v in (await mcp.ok("gitea_package_list_package_versions", **pkg)).data ] == ["1.0.0"] await mcp.ok("gitea_package_delete_package_version", version="1.0.0", **pkg) await mcp.fail("gitea_package_get_package", 404, version="1.0.0", **pkg)