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:pass@example.com/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):
        mock_api.side_effect = [{"state": "closed"}, {"login": "mgr-bot"}]
        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()
