Gitea-Tools/tests/test_in_progress_cleanup.py

"""Tests for status:in-progress auto-release on close/merge (#58).

Covers the deterministic helpers (extract_linked_issue_numbers,
release_in_progress_label, cleanup_in_progress_for_pr) with a fake api_request —
no real network, no real labels, no secrets.
"""
import os
import sys
import json
import tempfile
import unittest
from unittest.mock import patch

sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))

import mcp_server as m  # noqa: E402

BASE = "https://gitea.example.invalid/api/v1/repos/Org/Repo"
CTX = {"base": BASE, "auth": "token x", "host": "gitea.example.invalid",
       "remote": "prgs", "org": "Org", "repo": "Repo", "trigger": "test"}
LABEL = "status:in-progress"


class FakeApi:
    """Routes label lookups/removals; records calls; can fail chosen deletes."""

    def __init__(self, repo_has_label=True, issue_labels=None, fail_delete=()):
        self.repo_labels = [{"id": 10, "name": LABEL},
                            {"id": 11, "name": "important"}] if repo_has_label \
            else [{"id": 11, "name": "important"}]
        self.issue_labels = issue_labels or {}   # {issue_no: [label names]}
        self.fail_delete = set(fail_delete)
        self.calls = []

    def _issue_no(self, url):
        return int(url.split("/issues/")[1].split("/")[0])

    def __call__(self, method, url, auth=None, payload=None):
        self.calls.append((method, url))
        if "/api/v1/user" in url:
            return {"login": "bot"}
        if method == "GET" and url.endswith("/labels?limit=100"):
            return list(self.repo_labels)
        if method == "GET" and "/issues/" in url and url.endswith("/labels"):
            n = self._issue_no(url)
            return [{"name": x} for x in self.issue_labels.get(n, [])]
        if method == "DELETE" and "/issues/" in url and "/labels/" in url:
            n = self._issue_no(url)
            if n in self.fail_delete:
                raise RuntimeError("HTTP 500: label remove failed")
            return None
        return {}

    def deletes(self):
        return [u for (mth, u) in self.calls if mth == "DELETE"]


# ---------------------------------------------------------------------------
# extract_linked_issue_numbers (pure)
# ---------------------------------------------------------------------------
class TestExtract(unittest.TestCase):

    def test_keywords(self):
        self.assertEqual(m.extract_linked_issue_numbers("Closes #123"), [123])
        self.assertEqual(m.extract_linked_issue_numbers("fixes #7 and Resolves #8"), [7, 8])
        self.assertEqual(m.extract_linked_issue_numbers("Refs #42"), [42])

    def test_branch(self):
        self.assertEqual(
            m.extract_linked_issue_numbers("", "fix/issue-123-slug"), [123])
        self.assertEqual(
            m.extract_linked_issue_numbers("", "feat/issue-9-x"), [9])

    def test_review_branch_not_treated_as_issue(self):
        self.assertEqual(m.extract_linked_issue_numbers("", "review/pr-456-x"), [])

    def test_multiple_dedup_sorted(self):
        self.assertEqual(
            m.extract_linked_issue_numbers("Closes #5, refs #2", "fix/issue-2-a"),
            [2, 5])

    def test_none_when_no_links(self):
        self.assertEqual(m.extract_linked_issue_numbers("just a message"), [])


