296 lines
12 KiB
Python
296 lines
12 KiB
Python
"""Scenarios: issues (labels, milestones, time tracking, reactions) and pull requests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import time
|
|
|
|
from tests.integration.conftest import McpSession, RegularUser, RepoRef, b64
|
|
from tests.integration.harness import GiteaInstance
|
|
|
|
|
|
async def test_issue_lifecycle(
|
|
mcp: McpSession, regular_user: RegularUser, fresh_repo: RepoRef
|
|
) -> None:
|
|
repo = fresh_repo.args
|
|
watcher = regular_user.login
|
|
|
|
issue = await mcp.ok(
|
|
"gitea_issue_create_issue",
|
|
status=201,
|
|
body={"title": "Broken widget", "body": "It does not widget."},
|
|
**repo,
|
|
)
|
|
index = issue.data["number"]
|
|
assert issue.data["state"] == "open"
|
|
|
|
# Comments.
|
|
comment = await mcp.ok(
|
|
"gitea_issue_create_comment", status=201, index=index, body={"body": "first"}, **repo
|
|
)
|
|
comment_id = comment.data["id"]
|
|
edited = await mcp.ok(
|
|
"gitea_issue_edit_comment", id=comment_id, body={"body": "first, edited"}, **repo
|
|
)
|
|
assert edited.data["body"] == "first, edited"
|
|
comments = await mcp.ok("gitea_issue_get_comments", index=index, **repo)
|
|
assert [item["body"] for item in comments.data] == ["first, edited"]
|
|
|
|
# Labels: create, add, remove.
|
|
label = await mcp.ok(
|
|
"gitea_issue_create_label",
|
|
status=201,
|
|
body={"name": "bug", "color": "#d73a4a", "description": "something is broken"},
|
|
**repo,
|
|
)
|
|
label_id = label.data["id"]
|
|
added = await mcp.ok("gitea_issue_add_label", index=index, body={"labels": [label_id]}, **repo)
|
|
assert [item["name"] for item in added.data] == ["bug"]
|
|
assert (await mcp.ok("gitea_issue_get_labels", index=index, **repo)).data
|
|
await mcp.ok("gitea_issue_remove_label", index=index, id=label_id, **repo)
|
|
assert (await mcp.ok("gitea_issue_get_labels", index=index, **repo)).data == []
|
|
await mcp.ok("gitea_issue_replace_labels", index=index, body={"labels": [label_id]}, **repo)
|
|
await mcp.ok("gitea_issue_clear_labels", index=index, **repo)
|
|
|
|
# Milestone.
|
|
milestone = await mcp.ok(
|
|
"gitea_issue_create_milestone", status=201, body={"title": "v1"}, **repo
|
|
)
|
|
await mcp.ok(
|
|
"gitea_issue_edit_issue", index=index, body={"milestone": milestone.data["id"]}, **repo
|
|
)
|
|
assert (await mcp.ok("gitea_issue_get_issue", index=index, **repo)).data["milestone"][
|
|
"title"
|
|
] == "v1"
|
|
|
|
# Pin and unpin.
|
|
await mcp.ok("gitea_issue_pin_issue", index=index, **repo)
|
|
pinned = await mcp.ok("gitea_issue_list_pinned_issues", **repo)
|
|
assert [item["number"] for item in pinned.data] == [index]
|
|
await mcp.ok("gitea_issue_unpin_issue", index=index, **repo)
|
|
assert (await mcp.ok("gitea_issue_list_pinned_issues", **repo)).data == []
|
|
|
|
# Subscriptions.
|
|
await mcp.ok(
|
|
"gitea_issue_add_subscription", status=(200, 201), index=index, user=watcher, **repo
|
|
)
|
|
subscribers = await mcp.ok("gitea_issue_subscriptions", index=index, **repo)
|
|
assert watcher in [item["login"] for item in subscribers.data]
|
|
await mcp.ok("gitea_issue_delete_subscription", index=index, user=watcher, **repo)
|
|
assert (await mcp.ok("gitea_issue_subscriptions", index=index, **repo)).data == []
|
|
|
|
# Time tracking and the stopwatch.
|
|
await mcp.ok("gitea_issue_add_time", index=index, body={"time": 3600}, **repo)
|
|
times = await mcp.ok("gitea_issue_tracked_times", index=index, **repo)
|
|
assert [item["time"] for item in times.data] == [3600]
|
|
await mcp.ok("gitea_issue_delete_time", index=index, id=times.data[0]["id"], **repo)
|
|
assert (await mcp.ok("gitea_issue_tracked_times", index=index, **repo)).data == []
|
|
await mcp.ok("gitea_issue_start_stopwatch", status=201, index=index, **repo)
|
|
running = await mcp.ok("gitea_user_get_stopwatches")
|
|
assert index in [item["issue_index"] for item in running.data]
|
|
await mcp.ok("gitea_issue_stop_stopwatch", status=201, index=index, **repo)
|
|
assert (await mcp.ok("gitea_issue_tracked_times", index=index, **repo)).data
|
|
|
|
# Reactions.
|
|
await mcp.ok(
|
|
"gitea_issue_post_issue_reaction",
|
|
status=(200, 201),
|
|
index=index,
|
|
body={"content": "+1"},
|
|
**repo,
|
|
)
|
|
reactions = await mcp.ok("gitea_issue_get_issue_reactions", index=index, **repo)
|
|
assert [item["content"] for item in reactions.data] == ["+1"]
|
|
await mcp.ok("gitea_issue_delete_issue_reaction", index=index, body={"content": "+1"}, **repo)
|
|
|
|
# Search finds the issue (indexing is asynchronous, so poll briefly).
|
|
deadline = time.monotonic() + 20
|
|
hits: list[int] = []
|
|
while time.monotonic() < deadline and index not in hits:
|
|
found = await mcp.ok("gitea_issue_list_issues", q="widget", **repo)
|
|
hits = [item["number"] for item in found.data]
|
|
if index not in hits:
|
|
await asyncio.sleep(0.5)
|
|
assert index in hits
|
|
assert (await mcp.ok("gitea_issue_search_issues", q="Broken widget")).data is not None
|
|
|
|
# Close.
|
|
closed = await mcp.ok("gitea_issue_edit_issue", index=index, body={"state": "closed"}, **repo)
|
|
assert closed.data["state"] == "closed"
|
|
assert (await mcp.ok("gitea_issue_get_issue", index=index, **repo)).data["state"] == "closed"
|
|
await mcp.ok("gitea_issue_delete_comment", id=comment_id, **repo)
|
|
await mcp.ok("gitea_issue_delete_label", id=label_id, **repo)
|
|
await mcp.ok("gitea_issue_delete_milestone", id=str(milestone.data["id"]), **repo)
|
|
|
|
|
|
async def test_pull_request_review_and_merge(
|
|
mcp: McpSession, gitea: GiteaInstance, regular_user: RegularUser, fresh_repo: RepoRef
|
|
) -> None:
|
|
repo = fresh_repo.args
|
|
reviewer = regular_user.login
|
|
await mcp.ok(
|
|
"gitea_repo_add_collaborator",
|
|
collaborator=reviewer,
|
|
body={"permission": "write"},
|
|
**repo,
|
|
)
|
|
try:
|
|
# A branch with a change, pushed through the contents API.
|
|
await mcp.ok(
|
|
"gitea_repo_create_branch",
|
|
status=201,
|
|
body={"new_branch_name": "feat/greeting", "old_branch_name": "main"},
|
|
**repo,
|
|
)
|
|
await mcp.ok(
|
|
"gitea_repo_create_file",
|
|
status=201,
|
|
filepath="greeting.txt",
|
|
body={
|
|
"content": b64("hello pull request\n"),
|
|
"message": "add greeting",
|
|
"branch": "feat/greeting",
|
|
},
|
|
**repo,
|
|
)
|
|
pull = await mcp.ok(
|
|
"gitea_pull_create_pull_request",
|
|
status=201,
|
|
body={"title": "Add greeting", "base": "main", "head": "feat/greeting"},
|
|
**repo,
|
|
)
|
|
index = pull.data["number"]
|
|
assert pull.data["state"] == "open"
|
|
assert not pull.data["merged"]
|
|
|
|
files = await mcp.ok("gitea_pull_get_pull_request_files", index=index, **repo)
|
|
assert [item["filename"] for item in files.data] == ["greeting.txt"]
|
|
commits = await mcp.ok("gitea_pull_get_pull_request_commits", index=index, **repo)
|
|
assert len(commits.data) == 1
|
|
|
|
diff = await mcp.ok(
|
|
"gitea_pull_download_pull_diff_or_patch", index=index, diffType="diff", **repo
|
|
)
|
|
assert isinstance(diff.data, str)
|
|
assert "+hello pull request" in diff.data
|
|
|
|
# Requested reviewer, then a pending review with an inline comment (as the
|
|
# reviewer, through sudo), submitted as an approval.
|
|
await mcp.ok(
|
|
"gitea_pull_create_pull_review_requests",
|
|
status=201,
|
|
index=index,
|
|
body={"reviewers": [reviewer]},
|
|
**repo,
|
|
)
|
|
pending = await mcp.ok(
|
|
"gitea_pull_create_pull_review",
|
|
sudo=reviewer,
|
|
index=index,
|
|
body={
|
|
"body": "looks fine",
|
|
"comments": [{"path": "greeting.txt", "new_position": 1, "body": "nice"}],
|
|
},
|
|
**repo,
|
|
)
|
|
assert pending.identity["sudo"] == reviewer
|
|
review_id = pending.data["id"]
|
|
assert pending.data["state"] == "PENDING"
|
|
submitted = await mcp.ok(
|
|
"gitea_pull_submit_pull_review",
|
|
sudo=reviewer,
|
|
index=index,
|
|
id=review_id,
|
|
body={"event": "APPROVED", "body": "approved"},
|
|
**repo,
|
|
)
|
|
assert submitted.data["state"] == "APPROVED"
|
|
reviews = await mcp.ok("gitea_pull_list_pull_reviews", index=index, **repo)
|
|
assert "APPROVED" in [item["state"] for item in reviews.data]
|
|
comments = await mcp.ok(
|
|
"gitea_pull_get_pull_review_comments", index=index, id=review_id, **repo
|
|
)
|
|
assert [item["body"] for item in comments.data] == ["nice"]
|
|
|
|
# Not merged yet, then merged.
|
|
await mcp.fail("gitea_pull_pull_request_is_merged", 404, index=index, **repo)
|
|
await mcp.ok("gitea_pull_merge_pull_request", index=index, body={"do": "merge"}, **repo)
|
|
await mcp.ok("gitea_pull_pull_request_is_merged", index=index, **repo)
|
|
merged = await mcp.ok("gitea_pull_get_pull_request", index=index, **repo)
|
|
assert merged.data["merged"] is True
|
|
assert merged.data["state"] == "closed"
|
|
readme = await mcp.ok("gitea_repo_get_contents", filepath="greeting.txt", **repo)
|
|
assert readme.data["path"] == "greeting.txt"
|
|
|
|
# The merge commit resolves back to the pull request.
|
|
merge_sha = merged.data["merge_commit_sha"]
|
|
by_commit = await mcp.ok("gitea_repo_get_commit_pull_request", sha=merge_sha, **repo)
|
|
assert by_commit.data["number"] == index
|
|
|
|
# Merging it again is refused with the documented status, not silently accepted.
|
|
again = await mcp.fail(
|
|
"gitea_pull_merge_pull_request", 405, index=index, body={"do": "merge"}, **repo
|
|
)
|
|
assert again.error_kind == "method_not_allowed"
|
|
assert "already merged" in again.payload["error"]["message"]
|
|
finally:
|
|
gitea.api("DELETE", f"/repos/{fresh_repo.owner}/{fresh_repo.name}/collaborators/{reviewer}")
|
|
|
|
|
|
async def test_scheduled_auto_merge_can_be_cancelled(
|
|
mcp: McpSession, gitea: GiteaInstance, fresh_repo: RepoRef
|
|
) -> None:
|
|
"""A required status check that is still pending turns the merge into a scheduled one."""
|
|
repo = fresh_repo.args
|
|
await mcp.ok(
|
|
"gitea_repo_create_branch",
|
|
status=201,
|
|
body={"new_branch_name": "feat/auto", "old_branch_name": "main"},
|
|
**repo,
|
|
)
|
|
await mcp.ok(
|
|
"gitea_repo_create_file",
|
|
status=201,
|
|
filepath="auto.txt",
|
|
body={"content": b64("auto\n"), "message": "add auto", "branch": "feat/auto"},
|
|
**repo,
|
|
)
|
|
await mcp.ok(
|
|
"gitea_repo_create_branch_protection",
|
|
status=201,
|
|
body={
|
|
"rule_name": "main",
|
|
"enable_status_check": True,
|
|
"status_check_contexts": ["ci/required"],
|
|
},
|
|
**repo,
|
|
)
|
|
pull = await mcp.ok(
|
|
"gitea_pull_create_pull_request",
|
|
status=201,
|
|
body={"title": "Auto merge", "base": "main", "head": "feat/auto"},
|
|
**repo,
|
|
)
|
|
index = pull.data["number"]
|
|
await mcp.ok(
|
|
"gitea_repo_create_status",
|
|
status=201,
|
|
sha=pull.data["head"]["sha"],
|
|
body={"state": "pending", "context": "ci/required"},
|
|
**repo,
|
|
)
|
|
scheduled = await mcp.ok(
|
|
"gitea_pull_merge_pull_request",
|
|
status=201,
|
|
index=index,
|
|
body={"do": "merge", "merge_when_checks_succeed": True},
|
|
**repo,
|
|
)
|
|
assert scheduled.status == 201 # scheduled, not merged
|
|
assert (await mcp.ok("gitea_pull_get_pull_request", index=index, **repo)).data[
|
|
"merged"
|
|
] is False
|
|
await mcp.ok("gitea_pull_cancel_scheduled_auto_merge", index=index, **repo)
|
|
# Nothing is scheduled any more.
|
|
await mcp.fail("gitea_pull_cancel_scheduled_auto_merge", 404, index=index, **repo)
|