diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index ed5a428..0619fc1 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -986,11 +986,11 @@ def verify_preflight_purity( if not skip_purity_order: if not _preflight_whoami_called: - raise RuntimeError( + raise _PreflightOrderError( "Pre-flight order violation: Identity (gitea_whoami) has not been verified (fail closed)" ) if not _preflight_capability_called: - raise RuntimeError( + raise _PreflightOrderError( "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)" ) if ( @@ -998,7 +998,7 @@ def verify_preflight_purity( and _preflight_resolved_task is not None and task != _preflight_resolved_task ): - raise RuntimeError( + raise _PreflightOrderError( "Pre-flight task mismatch: " f"resolved '{_preflight_resolved_task}' but mutation requires " f"'{task}' (fail closed)" @@ -1044,13 +1044,13 @@ def verify_preflight_purity( ) else: if _preflight_whoami_violation: - raise RuntimeError( + raise _PreflightOrderError( "Pre-flight order violation: Workspace file edits occurred before " f"gitea_whoami verification (fail closed). Offending files: " f"{_format_preflight_files(_preflight_whoami_violation_files)}" ) if _preflight_capability_violation: - raise RuntimeError( + raise _PreflightOrderError( "Pre-flight order violation: Workspace file edits occurred before " f"gitea_resolve_task_capability verification (fail closed). Offending files: " f"{_format_preflight_files(_preflight_capability_violation_files)}" @@ -2491,6 +2491,40 @@ def _audit(action: str, *, host, remote, result, org=None, repo=None, pass # best-effort; never break the action +class _PreflightOrderError(RuntimeError): + """Pre-flight order/precondition violation for a mutation (fail closed). + + Carries a safe ``gitea_reason_code`` so the tool-error boundary converts it + into a typed structured error (``preflight_order_violation``) instead of an + opaque ``internal_error``. It is still a ``RuntimeError`` subclass so every + existing ``except RuntimeError`` / message assertion keeps working. + """ + + gitea_reason_code = "preflight_order_violation" + + +@contextlib.contextmanager +def _mutation_stage(stage: str): + """Tag any exception escaping *stage* with a safe stage name (#diagnostics). + + On exception, records ``stage`` on the exception (innermost wins) so the + tool-error boundary can report ``mutation_stage=`` alongside the + redacted class/detail. Never swallows, never alters control flow, and never + overwrites a more specific (inner) stage already set. + """ + try: + yield + except BaseException as exc: # noqa: BLE001 - tag-and-reraise only + try: + if isinstance(exc, Exception) and not getattr( + exc, "_gitea_mutation_stage", None + ): + setattr(exc, "_gitea_mutation_stage", stage) + except Exception: + pass + raise + + @contextlib.contextmanager def _audited(action: str, *, host, remote, org=None, repo=None, request_metadata=None, issue_number=None, pr_number=None, @@ -2661,9 +2695,10 @@ def gitea_create_issue( if blocked: return blocked try: - verify_preflight_purity( - remote, worktree_path=worktree_path, task="create_issue" - ) + with _mutation_stage("preflight_purity"): + verify_preflight_purity( + remote, worktree_path=worktree_path, task="create_issue" + ) except Exception as exc: typed = _production_guard_block_from_exc(exc, number=None) if typed is not None: @@ -3114,7 +3149,8 @@ def gitea_create_pr( ) if blocked: return blocked - verify_preflight_purity(remote, worktree_path=worktree_path, task="create_pr") + with _mutation_stage("preflight_purity"): + verify_preflight_purity(remote, worktree_path=worktree_path, task="create_pr") h, o, r = _resolve(remote, host, org, repo) # ── Issue Lock Validation (Issue #194 / #196 / #443) ── @@ -8616,9 +8652,11 @@ def gitea_delete_branch( "audit_phase": audit_reconciliation_mode.current_phase(), } - verify_preflight_purity(remote, task="delete_branch") - h, o, r = _resolve(remote, host, org, repo) - auth = _auth(h) + with _mutation_stage("preflight_purity"): + verify_preflight_purity(remote, task="delete_branch") + with _mutation_stage("resolve_auth"): + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) import urllib.parse encoded_branch = urllib.parse.quote(branch, safe="") url = f"{repo_api_url(h, o, r)}/branches/{encoded_branch}" @@ -8626,8 +8664,9 @@ def gitea_delete_branch( "branch": branch, "required_permission": "gitea.branch.delete", } - with _audited("delete_branch", host=h, remote=remote, org=o, repo=r, - target_branch=branch, request_metadata=request_metadata): + with _mutation_stage("api_delete"), _audited( + "delete_branch", host=h, remote=remote, org=o, repo=r, + target_branch=branch, request_metadata=request_metadata): api_request("DELETE", url, auth) return {"success": True, "message": f"Remote branch '{branch}' deleted."} diff --git a/mcp_tool_error_boundary.py b/mcp_tool_error_boundary.py index 81f6828..9cd9cdb 100644 --- a/mcp_tool_error_boundary.py +++ b/mcp_tool_error_boundary.py @@ -23,6 +23,7 @@ from __future__ import annotations import json import logging +import re import sys from typing import Any @@ -38,6 +39,9 @@ REASON_CONFIG_ERROR = "config_error" REASON_INTERNAL_ERROR = "internal_error" REASON_UPSTREAM_UNAVAILABLE = "upstream_unavailable" REASON_HTTP_ERROR = "http_error" +# Typed structured errors for previously-opaque raw mutation failures. +REASON_PREFLIGHT_ORDER = "preflight_order_violation" +REASON_MALFORMED_RESPONSE = "malformed_response" ERROR_CLASS_AUTHENTICATION = "authentication" ERROR_CLASS_AUTHORIZATION = "authorization" @@ -45,6 +49,7 @@ ERROR_CLASS_NETWORK = "network" ERROR_CLASS_CONFIGURATION = "configuration" ERROR_CLASS_INTERNAL = "internal" ERROR_CLASS_UPSTREAM = "upstream" +ERROR_CLASS_PRECONDITION = "precondition" # Fixed, secret-free operator messages. Never interpolate HTTP bodies, # Keychain contents, tokens, or arbitrary exception text. @@ -62,12 +67,128 @@ FIXED_MESSAGES: dict[str, str] = { REASON_INTERNAL_ERROR: "Internal tool error", REASON_UPSTREAM_UNAVAILABLE: "Gitea upstream unavailable", REASON_HTTP_ERROR: "Gitea HTTP request failed", + REASON_PREFLIGHT_ORDER: ( + "Mutation blocked: pre-flight order violation (fail closed)" + ), + REASON_MALFORMED_RESPONSE: "Malformed response from Gitea", } +# Error class for a self-declared safe reason code (``gitea_reason_code``). +_REASON_ERROR_CLASS: dict[str, str] = { + REASON_AUTH_FAILED: ERROR_CLASS_AUTHENTICATION, + REASON_AUTH_INVALID_TOKEN: ERROR_CLASS_AUTHENTICATION, + REASON_AUTHZ_INSUFFICIENT_SCOPE: ERROR_CLASS_AUTHORIZATION, + REASON_AUTHZ_DENIED: ERROR_CLASS_AUTHORIZATION, + REASON_NETWORK_ERROR: ERROR_CLASS_NETWORK, + REASON_CONFIG_ERROR: ERROR_CLASS_CONFIGURATION, + REASON_UPSTREAM_UNAVAILABLE: ERROR_CLASS_UPSTREAM, + REASON_HTTP_ERROR: ERROR_CLASS_INTERNAL, + REASON_PREFLIGHT_ORDER: ERROR_CLASS_PRECONDITION, + REASON_MALFORMED_RESPONSE: ERROR_CLASS_INTERNAL, + REASON_INTERNAL_ERROR: ERROR_CLASS_INTERNAL, +} + +# Bounds for the safe diagnostic fields added to the ``internal_error`` path. +_DETAIL_LIMIT = 200 +_CLASS_LIMIT = 120 +_STAGE_LIMIT = 64 + +# Absolute-path prefixes stripped from diagnostic detail (never leak layout). +_ABS_PATH_RE = re.compile(r"/(?:Users|home|private|var|tmp|opt|etc|root)/[^\s'\"]*") +_DETAIL_SECRET_PREFIXES = ("token ", "Basic ", "Bearer ", "Authorization: ") +_SECRET_KEY = ( + r"token|password|passwd|secret|authorization|api[_-]?key|" + r"access[_-]?token|refresh[_-]?token|session|cookie" +) +_SECRET_JSON_RE = re.compile( + r'"(' + _SECRET_KEY + r')"\s*:\s*"[^"]*"', re.IGNORECASE +) +_SECRET_KV_RE = re.compile( + r"\b(" + _SECRET_KEY + r")\s*=\s*[^\s&\"']+", re.IGNORECASE +) + +_MUTATION_STAGE_ATTR = "_gitea_mutation_stage" +_SELF_DECLARED_REASON_ATTR = "gitea_reason_code" + _INSTALL_FLAG = "_gitea_auth_boundary_installed" _ORIGINAL_ATTR = "_gitea_auth_boundary_original" +def _safe_str(exc: BaseException) -> str: + """``str(exc)`` that never raises (a poisoned ``__str__`` must not escape).""" + try: + return str(exc) + except Exception: + return "" + + +def _safe_exception_class(exc: BaseException) -> str: + """Fully-qualified type identifier for *exc* — never instance/message text.""" + try: + t = type(exc) + mod = getattr(t, "__module__", "") or "" + name = getattr(t, "__qualname__", None) or getattr(t, "__name__", "") or "" + ident = f"{mod}.{name}" if mod else name + return ident[:_CLASS_LIMIT] + except Exception: + return "unknown" + + +def _redact_detail(text: Any, limit: int = _DETAIL_LIMIT) -> str: + """Strictly redact free-form exception text for safe diagnostics. + + Removes token/Authorization credentials, raw URLs and hostnames (via the + shared audit redactor), and absolute filesystem paths, then collapses + whitespace and truncates. Never raises; on any failure returns "". + """ + if not text: + return "" + try: + out = str(text) + # Drop credential-prefixed runs (token/Basic/Bearer/Authorization). + for prefix in _DETAIL_SECRET_PREFIXES: + idx = 0 + while True: + i = out.find(prefix, idx) + if i == -1: + break + j = i + len(prefix) + while j < len(out) and not out[j].isspace(): + j += 1 + out = out[:i] + prefix + "[REDACTED]" + out[j:] + idx = i + len(prefix) + len("[REDACTED]") + # Secret VALUES carried in JSON ("key":"value") or kv (key=value) form, + # even short ones (e.g. a password), keyed by a sensitive field name. + out = _SECRET_JSON_RE.sub(r'"\1":"[REDACTED]"', out) + out = _SECRET_KV_RE.sub(r"\1=[REDACTED]", out) + # URLs / query secrets / hostnames. + try: + import gitea_audit + + out = gitea_audit.redact_urls(out) + except Exception: + pass + # Absolute filesystem paths (workspace layout is not for LLM output). + out = _ABS_PATH_RE.sub("[PATH]", out) + # Any bare hostname the URL redactor missed (defence in depth). + out = re.sub(r"\b[\w.-]+\.(?:cc|net|com|org|io|dev|local)\b", "[HOST]", out) + out = " ".join(out.split()) + return out[:limit] + except Exception: + return "" + + +def _safe_mutation_stage(exc: BaseException) -> str | None: + """Return the bounded, redacted mutation stage tagged on *exc* (or None).""" + try: + raw = getattr(exc, _MUTATION_STAGE_ATTR, None) + if not raw: + return None + return _redact_detail(raw, limit=_STAGE_LIMIT) or None + except Exception: + return None + + def fixed_message(reason_code: str) -> str: """Return the fixed sanitized message for *reason_code* (fail closed).""" return FIXED_MESSAGES.get(reason_code, FIXED_MESSAGES[REASON_INTERNAL_ERROR]) @@ -154,9 +275,34 @@ def classify_exception(exc: BaseException) -> dict[str, Any]: } +def _self_declared_classification(exc: BaseException) -> dict[str, Any] | None: + """Honor a safe ``gitea_reason_code`` attribute set by server-side raisers. + + Converts a raw exception that self-declares a **known** reason code into a + typed structured error (fixed message) instead of an opaque internal_error. + Unknown / non-string / secret values are ignored (fail closed to internal). + """ + reason = getattr(exc, _SELF_DECLARED_REASON_ATTR, None) + if not isinstance(reason, str) or reason not in FIXED_MESSAGES: + return None + if reason == REASON_INTERNAL_ERROR: + return None + return { + "reason_code": reason, + "error_class": _REASON_ERROR_CLASS.get(reason, ERROR_CLASS_INTERNAL), + "http_status": None, + "message": fixed_message(reason), + "transport_survives": True, + } + + def _classify_exception_impl(exc: BaseException) -> dict[str, Any]: import gitea_auth + declared = _self_declared_classification(exc) + if declared is not None: + return declared + if isinstance(exc, gitea_auth.GiteaAuthError): code = getattr(exc, "reason_code", None) or REASON_AUTH_INVALID_TOKEN if code not in ( @@ -233,13 +379,23 @@ def _classify_exception_impl(exc: BaseException) -> dict[str, Any]: return _classify_exception_impl(cause) # No message-substring authentication heuristics (reviewer finding #3). - return { + # Unexpected failure → internal_error, now carrying SAFE diagnostics so the + # failure is actionable: a type identifier + strictly-redacted detail + + # optional mutation-stage tag. The operator ``message`` stays the fixed + # constant; none of these fields carry secrets/bodies/paths/env. + result = { "reason_code": REASON_INTERNAL_ERROR, "error_class": ERROR_CLASS_INTERNAL, "http_status": None, "message": fixed_message(REASON_INTERNAL_ERROR), "transport_survives": True, + "exception_class": _safe_exception_class(exc), + "detail": _redact_detail(_safe_str(exc)), } + stage = _safe_mutation_stage(exc) + if stage: + result["mutation_stage"] = stage + return result def build_structured_error_payload( @@ -277,6 +433,19 @@ def build_structured_error_payload( payload["tool"] = tool_name[:120] if profile_name and isinstance(profile_name, str): payload["profile"] = profile_name[:80] + # Safe diagnostics — internal_error path only. These make an otherwise + # opaque crash actionable without leaking secrets. Re-derived here from + # the fixed message gate above; typed reasons never carry them. + if payload["reason_code"] == REASON_INTERNAL_ERROR: + exc_cls = classification.get("exception_class") + if isinstance(exc_cls, str) and exc_cls: + payload["exception_class"] = exc_cls[:_CLASS_LIMIT] + detail = classification.get("detail") + if isinstance(detail, str): + payload["detail"] = _redact_detail(detail) + stage = classification.get("mutation_stage") + if isinstance(stage, str) and stage: + payload["mutation_stage"] = stage[:_STAGE_LIMIT] return payload except Exception: return { @@ -316,7 +485,21 @@ def log_sanitized_daemon_reason( status = classification.get("http_status") if isinstance(status, int): parts.append(f"http_status={status}") - # Intentionally no detail= / message= field — secrets lived there. + # Safe diagnostics — internal_error path ONLY. Typed reasons still emit + # no detail= (secrets lived there). class/detail/stage are re-redacted + # here as defence in depth. + if reason == REASON_INTERNAL_ERROR: + exc_cls = classification.get("exception_class") + if isinstance(exc_cls, str) and exc_cls: + parts.append(f"exception_class={exc_cls[:_CLASS_LIMIT]}") + stage = classification.get("mutation_stage") + if isinstance(stage, str) and stage: + parts.append( + f"mutation_stage={_redact_detail(stage, limit=_STAGE_LIMIT)}" + ) + detail = classification.get("detail") + if isinstance(detail, str) and detail: + parts.append(f"detail={_redact_detail(detail)}") line = " ".join(parts) stream.write(line + "\n") if hasattr(stream, "flush"): diff --git a/tests/test_mutation_error_diagnostics.py b/tests/test_mutation_error_diagnostics.py new file mode 100644 index 0000000..ec721a4 --- /dev/null +++ b/tests/test_mutation_error_diagnostics.py @@ -0,0 +1,228 @@ +"""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:hunter2@gitea.prgs.cc/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()