Gitea-Tools/tests/test_mutation_error_diagnostics.py

"""Safe mutation-error diagnostics for the tool boundary.

The #699/#701 boundary intentionally collapses every non-typed exception to an
opaque ``internal_error`` with the class and message discarded. That makes a
mutation failure (create_issue / create_pr / delete_branch / issue-lock)
undiagnosable: reason_code=internal_error, retryable=false, no actionable
detail.

This suite pins the additive contract:

1. For the ``internal_error`` fallback ONLY, capture a **safe exception class**
   (a Python type identifier — never secrets) and a **strictly-redacted**
   detail (no tokens, Authorization headers, raw URLs/hosts, absolute paths,
   response bodies, env dumps), plus an optional ``mutation_stage`` tag.
2. Typed classifications (auth/authz/network/config/http) are unchanged and
   still emit fixed messages with NO detail in the daemon log.
3. Any exception may self-declare a safe ``gitea_reason_code`` to be converted
   into a typed structured error instead of an opaque internal_error.
"""
from __future__ import annotations

import io
import json
import unittest

import gitea_auth
import mcp_tool_error_boundary as boundary

# Adversarial text a poisoned internal exception might carry.
POISON = (
    "boom at /Users/jasonwalker/secret/creds.json "
    "Authorization: token ghp_leaked_secret_xyz "
    "https://sysadmin:[email protected]/api/v1/x?token=supersecretXYZ "
    'Bearer abc.def.ghi {"token":"supersecretXYZ","password":"hunter2"}'
)
SECRETS = (
    "ghp_leaked_secret_xyz",
    "supersecretXYZ",
    "hunter2",
    "/Users/jasonwalker/secret",
    "gitea.prgs.cc",
)


class TestRedactDetail(unittest.TestCase):
    def test_strips_secrets_paths_urls_and_truncates(self):
        out = boundary._redact_detail(POISON)
        for s in SECRETS:
            self.assertNotIn(s, out)
        self.assertLessEqual(len(out), boundary._DETAIL_LIMIT)

    def test_empty_is_safe(self):
        self.assertEqual(boundary._redact_detail(""), "")
        self.assertEqual(boundary._redact_detail(None), "")


class TestSafeExceptionClass(unittest.TestCase):
    def test_fully_qualified_type_identifier(self):
        self.assertEqual(
            boundary._safe_exception_class(RuntimeError("x")), "builtins.RuntimeError"
        )
        self.assertEqual(
            boundary._safe_exception_class(KeyError("x")), "builtins.KeyError"
        )

    def test_class_name_never_carries_instance_text(self):
        # The class identifier must not include the (possibly secret) message.
        cls = boundary._safe_exception_class(RuntimeError(POISON))
        for s in SECRETS:
            self.assertNotIn(s, cls)


class TestInternalErrorGainsDiagnostics(unittest.TestCase):
    def test_classify_internal_has_class_and_redacted_detail(self):
        c = boundary.classify_exception(RuntimeError(POISON))
        self.assertEqual(c["reason_code"], "internal_error")
        self.assertEqual(c["error_class"], "internal")
        self.assertEqual(c["exception_class"], "builtins.RuntimeError")
        self.assertIn("detail", c)
        for s in SECRETS:
            self.assertNotIn(s, json.dumps(c))

    def test_payload_surfaces_class_and_detail_but_fixed_message(self):
        p = boundary.build_structured_error_payload(
            boundary.classify_exception(RuntimeError(POISON)),
            tool_name="gitea_delete_branch",
        )
        self.assertEqual(p["reason_code"], "internal_error")
        self.assertEqual(p["exception_class"], "builtins.RuntimeError")
        self.assertIn("detail", p)
        self.assertEqual(p["message"], boundary.fixed_message("internal_error"))
        for s in SECRETS:
            self.assertNotIn(s, json.dumps(p))

    def test_daemon_log_internal_includes_class_and_detail_no_secrets(self):
        buf = io.StringIO()
        c = boundary.classify_exception(RuntimeError(POISON))
        boundary.log_sanitized_daemon_reason(
            c, tool_name="gitea_delete_branch", stream=buf
        )
        line = buf.getvalue()
        self.assertIn("reason_code=internal_error", line)
        self.assertIn("exception_class=builtins.RuntimeError", line)
        self.assertIn("detail=", line)
        for s in SECRETS:
            self.assertNotIn(s, line)


