Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 69 Pull Requests 1 Actions Packages Projects Releases Wiki Activity
Files
059ee77c1fd483da385d6ccb4d842e808fb54630
Gitea-Tools/tests/test_sentry_incident_bridge.py
T
jcwalker3 bc968dd2e0 fix: post AC4 recurrence comments on linked incident issues (#607) 
Remediates review 479/481 findings F1 and F2 on PR #767.

F1: issue #607 AC4 requires that an existing linked Gitea issue receives a
recurrence comment when Sentry events continue. That path did not exist.
This adds:

- incident_bridge.incident_recurred() — true only when event_count or
  last_seen genuinely advanced, compared against the pre-upsert link row, so
  an unchanged rescan stays silent and the exactly-once property holds.
- incident_bridge.build_recurrence_comment_body() — reuses the same
  redact_text path as build_gitea_issue_body, so redaction is not
  re-implemented.
- A comment_issue_fn hook on reconcile_incident, invoked only on
  OUTCOME_UPDATED with genuinely new events. The durable incident_links row
  is written first, so a comment failure degrades to "link updated, comment
  withheld" without corrupting the mapping or creating a duplicate issue.
- _incident_recurrence_comment_fn() in gitea_mcp_server.py, routing through
  the sanctioned gitea_create_issue_comment path named in issue #607. It
  returns None on dry runs and when the profile lacks gitea.issue.comment, so
  the AC8 dry-run default and disabled-mode safety hold.
- comment_issue_fn threading through sentry_incident_bridge.watchdog(), with
  the per-issue recurrence_comment outcome recorded in the scan result.

F2: observation_from_issue never populates a fingerprint, so the "deduped by
Sentry issue id and fingerprint" claim was unsupported. The
gitea_sentry_reconcile_issue and gitea_sentry_watchdog docstrings now state
the real dedupe basis: provider identity (provider, base URL, org, project,
Sentry issue id).

Tests: five new AC4 cases in tests/test_sentry_incident_bridge.py covering a
comment on the second scan, dry-run silence, no-new-events silence, link
durability when the comment fails, and comment redaction. Focused suite
43 passed (was 38).

Refs #607
2026-07-20 04:08:52 -05:00

666 lines
26 KiB
Python
Raw Blame History

"""Tests for the Sentry → Gitea incident bridge (#607).
Covers AC9: create, update, dedupe, closed-linked issue, redaction,
pagination, missing token, unavailable Sentry server, and self-hosted base URL.
No live Sentry: the HTTP layer is injected via ``http_fn``.
"""
from __future__ import annotations
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
import json
import os
import tempfile
import unittest
import unittest.mock
import urllib.error
import urllib.request
from control_plane_db import ControlPlaneDB
from incident_bridge import (
OUTCOME_CREATED,
OUTCOME_PREVIEW,
OUTCOME_UPDATED,
ProjectMapping,
)
import sentry_incident_bridge as bridge
BASE_URL = "https://sentry.prgs.cc"
SENTRY_ORG = "prgs"
SENTRY_PROJECT = "gitea-tools-mcp"
GITEA_ORG = "Scaled-Tech-Consulting"
GITEA_REPO = "Gitea-Tools"
TOKEN = "synthetic-test-token"
def _config(**kwargs) -> bridge.SentryBridgeConfig:
base = dict(
base_url=BASE_URL,
org=SENTRY_ORG,
project=SENTRY_PROJECT,
lookback="24h",
min_events_for_issue=2,
bridge_enabled=True,
)
base.update(kwargs)
return bridge.SentryBridgeConfig(**base)
def _mapping() -> ProjectMapping:
return ProjectMapping(
name="gitea-tools-mcp",
provider="sentry",
monitor_base_url=BASE_URL,
monitor_org=SENTRY_ORG,
monitor_project=SENTRY_PROJECT,
gitea_org=GITEA_ORG,
gitea_repo=GITEA_REPO,
default_labels=("type:bug", "observability", "sentry", "status:ready"),
)
def _raw_issue(issue_id: str = "4001", **kwargs) -> dict:
payload = {
"id": issue_id,
"shortId": "GITEA-TOOLS-1A",
"title": "RuntimeError: lease acquisition failed",
"culprit": "lease_lifecycle in acquire",
"level": "error",
"status": "unresolved",
"count": "7",
"userCount": 1,
"firstSeen": "2026-07-18T04:11:02.000000Z",
"lastSeen": "2026-07-19T22:40:17.000000Z",
"permalink": f"{BASE_URL}/organizations/{SENTRY_ORG}/issues/{issue_id}/",
"metadata": {"type": "RuntimeError", "value": "lease acquisition failed"},
}
payload.update(kwargs)
return payload
def _raw_event(event_id: str = "ev-1", **kwargs) -> dict:
payload = {
"eventID": event_id,
"message": "lease acquisition failed",
"dateCreated": "2026-07-19T22:40:17.000000Z",
"platform": "python",
"environment": "prod",
"release": "1.2.3",
"tags": [{"key": "role", "value": "author"}],
}
payload.update(kwargs)
return payload
class FakeHttp:
"""Routes synthetic Sentry responses and records requested URLs."""
def __init__(self, routes: list[tuple[int, object, dict[str, str]]] | None = None):
# routes: sequential responses for the issues endpoint
self.routes = routes or []
self.calls: list[str] = []
self.headers_seen: list[dict[str, str]] = []
self.issue_page = 0
def __call__(self, url, headers, timeout):
self.calls.append(url)
self.headers_seen.append(dict(headers))
if "/events/" in url:
return 200, json.dumps([_raw_event()]).encode(), {}
if self.routes:
index = min(self.issue_page, len(self.routes) - 1)
self.issue_page += 1
status, payload, resp_headers = self.routes[index]
body = payload if isinstance(payload, bytes) else json.dumps(payload).encode()
return status, body, resp_headers
return 200, json.dumps([_raw_issue()]).encode(), {}
class SentryBridgeTestCase(unittest.TestCase):
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.addCleanup(self._tmp.cleanup)
self.db_path = os.path.join(self._tmp.name, "cp.sqlite3")
self.db = ControlPlaneDB(self.db_path)
self.created: list[dict] = []
self.comments: list[dict] = []
self._next_issue_number = 900
self._next_comment_id = 5000
def _create_issue_fn(self):
def create_fn(title, body, labels, g_org, g_repo):
self._next_issue_number += 1
self.created.append(
{
"title": title,
"body": body,
"labels": list(labels),
"org": g_org,
"repo": g_repo,
"number": self._next_issue_number,
}
)
return {"success": True, "number": self._next_issue_number}
return create_fn
def _comment_issue_fn(self):
def comment_fn(issue_number, body, g_org, g_repo):
self._next_comment_id += 1
self.comments.append(
{
"issue_number": issue_number,
"body": body,
"org": g_org,
"repo": g_repo,
"comment_id": self._next_comment_id,
}
)
return {"success": True, "comment_id": self._next_comment_id}
return comment_fn
def _link(self, issue_id: str = "4001"):
return self.db.get_incident_link_by_provider(
provider="sentry",
provider_issue_id=issue_id,
provider_base_url=BASE_URL,
provider_org=SENTRY_ORG,
provider_project=SENTRY_PROJECT,
)
def _watchdog(self, http, *, apply=True, config=None, **kwargs):
return bridge.watchdog(
self.db,
config or _config(),
token=TOKEN,
apply=apply,
mappings=[_mapping()],
http_fn=http,
create_issue_fn=self._create_issue_fn(),
# Always supplied, including dry runs: the bridge itself must
# withhold the comment when apply=False (AC4 + AC8).
comment_issue_fn=self._comment_issue_fn(),
**kwargs,
)
class TestConfigAndSelfHosted(SentryBridgeTestCase):
def test_self_hosted_base_url_is_used_and_flagged(self):
config = _config()
self.assertTrue(config.as_dict()["self_hosted"])
http = FakeHttp()
bridge.list_issues(config, token=TOKEN, http_fn=http)
self.assertTrue(http.calls[0].startswith(f"{BASE_URL}/api/0/projects/"))
self.assertIn(f"/projects/{SENTRY_ORG}/{SENTRY_PROJECT}/issues/", http.calls[0])
self.assertIn("statsPeriod=24h", http.calls[0])
def test_config_never_exposes_token(self):
config = bridge.load_bridge_config(
{
bridge.ENV_BASE_URL: BASE_URL,
bridge.ENV_ORG: SENTRY_ORG,
bridge.ENV_PROJECT: SENTRY_PROJECT,
bridge.ENV_AUTH_TOKEN: "super-secret-value",
}
)
serialized = json.dumps(config.as_dict())
self.assertNotIn("super-secret-value", serialized)
self.assertNotIn("token", serialized.lower())
def test_invalid_lookback_falls_back_to_default(self):
config = bridge.load_bridge_config({bridge.ENV_LOOKBACK: "not-a-window"})
self.assertEqual(config.lookback, bridge.DEFAULT_LOOKBACK)
class TestMissingTokenAndUnavailable(SentryBridgeTestCase):
def test_missing_token_fails_closed_without_http_call(self):
http = FakeHttp()
with self.assertRaises(bridge.SentryApiError) as ctx:
bridge.list_issues(_config(), token="", http_fn=http)
self.assertEqual(ctx.exception.kind, bridge.ERROR_MISSING_TOKEN)
self.assertEqual(http.calls, [], "no HTTP call may be made without a token")
def test_unconfigured_project_fails_closed(self):
with self.assertRaises(bridge.SentryApiError) as ctx:
bridge.list_issues(_config(project=""), token=TOKEN, http_fn=FakeHttp())
self.assertEqual(ctx.exception.kind, bridge.ERROR_NOT_CONFIGURED)
def test_unauthorized_status_maps_to_missing_token(self):
http = FakeHttp(routes=[(401, {"detail": "Invalid token"}, {})])
with self.assertRaises(bridge.SentryApiError) as ctx:
bridge.list_issues(_config(), token=TOKEN, http_fn=http)
self.assertEqual(ctx.exception.kind, bridge.ERROR_MISSING_TOKEN)
def test_server_error_maps_to_unavailable(self):
http = FakeHttp(routes=[(502, {"detail": "bad gateway"}, {})])
with self.assertRaises(bridge.SentryApiError) as ctx:
bridge.list_issues(_config(), token=TOKEN, http_fn=http)
self.assertEqual(ctx.exception.kind, bridge.ERROR_UNAVAILABLE)
def test_urlerror_from_default_handler_maps_to_unavailable(self):
"""The real urllib handler must translate URLError, not leak it."""
def boom(request, timeout=None):
raise urllib.error.URLError("connection refused")
with unittest.mock.patch.object(urllib.request, "urlopen", boom):
with self.assertRaises(bridge.SentryApiError) as ctx:
bridge._default_http_fn("https://sentry.prgs.cc/api/0/x/", {}, 1.0)
self.assertEqual(ctx.exception.kind, bridge.ERROR_UNAVAILABLE)
def test_watchdog_reports_unavailable_without_mutating(self):
def failing(url, headers, timeout):
raise bridge.SentryApiError(
"Sentry unreachable", kind=bridge.ERROR_UNAVAILABLE
)
result = self._watchdog(failing)
self.assertFalse(result["success"])
self.assertEqual(result["error_kind"], bridge.ERROR_UNAVAILABLE)
self.assertEqual(self.created, [], "no Gitea issue on Sentry outage")
def test_invalid_json_fails_closed(self):
http = FakeHttp(routes=[(200, b"<html>not json</html>", {})])
with self.assertRaises(bridge.SentryApiError) as ctx:
bridge.list_issues(_config(), token=TOKEN, http_fn=http)
self.assertEqual(ctx.exception.kind, bridge.ERROR_INVALID_RESPONSE)
class TestPagination(SentryBridgeTestCase):
def test_link_header_cursor_is_followed(self):
page1 = (
200,
[_raw_issue("4001")],
{
"link": (
f'<{BASE_URL}/api/0/x/?cursor=c1>; rel="previous"; results="false", '
f'<{BASE_URL}/api/0/x/?cursor=c2>; rel="next"; results="true"; cursor="c2"'
)
},
)
page2 = (
200,
[_raw_issue("4002")],
{
"link": (
f'<{BASE_URL}/api/0/x/?cursor=c3>; rel="next"; '
'results="false"; cursor="c3"'
)
},
)
http = FakeHttp(routes=[page1, page2])
result = bridge.list_issues(_config(), token=TOKEN, http_fn=http)
self.assertEqual(result["pages_fetched"], 2)
self.assertEqual([i["id"] for i in result["issues"]], ["4001", "4002"])
self.assertTrue(result["inventory_complete"])
self.assertIn("cursor=c2", http.calls[1])
def test_max_pages_caps_traversal_and_reports_incomplete(self):
page = (
200,
[_raw_issue("4001")],
{"link": f'<{BASE_URL}/x>; rel="next"; results="true"; cursor="cN"'},
)
http = FakeHttp(routes=[page])
result = bridge.list_issues(_config(), token=TOKEN, http_fn=http, max_pages=3)
self.assertEqual(result["pages_fetched"], 3)
self.assertFalse(result["inventory_complete"])
def test_parse_next_cursor_ignores_exhausted_results(self):
self.assertIsNone(
bridge.parse_next_cursor('<u>; rel="next"; results="false"; cursor="c"')
)
self.assertEqual(
bridge.parse_next_cursor('<u>; rel="next"; results="true"; cursor="c9"'),
"c9",
)
self.assertIsNone(bridge.parse_next_cursor(None))
class TestRedaction(SentryBridgeTestCase):
def test_secrets_and_paths_are_scrubbed(self):
raw = _raw_issue(
title="RuntimeError: token=abc123supersecret failed",
culprit="/Users/jasonwalker/Development/Gitea-Tools/lease_lifecycle.py",
metadata={"type": "RuntimeError", "value": "password=hunter2"},
)
sanitized = bridge.sanitize_issue(raw)
blob = json.dumps(sanitized)
self.assertNotIn("abc123supersecret", blob)
self.assertNotIn("hunter2", blob)
self.assertNotIn("/Users/jasonwalker", blob)
self.assertIn("[REDACTED]", sanitized["title"])
def test_permalink_with_embedded_credentials_is_dropped(self):
raw = _raw_issue(permalink="https://user:[email protected]/issues/4001/")
self.assertIsNone(bridge.sanitize_issue(raw)["permalink"])
def test_sensitive_event_tags_are_removed(self):
event = _raw_event(
tags=[
{"key": "authorization", "value": "Bearer abc123secrettoken"},
{"key": "role", "value": "author"},
]
)
sanitized = bridge.sanitize_event(event)
blob = json.dumps(sanitized)
self.assertNotIn("abc123secrettoken", blob)
self.assertEqual(sanitized["tags"].get("role"), "author")
def test_token_never_appears_in_watchdog_output(self):
result = self._watchdog(FakeHttp())
self.assertNotIn(TOKEN, json.dumps(result))
def test_issue_without_id_fails_closed(self):
with self.assertRaises(bridge.SentryApiError) as ctx:
bridge.sanitize_issue({"title": "no id"})
self.assertEqual(ctx.exception.kind, bridge.ERROR_INVALID_RESPONSE)
class TestCreateUpdateDedupe(SentryBridgeTestCase):
def test_dry_run_creates_nothing(self):
result = self._watchdog(FakeHttp(), apply=False)
self.assertTrue(result["success"])
self.assertEqual(result["reconciled"], 1)
self.assertEqual(result["results"][0]["outcome"], OUTCOME_PREVIEW)
self.assertEqual(self.created, [], "dry-run must not create Gitea issues")
def test_apply_creates_one_durable_gitea_issue(self):
result = self._watchdog(FakeHttp())
self.assertTrue(result["success"])
self.assertEqual(result["results"][0]["outcome"], OUTCOME_CREATED)
self.assertEqual(len(self.created), 1)
created = self.created[0]
self.assertEqual(created["org"], GITEA_ORG)
self.assertEqual(created["repo"], GITEA_REPO)
self.assertIn("sentry", created["labels"])
# AC5: body carries the Sentry id and the first-seen window.
self.assertIn("4001", created["body"])
self.assertIn("2026-07-18T04:11:02", created["body"])
def test_repeat_scan_dedupes_to_a_single_issue(self):
first = self._watchdog(FakeHttp())
second = self._watchdog(FakeHttp())
self.assertEqual(first["results"][0]["outcome"], OUTCOME_CREATED)
self.assertEqual(second["results"][0]["outcome"], OUTCOME_UPDATED)
self.assertEqual(len(self.created), 1, "recurrence must not create a duplicate")
def test_recurrence_updates_link_event_count(self):
self._watchdog(FakeHttp())
recurring = FakeHttp(routes=[(200, [_raw_issue("4001", count="42")], {})])
result = self._watchdog(recurring)
self.assertEqual(result["results"][0]["outcome"], OUTCOME_UPDATED)
self.assertEqual(self._link()["event_count"], 42)
def test_recurrence_posts_a_comment_on_the_second_scan(self):
"""AC4: continued Sentry events comment on the linked Gitea issue."""
first = self._watchdog(FakeHttp())
self.assertEqual(first["results"][0]["outcome"], OUTCOME_CREATED)
self.assertEqual(self.comments, [], "creation must not post a recurrence comment")
recurring = FakeHttp(routes=[(200, [_raw_issue("4001", count="42")], {})])
second = self._watchdog(recurring)
self.assertEqual(second["results"][0]["outcome"], OUTCOME_UPDATED)
self.assertEqual(len(self.comments), 1, "recurrence must post exactly one comment")
comment = self.comments[0]
linked_number = int(self._link()["gitea_issue_number"])
self.assertEqual(comment["issue_number"], linked_number)
self.assertEqual(comment["org"], GITEA_ORG)
self.assertEqual(comment["repo"], GITEA_REPO)
# AC5 fields carried on the recurrence record.
self.assertIn("4001", comment["body"])
self.assertIn("42", comment["body"])
self.assertIn("recurrence_basis", comment["body"])
reported = second["results"][0]["recurrence_comment"]
self.assertTrue(reported["posted"])
self.assertEqual(reported["comment_id"], comment["comment_id"])
self.assertEqual(len(self.created), 1, "recurrence must not create a duplicate issue")
def test_dry_run_scan_posts_no_recurrence_comment(self):
"""AC4 + AC8: dry run never comments, even on a linked recurrence."""
self._watchdog(FakeHttp())
recurring = FakeHttp(routes=[(200, [_raw_issue("4001", count="42")], {})])
result = self._watchdog(recurring, apply=False)
self.assertEqual(result["results"][0]["outcome"], OUTCOME_PREVIEW)
self.assertEqual(self.comments, [], "dry-run must not post recurrence comments")
def test_repeat_scan_without_new_events_posts_no_comment(self):
"""A scan that observes no new events must stay silent."""
self._watchdog(FakeHttp())
result = self._watchdog(FakeHttp())
self.assertEqual(result["results"][0]["outcome"], OUTCOME_UPDATED)
self.assertEqual(self.comments, [], "unchanged event state must not comment")
self.assertFalse(result["results"][0]["recurrence_comment"]["posted"])
def test_recurrence_comment_failure_keeps_the_link_durable(self):
"""A failed comment must not roll back or block the incident_links row."""
self._watchdog(FakeHttp())
def failing_comment(issue_number, body, g_org, g_repo):
raise RuntimeError("gitea comment route unavailable")
recurring = FakeHttp(routes=[(200, [_raw_issue("4001", count="42")], {})])
result = bridge.watchdog(
self.db,
_config(),
token=TOKEN,
apply=True,
mappings=[_mapping()],
http_fn=recurring,
create_issue_fn=self._create_issue_fn(),
comment_issue_fn=failing_comment,
)
entry = result["results"][0]
self.assertEqual(entry["outcome"], OUTCOME_UPDATED)
self.assertFalse(entry["recurrence_comment"]["posted"])
self.assertEqual(self._link()["event_count"], 42, "link must still be updated")
def test_recurrence_comment_is_redacted(self):
"""AC2: recurrence comments pass through the same redaction path."""
self._watchdog(FakeHttp())
recurring = FakeHttp(
routes=[
(
200,
[
_raw_issue(
"4001",
count="42",
metadata={
"type": "RuntimeError",
"value": "token=abc123supersecret",
},
)
],
{},
)
]
)
self._watchdog(recurring)
self.assertEqual(len(self.comments), 1)
body = self.comments[0]["body"]
self.assertNotIn("abc123supersecret", body)
self.assertNotIn(TOKEN, body)
def test_link_survives_a_new_db_handle(self):
"""AC6: bridge mapping survives process restarts."""
self._watchdog(FakeHttp())
reopened = ControlPlaneDB(self.db_path)
link = reopened.get_incident_link_by_provider(
provider="sentry",
provider_issue_id="4001",
provider_base_url=BASE_URL,
provider_org=SENTRY_ORG,
provider_project=SENTRY_PROJECT,
)
self.assertIsNotNone(link)
self.assertEqual(int(link["gitea_issue_number"]), 901)
def test_resolved_issue_is_not_recreated_or_reopened(self):
"""AC7: a resolved Sentry issue never creates or reopens Gitea work."""
self._watchdog(FakeHttp())
linked_number = int(self._link()["gitea_issue_number"])
closed = FakeHttp(routes=[(200, [_raw_issue("4001", status="resolved")], {})])
result = self._watchdog(closed)
self.assertEqual(result["skipped"], 1)
self.assertEqual(result["results"][0]["action"], bridge.ACTION_SKIPPED_STATUS)
self.assertEqual(len(self.created), 1)
self.assertEqual(linked_number, 901)
class TestPolicyGates(SentryBridgeTestCase):
def test_below_threshold_issue_is_skipped(self):
http = FakeHttp(routes=[(200, [_raw_issue("4001", count="1")], {})])
result = self._watchdog(http)
self.assertEqual(result["skipped"], 1)
self.assertEqual(result["results"][0]["action"], bridge.ACTION_SKIPPED_THRESHOLD)
self.assertEqual(self.created, [])
def test_apply_refused_when_bridge_disabled(self):
result = self._watchdog(FakeHttp(), config=_config(bridge_enabled=False))
self.assertFalse(result["success"])
self.assertEqual(result["error_kind"], bridge.ERROR_BRIDGE_DISABLED)
self.assertEqual(self.created, [])
def test_dry_run_allowed_while_bridge_disabled(self):
result = self._watchdog(
FakeHttp(), apply=False, config=_config(bridge_enabled=False)
)
self.assertTrue(result["success"])
def test_raw_incident_is_never_assignable_work(self):
result = self._watchdog(FakeHttp())
self.assertFalse(result["raw_incident_assignable"])
self.assertEqual(result["durable_work_system"], "gitea_issues")
def test_reconcile_failure_is_isolated_and_redacted(self):
def exploding(db, **kwargs):
raise RuntimeError("token=abc123 boom")
result = self._watchdog(FakeHttp(), reconcile_fn=exploding)
self.assertFalse(result["success"])
self.assertEqual(result["failed"], 1)
self.assertNotIn("abc123", json.dumps(result))
class TestObservationMapping(SentryBridgeTestCase):
def test_observation_carries_provider_identity_and_targets(self):
issue = bridge.sanitize_issue(_raw_issue())
event = bridge.sanitize_event(_raw_event())
obs = bridge.observation_from_issue(
issue,
_config(),
gitea_org=GITEA_ORG,
gitea_repo=GITEA_REPO,
latest_event=event,
)
self.assertEqual(obs["provider"], "sentry")
self.assertEqual(obs["provider_base_url"], BASE_URL)
self.assertEqual(obs["provider_issue_id"], "4001")
self.assertEqual(obs["event_count"], 7)
self.assertEqual(obs["environment"], "prod")
self.assertEqual(obs["gitea_repo"], GITEA_REPO)
self.assertTrue(_mapping().matches_observation(obs))
def test_events_fetch_returns_latest_first(self):
result = bridge.get_issue_events(
_config(), "4001", token=TOKEN, http_fn=FakeHttp()
)
self.assertEqual(result["count"], 1)
self.assertEqual(result["latest_event"]["environment"], "prod")
class TestMcpToolWrappers(unittest.TestCase):
"""The registered MCP tools must fail closed, never raise, never leak."""
def setUp(self):
# Read-capable profile, fully configured Sentry target, but
# deliberately no SENTRY_AUTH_TOKEN — the token gap is the only fault.
self.env = shared_mutation_env(
"test-author-prgs",
**{
bridge.ENV_BASE_URL: BASE_URL,
bridge.ENV_ORG: SENTRY_ORG,
bridge.ENV_PROJECT: SENTRY_PROJECT,
},
)
self.env.pop(bridge.ENV_AUTH_TOKEN, None)
def _server(self):
import gitea_mcp_server
return gitea_mcp_server
def test_all_five_tools_are_registered(self):
import asyncio
tools = asyncio.run(self._server().mcp.list_tools())
registered = {t.name for t in tools if t.name.startswith("gitea_sentry_")}
self.assertEqual(
registered,
{
"gitea_sentry_list_issues",
"gitea_sentry_get_issue_events",
"gitea_sentry_reconcile_issue",
"gitea_sentry_link_gitea_issue",
"gitea_sentry_watchdog",
},
)
def test_list_issues_without_token_fails_closed(self):
with unittest.mock.patch.dict(os.environ, self.env, clear=True):
result = self._server().gitea_sentry_list_issues()
self.assertFalse(result["success"])
self.assertEqual(result.get("error_kind"), bridge.ERROR_MISSING_TOKEN)
self.assertEqual(result["issues"], [])
def test_get_issue_events_without_token_fails_closed(self):
with unittest.mock.patch.dict(os.environ, self.env, clear=True):
result = self._server().gitea_sentry_get_issue_events("4001")
self.assertFalse(result["success"])
self.assertEqual(result.get("error_kind"), bridge.ERROR_MISSING_TOKEN)
def test_reconcile_without_token_fails_closed_without_mutation(self):
with unittest.mock.patch.dict(os.environ, self.env, clear=True):
result = self._server().gitea_sentry_reconcile_issue("4001", apply=True)
self.assertFalse(result["success"])
self.assertFalse(result["raw_incident_assignable"])
self.assertEqual(result.get("error_kind"), bridge.ERROR_MISSING_TOKEN)
def test_watchdog_without_token_fails_closed(self):
with unittest.mock.patch.dict(os.environ, self.env, clear=True):
result = self._server().gitea_sentry_watchdog()
self.assertFalse(result["success"])
self.assertEqual(result.get("error_kind"), bridge.ERROR_MISSING_TOKEN)
def test_unconfigured_target_reports_not_configured_before_token(self):
env = {k: v for k, v in self.env.items() if not k.startswith("SENTRY_")}
with unittest.mock.patch.dict(os.environ, env, clear=True):
result = self._server().gitea_sentry_list_issues()
self.assertFalse(result["success"])
self.assertEqual(result.get("error_kind"), bridge.ERROR_NOT_CONFIGURED)
def test_tool_output_never_contains_a_token_value(self):
env = dict(self.env)
env[bridge.ENV_AUTH_TOKEN] = "leaky-token-value"
with unittest.mock.patch.dict(os.environ, env, clear=True):
result = self._server().gitea_sentry_list_issues()
self.assertNotIn("leaky-token-value", json.dumps(result))
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink