test: issue, pull request, release, secret, variable and webhook scenarios
This commit is contained in:
Latte committed
1 parent
51c766b231
commit
825c3c93e2
2 files changed
+489
No files matched your search
@@ -0,0 +1,265 @@
|
||||
"""Scenarios: releases and attachments, Action secrets and variables, webhooks."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import secrets
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from tests.integration.conftest import McpSession, RepoRef, b64, drop_org
|
||||
from tests.integration.harness import GiteaInstance
|
||||
|
||||
HOOK_CONFIG = {"url": "http://127.0.0.1:9/hook", "content_type": "json"}
|
||||
|
||||
|
||||
async def test_release_with_attachment(
|
||||
mcp: McpSession, gitea: GiteaInstance, fresh_repo: RepoRef
|
||||
) -> None:
|
||||
repo = fresh_repo.args
|
||||
release = await mcp.ok(
|
||||
"gitea_release_create_release",
|
||||
status=201,
|
||||
body={
|
||||
"tag_name": "v1.0.0",
|
||||
"target_commitish": "main",
|
||||
"name": "One point oh",
|
||||
"body": "first release",
|
||||
},
|
||||
**repo,
|
||||
)
|
||||
release_id = release.data["id"]
|
||||
assert release.data["tag_name"] == "v1.0.0"
|
||||
assert (await mcp.ok("gitea_release_get_latest_release", **repo)).data["id"] == release_id
|
||||
assert (await mcp.ok("gitea_release_get_release_by_tag", tag="v1.0.0", **repo)).data["id"] == (
|
||||
release_id
|
||||
)
|
||||
|
||||
# Upload an attachment through the file argument (base64), then list and download it.
|
||||
payload = b"release asset " + secrets.token_bytes(24)
|
||||
asset = await mcp.ok(
|
||||
"gitea_release_create_release_attachment",
|
||||
status=201,
|
||||
id=release_id,
|
||||
name="notes.bin",
|
||||
attachment={
|
||||
"filename": "notes.bin",
|
||||
"content_base64": b64(payload),
|
||||
"content_type": "application/octet-stream",
|
||||
},
|
||||
**repo,
|
||||
)
|
||||
attachment_id = asset.data["id"]
|
||||
assert asset.data["name"] == "notes.bin"
|
||||
assert asset.data["size"] == len(payload)
|
||||
listed = await mcp.ok("gitea_release_list_release_attachments", id=release_id, **repo)
|
||||
assert [item["name"] for item in listed.data] == ["notes.bin"]
|
||||
meta = await mcp.ok(
|
||||
"gitea_release_get_release_attachment", id=release_id, attachment_id=attachment_id, **repo
|
||||
)
|
||||
download = gitea.request("GET", meta.data["browser_download_url"].removeprefix(gitea.base_url))
|
||||
assert download.status_code == 200
|
||||
assert download.content == payload
|
||||
|
||||
renamed = await mcp.ok(
|
||||
"gitea_release_edit_release_attachment",
|
||||
status=(200, 201),
|
||||
id=release_id,
|
||||
attachment_id=attachment_id,
|
||||
body={"name": "renamed.bin"},
|
||||
**repo,
|
||||
)
|
||||
assert renamed.data["name"] == "renamed.bin"
|
||||
edited = await mcp.ok(
|
||||
"gitea_release_edit_release",
|
||||
id=release_id,
|
||||
body={"name": "Renamed", "draft": False},
|
||||
**repo,
|
||||
)
|
||||
assert edited.data["name"] == "Renamed"
|
||||
assert [r["id"] for r in (await mcp.ok("gitea_release_list_releases", **repo)).data] == [
|
||||
release_id
|
||||
]
|
||||
|
||||
await mcp.ok(
|
||||
"gitea_release_delete_release_attachment",
|
||||
id=release_id,
|
||||
attachment_id=attachment_id,
|
||||
**repo,
|
||||
)
|
||||
assert (
|
||||
await mcp.ok("gitea_release_list_release_attachments", id=release_id, **repo)
|
||||
).data == []
|
||||
await mcp.ok("gitea_release_delete_release", id=release_id, **repo)
|
||||
await mcp.fail("gitea_release_get_release", 404, id=release_id, **repo)
|
||||
|
||||
# A second release is removed through its tag.
|
||||
await mcp.ok(
|
||||
"gitea_release_create_release",
|
||||
status=201,
|
||||
body={"tag_name": "v1.1.0", "target_commitish": "main"},
|
||||
**repo,
|
||||
)
|
||||
await mcp.ok("gitea_release_delete_release_by_tag", tag="v1.1.0", **repo)
|
||||
|
||||
|
||||
def _names(result: Any) -> list[str]:
|
||||
return [item["name"] for item in result.data]
|
||||
|
||||
|
||||
async def test_secrets_and_variables_never_leak_values(
|
||||
mcp: McpSession, gitea: GiteaInstance, fresh_repo: RepoRef
|
||||
) -> None:
|
||||
repo = fresh_repo.args
|
||||
org = gitea.unique("it-secrets")
|
||||
values = [f"S3cr3t-{secrets.token_hex(12)}" for _ in range(4)]
|
||||
repo_secret, org_secret, user_secret, probe = values
|
||||
try:
|
||||
await mcp.ok("gitea_org_create", status=201, body={"username": org})
|
||||
|
||||
# Repository, organization and user-level secrets: set, list (names only), delete.
|
||||
await mcp.ok(
|
||||
"gitea_actions_update_repo_secret",
|
||||
secretname="REPO_KEY",
|
||||
body={"data": repo_secret, "description": "repo key"},
|
||||
**repo,
|
||||
)
|
||||
await mcp.ok(
|
||||
"gitea_actions_update_repo_secret",
|
||||
secretname="REPO_KEY",
|
||||
body={"data": repo_secret + "x"},
|
||||
**repo,
|
||||
)
|
||||
listed = await mcp.ok("gitea_actions_repo_list_secrets", **repo)
|
||||
assert _names(listed) == ["REPO_KEY"]
|
||||
assert set(listed.data[0]) <= {"name", "description", "created_at"}
|
||||
await mcp.ok(
|
||||
"gitea_actions_update_org_secret",
|
||||
org=org,
|
||||
secretname="ORG_KEY",
|
||||
body={"data": org_secret},
|
||||
)
|
||||
org_listed = await mcp.ok("gitea_actions_org_list_secrets", org=org)
|
||||
assert _names(org_listed) == ["ORG_KEY"]
|
||||
assert set(org_listed.data[0]) <= {"name", "description", "created_at"}
|
||||
await mcp.ok(
|
||||
"gitea_actions_update_user_secret", secretname="USER_KEY", body={"data": user_secret}
|
||||
)
|
||||
|
||||
# A failing call carrying a secret must not echo it either.
|
||||
denied = await mcp.fail(
|
||||
"gitea_actions_update_repo_secret",
|
||||
404,
|
||||
owner=gitea.admin_login,
|
||||
repo="does-not-exist-" + secrets.token_hex(3),
|
||||
secretname="NOPE",
|
||||
body={"data": probe},
|
||||
)
|
||||
assert probe not in denied.text
|
||||
|
||||
await mcp.ok("gitea_actions_delete_repo_secret", secretname="REPO_KEY", **repo)
|
||||
await mcp.ok("gitea_actions_delete_org_secret", org=org, secretname="ORG_KEY")
|
||||
await mcp.ok("gitea_actions_delete_user_secret", secretname="USER_KEY")
|
||||
assert (await mcp.ok("gitea_actions_repo_list_secrets", **repo)).data == []
|
||||
|
||||
# Variables are not secret: create, get, update, list, delete at every level.
|
||||
scopes: list[tuple[str, dict[str, str]]] = [
|
||||
("repo", repo),
|
||||
("org", {"org": org}),
|
||||
("user", {}),
|
||||
]
|
||||
for level, scope in scopes:
|
||||
create = f"gitea_actions_create_{level}_variable"
|
||||
await mcp.ok(
|
||||
create,
|
||||
status=(200, 201, 204),
|
||||
variablename="GREETING",
|
||||
body={"value": "one"},
|
||||
**scope,
|
||||
)
|
||||
got = await mcp.ok(
|
||||
f"gitea_actions_get_{level}_variable", variablename="GREETING", **scope
|
||||
)
|
||||
assert got.data["data"] == "one"
|
||||
await mcp.ok(
|
||||
f"gitea_actions_update_{level}_variable",
|
||||
variablename="GREETING",
|
||||
body={"value": "two"},
|
||||
**scope,
|
||||
)
|
||||
changed = await mcp.ok(
|
||||
f"gitea_actions_get_{level}_variable", variablename="GREETING", **scope
|
||||
)
|
||||
assert changed.data["data"] == "two"
|
||||
names = await mcp.ok(f"gitea_actions_get_{level}_variables_list", **scope)
|
||||
assert "GREETING" in _names(names)
|
||||
await mcp.ok(f"gitea_actions_delete_{level}_variable", variablename="GREETING", **scope)
|
||||
await mcp.fail(
|
||||
f"gitea_actions_get_{level}_variable", 404, variablename="GREETING", **scope
|
||||
)
|
||||
finally:
|
||||
drop_org(gitea, org)
|
||||
gitea.api("DELETE", "/user/actions/secrets/USER_KEY")
|
||||
|
||||
# No secret value in any tool result, and none in the audit log.
|
||||
for value in values:
|
||||
assert all(value not in text for text in mcp.texts)
|
||||
assert value not in mcp.audit_text()
|
||||
audited = [entry["tool_name"] for entry in mcp.audit_entries()]
|
||||
assert "gitea_actions_update_repo_secret" in audited
|
||||
|
||||
|
||||
@pytest.mark.parametrize("level", ["repo", "org", "user", "admin"])
|
||||
async def test_webhook_lifecycle(
|
||||
mcp: McpSession, gitea: GiteaInstance, fresh_repo: RepoRef, level: str
|
||||
) -> None:
|
||||
org = gitea.unique("it-hooks")
|
||||
scope: dict[str, str] = {}
|
||||
prefix = f"gitea_hook_{level}"
|
||||
try:
|
||||
if level == "repo":
|
||||
scope = fresh_repo.args
|
||||
elif level == "org":
|
||||
await mcp.ok("gitea_org_create", status=201, body={"username": org})
|
||||
scope = {"org": org}
|
||||
if level == "admin":
|
||||
create, get, listing, edit, delete = (
|
||||
"gitea_admin_create_hook",
|
||||
"gitea_admin_get_hook",
|
||||
"gitea_admin_list_hooks",
|
||||
"gitea_admin_edit_hook",
|
||||
"gitea_admin_delete_hook",
|
||||
)
|
||||
else:
|
||||
create, get, listing, edit, delete = (
|
||||
f"{prefix}_create_hook",
|
||||
f"{prefix}_get_hook",
|
||||
f"{prefix}_list_hooks",
|
||||
f"{prefix}_edit_hook",
|
||||
f"{prefix}_delete_hook",
|
||||
)
|
||||
hook = await mcp.ok(
|
||||
create,
|
||||
status=201,
|
||||
body={"type": "gitea", "config": HOOK_CONFIG, "events": ["push"], "active": True},
|
||||
**scope,
|
||||
)
|
||||
hook_id = hook.data["id"]
|
||||
assert hook.data["type"] == "gitea"
|
||||
# Hooks created through /admin/hooks count as "default" hooks, so list both kinds.
|
||||
query = {"type": "all"} if level == "admin" else {}
|
||||
hooks = await mcp.ok(listing, **scope, **query)
|
||||
assert hook_id in [item["id"] for item in hooks.data]
|
||||
assert (await mcp.ok(get, id=hook_id, **scope)).data["config"]["url"] == HOOK_CONFIG["url"]
|
||||
edited = await mcp.ok(
|
||||
edit, id=hook_id, body={"active": False, "events": ["push", "issues"]}, **scope
|
||||
)
|
||||
assert edited.data["active"] is False
|
||||
assert "issues" in edited.data["events"]
|
||||
if level == "repo":
|
||||
await mcp.ok("gitea_hook_repo_test_hook", id=hook_id, **scope)
|
||||
await mcp.ok(delete, id=hook_id, **scope)
|
||||
await mcp.fail(get, 404, id=hook_id, **scope)
|
||||
finally:
|
||||
if level == "org":
|
||||
drop_org(gitea, org)
|
||||
Reference in new issue
Block a user