class TestTypedPathsUnchanged(unittest.TestCase):
    def test_auth_payload_has_no_detail_or_class(self):
        p = boundary.build_structured_error_payload(
            boundary.classify_exception(gitea_auth.GiteaAuthError())
        )
        self.assertNotIn("detail", p)
        self.assertNotIn("exception_class", p)

    def test_auth_daemon_log_has_no_detail(self):
        buf = io.StringIO()
        boundary.log_sanitized_daemon_reason(
            boundary.classify_exception(gitea_auth.GiteaAuthError()),
            tool_name="gitea_whoami",
            stream=buf,
        )
        self.assertNotIn("detail=", buf.getvalue())
        self.assertNotIn("exception_class=", buf.getvalue())


class TestMutationStageTag(unittest.TestCase):
    def test_stage_flows_to_payload_and_log(self):
        exc = RuntimeError("boom")
        setattr(exc, "_gitea_mutation_stage", "preflight_purity")
        c = boundary.classify_exception(exc)
        self.assertEqual(c["mutation_stage"], "preflight_purity")
        p = boundary.build_structured_error_payload(c, tool_name="gitea_create_pr")
        self.assertEqual(p["mutation_stage"], "preflight_purity")
        buf = io.StringIO()
        boundary.log_sanitized_daemon_reason(c, tool_name="gitea_create_pr", stream=buf)
        self.assertIn("mutation_stage=preflight_purity", buf.getvalue())

    def test_stage_value_is_bounded_and_safe(self):
        exc = RuntimeError("boom")
        setattr(exc, "_gitea_mutation_stage", "x" * 500 + POISON)
        c = boundary.classify_exception(exc)
        for s in SECRETS:
            self.assertNotIn(s, c["mutation_stage"])
        self.assertLessEqual(len(c["mutation_stage"]), 64)


class TestGiteaReasonCodeConversion(unittest.TestCase):
    def test_self_declared_reason_becomes_typed(self):
        exc = RuntimeError("Pre-flight order violation: whoami missing")
        setattr(exc, "gitea_reason_code", "preflight_order_violation")
        c = boundary.classify_exception(exc)
        self.assertEqual(c["reason_code"], "preflight_order_violation")
        self.assertNotEqual(c["reason_code"], "internal_error")
        p = boundary.build_structured_error_payload(c, tool_name="gitea_delete_branch")
        self.assertEqual(p["reason_code"], "preflight_order_violation")
        self.assertEqual(
            p["message"], boundary.fixed_message("preflight_order_violation")
        )

    def test_unknown_self_declared_reason_ignored(self):
        exc = RuntimeError("boom")
        setattr(exc, "gitea_reason_code", "not_a_real_reason")
        c = boundary.classify_exception(exc)
        self.assertEqual(c["reason_code"], "internal_error")

    def test_preflight_reason_is_registered(self):
        self.assertIn("preflight_order_violation", boundary.FIXED_MESSAGES)


class TestAdversarialInternalNoLeak(unittest.TestCase):
    def test_poisoned_internal_never_leaks_in_result_or_log(self):
        exc = RuntimeError(POISON)
        buf = io.StringIO()
        result = boundary.to_call_tool_result(
            exc, tool_name="gitea_delete_branch", log=False
        )
        c = boundary.classify_exception(exc)
        boundary.log_sanitized_daemon_reason(
            c, tool_name="gitea_delete_branch", stream=buf
        )
        blob = (
            json.dumps(result.structuredContent)
            + result.content[0].text
            + buf.getvalue()
        )
        for s in SECRETS:
            self.assertNotIn(s, blob)


class TestServerSideTypedAndStage(unittest.TestCase):
    """Server raisers are typed and stage-taggable, end-to-end into the boundary."""

    def test_preflight_order_error_is_typed_runtimeerror(self):
        import gitea_mcp_server as s

        exc = s._PreflightOrderError("whoami missing")
        self.assertIsInstance(exc, RuntimeError)  # legacy except RuntimeError
        self.assertEqual(exc.gitea_reason_code, "preflight_order_violation")
        c = boundary.classify_exception(exc)
        self.assertEqual(c["reason_code"], "preflight_order_violation")
        self.assertNotEqual(c["reason_code"], "internal_error")

    def test_mutation_stage_tags_and_reraises(self):
        import gitea_mcp_server as s

        with self.assertRaises(RuntimeError) as ctx:
            with s._mutation_stage("preflight_purity"):
                raise RuntimeError("boom")
        self.assertEqual(
            getattr(ctx.exception, "_gitea_mutation_stage", None), "preflight_purity"
        )

    def test_mutation_stage_inner_wins(self):
        import gitea_mcp_server as s

        with self.assertRaises(RuntimeError) as ctx:
            with s._mutation_stage("outer"):
                with s._mutation_stage("inner"):
                    raise RuntimeError("boom")
        self.assertEqual(
            getattr(ctx.exception, "_gitea_mutation_stage", None), "inner"
        )


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