From 29b12bfe0df85afb54d99c0757e58cfc0718f57b Mon Sep 17 00:00:00 2001 From: latte Date: Wed, 14 Oct 2026 17:00:53 +0200 Subject: [PATCH] test: admin, sudo, notification, user and package scenarios --- tests/integration/test_scenario_admin_user.py | 270 ++++++++++++++++++ 1 file changed, 270 insertions(+) create mode 100644 tests/integration/test_scenario_admin_user.py diff --git a/tests/integration/test_scenario_admin_user.py b/tests/integration/test_scenario_admin_user.py new file mode 100644 index 0000000..2b4d9b1 --- /dev/null +++ b/tests/integration/test_scenario_admin_user.py @@ -0,0 +1,270 @@ +"""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)