# ---------------------------------------------------------------------------
# release_in_progress_label
# ---------------------------------------------------------------------------
class TestRelease(unittest.TestCase):

    def test_removes_present_label(self):
        api = FakeApi(issue_labels={123: [LABEL, "bug"]})
        r = m.release_in_progress_label([123], CTX, api_request=api)
        self.assertEqual(r["released"], [123])
        self.assertEqual(r["failed"], {})
        # Only the status:in-progress label id (10) is deleted; "bug" untouched.
        self.assertEqual(api.deletes(), [f"{BASE}/issues/123/labels/10"])

    def test_absent_label_is_noop(self):
        api = FakeApi(issue_labels={123: ["bug"]})
        r = m.release_in_progress_label([123], CTX, api_request=api)
        self.assertEqual(r["absent"], [123])
        self.assertEqual(r["released"], [])
        self.assertEqual(api.deletes(), [])  # nothing removed

    def test_no_issues_is_noop(self):
        api = FakeApi()
        r = m.release_in_progress_label([], CTX, api_request=api)
        self.assertEqual(r["linked_issues"], [])
        self.assertEqual(api.calls, [])  # zero network

    def test_multiple_issues(self):
        api = FakeApi(issue_labels={1: [LABEL], 2: [LABEL]})
        r = m.release_in_progress_label([1, 2], CTX, api_request=api)
        self.assertEqual(sorted(r["released"]), [1, 2])

    def test_failure_reported_not_clean(self):
        api = FakeApi(issue_labels={1: [LABEL], 2: [LABEL]}, fail_delete={2})
        r = m.release_in_progress_label([1, 2], CTX, api_request=api)
        self.assertEqual(r["released"], [1])
        self.assertIn(2, r["failed"])
        self.assertIn("failed", r["failed"][2].lower())

    def test_label_not_defined_in_repo(self):
        api = FakeApi(repo_has_label=False, issue_labels={1: ["important"]})
        r = m.release_in_progress_label([1], CTX, api_request=api)
        self.assertEqual(r["absent"], [1])
        self.assertEqual(api.deletes(), [])

    def test_only_status_label_removed(self):
        api = FakeApi(issue_labels={1: [LABEL, "important", "bug"]})
        m.release_in_progress_label([1], CTX, api_request=api)
        self.assertEqual(api.deletes(), [f"{BASE}/issues/1/labels/10"])  # id 10 only


# ---------------------------------------------------------------------------
# cleanup_in_progress_for_pr (merge / close-without-merge share this)
# ---------------------------------------------------------------------------
class TestCleanupForPr(unittest.TestCase):

    def test_closes_keyword_releases(self):
        api = FakeApi(issue_labels={123: [LABEL]})
        pr = {"title": "x", "body": "Closes #123", "head": {"ref": "wip"}}
        r = m.cleanup_in_progress_for_pr(pr, CTX, api_request=api)
        self.assertEqual(r["released"], [123])

    def test_branch_releases(self):
        api = FakeApi(issue_labels={123: [LABEL]})
        pr = {"title": "x", "body": "", "head": {"ref": "fix/issue-123-slug"}}
        r = m.cleanup_in_progress_for_pr(pr, CTX, api_request=api)
        self.assertEqual(r["released"], [123])

    def test_close_without_merge_does_not_close_issue(self):
        # cleanup only ever removes a label; it never PATCHes issue state.
        api = FakeApi(issue_labels={123: [LABEL]})
        pr = {"title": "", "body": "Refs #123", "head": {"ref": "wip"}}
        m.cleanup_in_progress_for_pr(pr, CTX, api_request=api)
        self.assertFalse(any(mth == "PATCH" for (mth, _u) in api.calls))

    def test_no_linked_issue_no_mutation(self):
        api = FakeApi()
        pr = {"title": "misc", "body": "no links", "head": {"ref": "wip"}}
        r = m.cleanup_in_progress_for_pr(pr, CTX, api_request=api)
        self.assertEqual(r["linked_issues"], [])
        self.assertEqual(api.deletes(), [])


# ---------------------------------------------------------------------------
# Audit records the label mutation when enabled
# ---------------------------------------------------------------------------
class TestAudit(unittest.TestCase):

    def test_audit_records_release(self):
        api = FakeApi(issue_labels={7: [LABEL]})
        with tempfile.TemporaryDirectory() as d:
            log = os.path.join(d, "audit.log")
            env = {"GITEA_AUDIT_LOG": log, "GITEA_PROFILE_NAME": "gitea-merger"}
            # get_auth_header → None so _audit resolves username without network.
            with patch.dict(os.environ, env, clear=True), \
                    patch("mcp_server.get_auth_header", return_value=None):
                m.release_in_progress_label([7], CTX, api_request=api)
            recs = [json.loads(x) for x in open(log, encoding="utf-8") if x.strip()]
        self.assertTrue(any(r["action"] == "unlabel_issue" and r["issue_number"] == 7
                            and r["result"] == "succeeded" for r in recs))


if __name__ == "__main__":
    unittest.main()