sysadminandClaude Opus 4.8
ed0e8c82de
status:pr-open was applied by gitea_create_pr and never removed again. Every
terminal path finished without touching it, so a repository audit found 40
closed issues still advertising an open PR.
Add terminal_pr_label_cleanup.py as the single authoritative rule and route
every sanctioned terminal path through it, so the paths cannot drift:
- merge (gitea_merge_pr)
- close without merge (gitea_edit_pr)
- supersession/abandonment (gitea_reconcile_superseded_by_merged_pr)
- already-landed reconciliation (gitea_reconcile_already_landed_pr)
- controller closure (gitea_close_issue)
- retry/recovery (new gitea_cleanup_terminal_pr_labels)
The rule removes only status:pr-open, preserves every other label, allows an
empty resulting set, is a no-op when the label is absent (so retries are
safe), and confirms the outcome by read-after-write rather than assumption.
Controller closure runs the cleanup before the state change and fails closed
if it cannot be completed and verified; closing first would bake in the stale
label with no later step to catch it. Post-merge cleanup never blocks the
merge, which already happened, and reports failures with a safe next action.
Also:
- gitea_assess_terminal_label_hygiene: read-only terminal validation that
reports residual status:pr-open, exempting issues with a genuinely open PR.
- _put_issue_label_names now accepts Gitea's empty response body when the
requested set is empty, so clearing the last label works.
- test_audit's close_issue fixture keys on the request instead of call order,
since closing now also reads labels for the cleanup and its read-back.
Docs: label-taxonomy terminal-transition section, runbook pointer, and the
review-merge / reconcile-landed final-report terminal-label requirements.
Suite: 4045 passed, 11 failed, 6 skipped. The same 11 failures reproduce on
clean master df31674 (4010 passed, 11 failed) and are pre-existing.
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
451 lines
20 KiB
Python
451 lines
20 KiB
Python
import sys as _sys
|
|
from pathlib import Path as _Path
|
|
_sys.path.insert(0, str(_Path(__file__).resolve().parent))
|
|
from mutation_profile_fixture import shared_mutation_env # noqa: E402
|
|
"""Tests for Gitea MCP mutating-action audit logging (issue #18).
|
|
|
|
Covers the pure audit module (redaction, event building, sink writes) and the
|
|
wiring in mcp_server: mutating tools emit one record with profile +
|
|
authenticated username + outcome, secrets are redacted, and — critically —
|
|
auditing is a no-op (no records, no extra API calls) when GITEA_AUDIT_LOG is
|
|
unset so existing behaviour is unchanged.
|
|
"""
|
|
import os
|
|
import sys
|
|
import json
|
|
import datetime
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
|
|
|
import gitea_audit # noqa: E402
|
|
import mcp_server # noqa: E402
|
|
from mcp_server import ( # noqa: E402
|
|
gitea_create_issue,
|
|
gitea_close_issue,
|
|
gitea_merge_pr,
|
|
gitea_submit_pr_review,
|
|
)
|
|
|
|
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
|
|
FIXED_TS = datetime.datetime(2026, 7, 1, 12, 0, 0, tzinfo=datetime.timezone.utc)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pure audit module
|
|
# ---------------------------------------------------------------------------
|
|
class TestRedaction(unittest.TestCase):
|
|
|
|
def test_redacts_secret_keys(self):
|
|
red = gitea_audit.redact({"token": "abc", "title": "hi", "Password": "x"})
|
|
self.assertEqual(red["token"], gitea_audit.REDACTED)
|
|
self.assertEqual(red["Password"], gitea_audit.REDACTED)
|
|
self.assertEqual(red["title"], "hi")
|
|
|
|
def test_redacts_nested_and_lists(self):
|
|
red = gitea_audit.redact({"outer": {"authorization": "Basic xyz"},
|
|
"items": [{"secret": "s"}, "plain"]})
|
|
self.assertEqual(red["outer"]["authorization"], gitea_audit.REDACTED)
|
|
self.assertEqual(red["items"][0]["secret"], gitea_audit.REDACTED)
|
|
self.assertEqual(red["items"][1], "plain")
|
|
|
|
def test_redacts_credential_value_prefixes(self):
|
|
self.assertEqual(
|
|
gitea_audit._redact_str("failed: token abc-123 rejected"),
|
|
"failed: token [REDACTED] rejected",
|
|
)
|
|
self.assertIn(gitea_audit.REDACTED, gitea_audit._redact_str("Basic Zm9v"))
|
|
self.assertIn(gitea_audit.REDACTED, gitea_audit._redact_str("Bearer tok99"))
|
|
|
|
def test_non_string_untouched(self):
|
|
self.assertEqual(gitea_audit.redact(42), 42)
|
|
self.assertIsNone(gitea_audit.redact(None))
|
|
|
|
def test_redacts_urls(self):
|
|
# 1. Real service URLs are completely redacted
|
|
self.assertEqual(
|
|
gitea_audit._redact_str("failed for http://gitea.prgs.cc/api/v1/repos/x"),
|
|
"failed for [REDACTED_URL]"
|
|
)
|
|
# 2. Hostname is redacted even without full URL prefix
|
|
self.assertEqual(
|
|
gitea_audit._redact_str("error contacting gitea.prgs.cc directly"),
|
|
"error contacting [REDACTED_HOST] directly"
|
|
)
|
|
# 3. Synthetic test-only URLs are preserved
|
|
self.assertEqual(
|
|
gitea_audit._redact_str("mock URL: https://gitea.example.com/api/v1/repos/x"),
|
|
"mock URL: https://gitea.example.com/api/v1/repos/x"
|
|
)
|
|
# 4. Embedded credentials in synthetic URLs are redacted
|
|
self.assertEqual(
|
|
gitea_audit._redact_str("mock creds: https://user:[email protected]/path"),
|
|
"mock creds: https://[REDACTED_USER]:[REDACTED_PASS]@example.com/path"
|
|
)
|
|
# 5. Query secrets in synthetic URLs are redacted
|
|
self.assertEqual(
|
|
gitea_audit._redact_str("mock query: https://localhost:3003/api?token=abc-123&other=val"),
|
|
"mock query: https://localhost:3003/api?token=%5BREDACTED%5D&other=val"
|
|
)
|
|
|
|
|
|
class TestBuildEvent(unittest.TestCase):
|
|
|
|
def test_core_fields_and_injected_timestamp(self):
|
|
ev = gitea_audit.build_event(
|
|
action="create_issue", result=gitea_audit.SUCCEEDED,
|
|
remote="prgs", server="https://gitea.prgs.cc", repository="Repo",
|
|
issue_number=5, profile_name="gitea-author", audit_label="author-rt",
|
|
authenticated_username="bot", now=FIXED_TS,
|
|
)
|
|
self.assertEqual(ev["timestamp"], "2026-07-01T12:00:00+00:00")
|
|
self.assertEqual(ev["action"], "create_issue")
|
|
self.assertEqual(ev["action_type"], "mutating")
|
|
self.assertEqual(ev["result"], "succeeded")
|
|
self.assertEqual(ev["profile_name"], "gitea-author")
|
|
self.assertEqual(ev["audit_label"], "author-rt")
|
|
self.assertEqual(ev["authenticated_username"], "bot")
|
|
self.assertEqual(ev["issue_number"], 5)
|
|
|
|
def test_metadata_and_reason_redacted(self):
|
|
ev = gitea_audit.build_event(
|
|
action="create_pr", result=gitea_audit.FAILED,
|
|
reason="HTTP 500: token secret-xyz bad",
|
|
request_metadata={"title": "t", "token": "leak"}, now=FIXED_TS,
|
|
)
|
|
self.assertNotIn("secret-xyz", ev["reason"])
|
|
self.assertEqual(ev["request_metadata"]["token"], gitea_audit.REDACTED)
|
|
self.assertEqual(ev["request_metadata"]["title"], "t")
|
|
|
|
|
|
class TestSink(unittest.TestCase):
|
|
|
|
def test_enabled_reflects_env(self):
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
self.assertFalse(gitea_audit.audit_enabled())
|
|
with patch.dict(os.environ, {"GITEA_AUDIT_LOG": "/tmp/x.log"}, clear=True):
|
|
self.assertTrue(gitea_audit.audit_enabled())
|
|
|
|
def test_write_event_appends_json_lines(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
path = os.path.join(d, "audit.log")
|
|
self.assertTrue(gitea_audit.write_event({"a": 1}, path=path))
|
|
self.assertTrue(gitea_audit.write_event({"b": 2}, path=path))
|
|
with open(path, encoding="utf-8") as fh:
|
|
lines = fh.read().splitlines()
|
|
self.assertEqual(len(lines), 2)
|
|
self.assertEqual(json.loads(lines[0])["a"], 1)
|
|
self.assertEqual(json.loads(lines[1])["b"], 2)
|
|
|
|
def test_write_event_noop_without_path(self):
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
self.assertFalse(gitea_audit.write_event({"a": 1}))
|
|
|
|
def test_write_event_never_raises_on_bad_path(self):
|
|
# A path inside a non-existent directory cannot be opened; must not raise.
|
|
self.assertFalse(gitea_audit.write_event({"a": 1}, path="/no/such/dir/x.log"))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# mcp_server wiring
|
|
# ---------------------------------------------------------------------------
|
|
class _AuditWiringBase(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self._dir = tempfile.TemporaryDirectory()
|
|
self.audit_path = os.path.join(self._dir.name, "audit.log")
|
|
# Identity cache is process-global; clear so lookups are deterministic.
|
|
mcp_server._IDENTITY_CACHE.clear()
|
|
|
|
def tearDown(self):
|
|
mcp_server._IDENTITY_CACHE.clear()
|
|
mcp_server.review_workflow_load.clear_review_workflow_load()
|
|
self._dir.cleanup()
|
|
|
|
def _env(self, **extra):
|
|
# Default: prgs-aligned author with create/close (remote=prgs tests).
|
|
# Callers override GITEA_MCP_PROFILE for merger/reviewer paths.
|
|
profile = extra.pop("GITEA_MCP_PROFILE", None) or extra.pop(
|
|
"GITEA_PROFILE_NAME", None
|
|
) or "test-author-prgs"
|
|
env = shared_mutation_env(
|
|
profile,
|
|
GITEA_AUDIT_LOG=self.audit_path,
|
|
GITEA_TOKEN_TEST="test-token",
|
|
)
|
|
# Strip legacy env-only authority knobs if callers still pass them;
|
|
# config-backed profiles own operations and repositories (#714).
|
|
extra.pop("GITEA_ALLOWED_OPERATIONS", None)
|
|
extra.pop("GITEA_FORBIDDEN_OPERATIONS", None)
|
|
extra.pop("GITEA_PROFILE_NAME", None)
|
|
env.update(extra)
|
|
env["GITEA_MCP_PROFILE"] = profile
|
|
return env
|
|
|
|
def _records(self):
|
|
if not os.path.exists(self.audit_path):
|
|
return []
|
|
with open(self.audit_path, encoding="utf-8") as fh:
|
|
return [json.loads(line) for line in fh if line.strip()]
|
|
|
|
|
|
class TestSimpleToolAudit(_AuditWiringBase):
|
|
|
|
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
|
return_value=(True, []))
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.api_get_all", return_value=[])
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_create_issue_success_audited(self, _auth, _get_all, mock_api, _role):
|
|
# 1: create POST result, 2: identity /user lookup for the audit record.
|
|
mock_api.side_effect = [
|
|
{"number": 11, "html_url": "https://gitea.prgs.cc/issues/11"},
|
|
{"login": "author-bot"},
|
|
]
|
|
with patch.dict(os.environ, self._env(), clear=True):
|
|
result = gitea_create_issue(title="Add thing", remote="prgs")
|
|
self.assertEqual(result["number"], 11)
|
|
recs = self._records()
|
|
self.assertEqual(len(recs), 1)
|
|
rec = recs[0]
|
|
self.assertEqual(rec["action"], "create_issue")
|
|
self.assertEqual(rec["result"], "succeeded")
|
|
self.assertIn(rec["profile_name"], ("gitea-author", "test-author-prgs", "prgs-author"))
|
|
self.assertEqual(rec["authenticated_username"], "author-bot")
|
|
self.assertEqual(rec["issue_number"], 11)
|
|
self.assertEqual(rec["request_metadata"]["title"], "Add thing")
|
|
|
|
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
|
return_value=(True, []))
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.api_get_all", return_value=[])
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_create_issue_failure_audited(self, _auth, _get_all, mock_api, _role):
|
|
mock_api.side_effect = [
|
|
RuntimeError("HTTP 500: boom"),
|
|
{"login": "author-bot"}, # identity lookup for the audit record
|
|
]
|
|
with patch.dict(os.environ, self._env(), clear=True):
|
|
with self.assertRaises(RuntimeError):
|
|
gitea_create_issue(title="X", remote="prgs")
|
|
recs = self._records()
|
|
self.assertEqual(len(recs), 1)
|
|
self.assertEqual(recs[0]["result"], "failed")
|
|
self.assertIn("HTTP 500", recs[0]["reason"])
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_close_issue_audited(self, _auth, mock_api):
|
|
# Keyed rather than positional: closing an issue also reads its labels
|
|
# before and after the state change for the #780 terminal cleanup and
|
|
# its read-after-write check, so call order is not a fixed sequence.
|
|
def api(method, url, auth, payload=None):
|
|
if method == "PATCH":
|
|
return {"state": "closed"}
|
|
if "/issues/" in url:
|
|
return {"number": 42, "labels": []}
|
|
return {"login": "mgr-bot"}
|
|
|
|
mock_api.side_effect = api
|
|
with patch.dict(os.environ, self._env(), clear=True):
|
|
gitea_close_issue(issue_number=42, remote="prgs")
|
|
recs = self._records()
|
|
self.assertEqual(len(recs), 1)
|
|
self.assertEqual(recs[0]["action"], "close_issue")
|
|
self.assertEqual(recs[0]["issue_number"], 42)
|
|
self.assertEqual(recs[0]["authenticated_username"], "mgr-bot")
|
|
|
|
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
|
return_value=(True, []))
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.api_get_all", return_value=[])
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_disabled_writes_nothing_and_no_extra_call(self, _auth, _get_all, mock_api, _role):
|
|
# No GITEA_AUDIT_LOG -> audit is a no-op: one create POST, no file.
|
|
mock_api.return_value = {"number": 1, "html_url": "http://x/1"}
|
|
with patch.dict(
|
|
os.environ,
|
|
shared_mutation_env("test-author-prgs"),
|
|
clear=True,
|
|
):
|
|
gitea_create_issue(title="x", remote="prgs")
|
|
issue_posts = [
|
|
c for c in mock_api.call_args_list if c.args[0] == "POST"
|
|
]
|
|
self.assertEqual(len(issue_posts), 1)
|
|
self.assertEqual(self._records(), [])
|
|
|
|
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
|
return_value=(True, []))
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.api_get_all", return_value=[])
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_secrets_never_written(self, _auth, _get_all, mock_api, _role):
|
|
mock_api.side_effect = [
|
|
{"number": 3, "html_url": "http://x/3"},
|
|
{"login": "author-bot"},
|
|
]
|
|
env = self._env(GITEA_TOKEN="super-secret-token")
|
|
with patch.dict(os.environ, env, clear=True):
|
|
gitea_create_issue(title="t", remote="prgs")
|
|
with open(self.audit_path, encoding="utf-8") as fh:
|
|
blob = fh.read().lower()
|
|
for secret in ("super-secret-token", "authorization", "basic ", FAKE_AUTH.lower()):
|
|
self.assertNotIn(secret, blob)
|
|
|
|
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
|
return_value=(True, []))
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.api_get_all", return_value=[])
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_audit_failure_never_breaks_action(self, _auth, _get_all, mock_api, _role):
|
|
mock_api.side_effect = [
|
|
{"number": 9, "html_url": "http://x/9"},
|
|
{"login": "author-bot"},
|
|
]
|
|
with patch.dict(os.environ, self._env(), clear=True):
|
|
with patch("gitea_audit.write_event", side_effect=RuntimeError("disk full")):
|
|
result = gitea_create_issue(title="t", remote="prgs")
|
|
# The mutation result is returned even though the sink write blew up.
|
|
self.assertEqual(result["number"], 9)
|
|
|
|
|
|
_NO_PR_WORK_LEASE_BLOCK = {"block": False, "reasons": [], "mutation_allowed": True}
|
|
|
|
|
|
class TestGatedToolAudit(_AuditWiringBase):
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
from tests.test_mcp_server import _init_reviewer_session, _install_owned_reviewer_lease
|
|
import reviewer_pr_lease
|
|
import review_workflow_boundary
|
|
import review_workflow_load
|
|
|
|
# Session init clears any prior session lease (#407) and loads workflow (#389).
|
|
_init_reviewer_session("prgs")
|
|
self.addCleanup(review_workflow_load.clear_review_workflow_load)
|
|
self.addCleanup(review_workflow_boundary.clear_pre_review_commands)
|
|
self._lease_patch = _install_owned_reviewer_lease(8)
|
|
self._lease_patch.start()
|
|
self._auth_identity_patch = patch(
|
|
"mcp_server._authenticated_username", return_value="reviewer-bot"
|
|
)
|
|
self._auth_identity_patch.start()
|
|
self._pr_lease_comments_patch = patch(
|
|
"mcp_server._list_pr_lease_comments", return_value=[]
|
|
)
|
|
self._pr_lease_comments_patch.start()
|
|
self._pr_work_lease_patch = patch(
|
|
"mcp_server._pr_work_lease_reviewer_block",
|
|
return_value=dict(_NO_PR_WORK_LEASE_BLOCK),
|
|
)
|
|
self._pr_work_lease_patch.start()
|
|
self.addCleanup(self._auth_identity_patch.stop)
|
|
self.addCleanup(self._lease_patch.stop)
|
|
self.addCleanup(self._pr_lease_comments_patch.stop)
|
|
self.addCleanup(self._pr_work_lease_patch.stop)
|
|
self.addCleanup(reviewer_pr_lease.clear_session_lease)
|
|
|
|
def _pr(self, author, state="open", sha="abc123", mergeable=True):
|
|
return {"user": {"login": author}, "state": state,
|
|
"head": {"sha": sha}, "mergeable": mergeable}
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_merge_success_audited(self, _auth, mock_api):
|
|
# user, pr, eligibility feedback pr+reviews (#695), gate-7 feedback
|
|
# pr+reviews, merge POST, readback.
|
|
approval = [{
|
|
"id": 1, "user": {"login": "reviewer-bot"}, "state": "APPROVED",
|
|
"commit_id": "abc123", "submitted_at": "2026-07-06T10:00:00Z",
|
|
"dismissed": False,
|
|
}]
|
|
mock_api.side_effect = [
|
|
{"login": "merger-bot"}, self._pr("author-bot"),
|
|
self._pr("author-bot"), approval, # eligibility merge feedback
|
|
self._pr("author-bot"), approval, # gate 7 feedback
|
|
{}, {"merged_commit_sha": "c1"},
|
|
]
|
|
env = self._env(GITEA_MCP_PROFILE="test-merger-prgs")
|
|
with patch.dict(os.environ, env, clear=True):
|
|
mcp_server.gitea_load_review_workflow()
|
|
r = gitea_merge_pr(pr_number=8, confirmation="MERGE PR 8",
|
|
expected_head_sha="abc123", remote="prgs")
|
|
self.assertTrue(r["performed"])
|
|
recs = self._records()
|
|
self.assertEqual(len(recs), 1)
|
|
self.assertEqual(recs[0]["action"], "merge_pr")
|
|
self.assertEqual(recs[0]["result"], "succeeded")
|
|
self.assertEqual(recs[0]["authenticated_username"], "merger-bot")
|
|
self.assertEqual(recs[0]["pr_number"], 8)
|
|
self.assertEqual(recs[0]["head_sha"], "abc123")
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_merge_blocked_audited(self, _auth, mock_api):
|
|
# Self-author merge is blocked; must still be recorded as blocked.
|
|
mock_api.side_effect = [{"login": "jcwalker3"}, self._pr("jcwalker3")]
|
|
env = self._env(GITEA_MCP_PROFILE="test-merger-prgs")
|
|
with patch.dict(os.environ, env, clear=True):
|
|
mcp_server.gitea_load_review_workflow()
|
|
r = gitea_merge_pr(pr_number=8, confirmation="MERGE PR 8", remote="prgs")
|
|
self.assertFalse(r["performed"])
|
|
recs = self._records()
|
|
self.assertEqual(len(recs), 1)
|
|
self.assertEqual(recs[0]["result"], "blocked")
|
|
self.assertEqual(recs[0]["authenticated_username"], "jcwalker3")
|
|
self.assertIn("PR author", recs[0]["reason"])
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_submit_review_success_audited(self, _auth, mock_api):
|
|
# mark_final_review_decision and submit each run eligibility (user + PR).
|
|
mock_api.side_effect = [
|
|
{"login": "reviewer-bot"}, self._pr("author-bot"),
|
|
{"login": "reviewer-bot"}, self._pr("author-bot"),
|
|
{"id": 7, "state": "APPROVED"},
|
|
[{"id": 7, "user": {"login": "reviewer-bot"}, "state": "APPROVED",
|
|
"submitted_at": "2026-07-06T10:00:00Z", "dismissed": False}],
|
|
]
|
|
env = self._env(GITEA_MCP_PROFILE="test-reviewer-prgs")
|
|
with patch.dict(os.environ, env, clear=True):
|
|
import session_context_binding as _sc
|
|
from tests.test_mcp_server import (
|
|
_init_reviewer_session,
|
|
_install_owned_reviewer_lease,
|
|
)
|
|
from mcp_server import gitea_mark_final_review_decision
|
|
|
|
# Rebind after profile env is applied so durable session state
|
|
# matches the active reviewer profile (#714 / #695).
|
|
_sc._reset_session_context_for_testing()
|
|
_init_reviewer_session("prgs")
|
|
lease = _install_owned_reviewer_lease(8)
|
|
lease.start()
|
|
self.addCleanup(lease.stop)
|
|
mcp_server.gitea_load_review_workflow()
|
|
gitea_mark_final_review_decision(
|
|
8, "approve", expected_head_sha="abc123", remote="prgs",
|
|
)
|
|
r = gitea_submit_pr_review(
|
|
pr_number=8, action="approve",
|
|
body="LGTM", remote="prgs",
|
|
final_review_decision_ready=True,
|
|
)
|
|
self.assertTrue(
|
|
r["performed"],
|
|
msg=f"submit_pr_review blocked: {r}",
|
|
)
|
|
recs = self._records()
|
|
self.assertEqual(len(recs), 1)
|
|
self.assertEqual(recs[0]["action"], "submit_pr_review")
|
|
self.assertEqual(recs[0]["result"], "succeeded")
|
|
self.assertEqual(recs[0]["authenticated_username"], "reviewer-bot")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|