sysadmin
6a179aeb85
Add phase-1 observability bridge that turns provider observations into normal Gitea issues and control-plane incident_links rows. Dry-run is default; apply creates/links through sanctioned Gitea issue paths only. Raw incidents remain non-assignable; the #600 allocator sees bridge work only as ordinary Gitea issues. Builds on #613 substrate and #600 allocator. Closes #612
346 lines
12 KiB
Python
346 lines
12 KiB
Python
"""Tests for observability incident bridge (#612) on #613 substrate."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
|
|
from control_plane_db import ControlPlaneDB, InvalidWorkKindError, WORK_KINDS
|
|
from incident_bridge import (
|
|
OUTCOME_BLOCKED,
|
|
OUTCOME_CREATED,
|
|
OUTCOME_LINKED,
|
|
OUTCOME_PREVIEW,
|
|
OUTCOME_UPDATED,
|
|
ProjectMapping,
|
|
assert_not_raw_incident_work_item,
|
|
build_gitea_issue_body,
|
|
load_project_mappings,
|
|
normalize_incident,
|
|
project_mapping_from_dict,
|
|
reconcile_incident,
|
|
redact_text,
|
|
sanitize_tags,
|
|
)
|
|
|
|
|
|
def _mapping(**kwargs) -> ProjectMapping:
|
|
base = dict(
|
|
name="gitea-tools-mcp",
|
|
provider="sentry",
|
|
monitor_base_url="https://sentry.prgs.cc",
|
|
monitor_org="prgs",
|
|
monitor_project="gitea-tools-mcp",
|
|
gitea_org="Scaled-Tech-Consulting",
|
|
gitea_repo="Gitea-Tools",
|
|
default_labels=("type:bug", "observability", "sentry", "status:ready"),
|
|
)
|
|
base.update(kwargs)
|
|
return ProjectMapping(**base)
|
|
|
|
|
|
def _obs(**kwargs) -> dict:
|
|
base = dict(
|
|
provider="sentry",
|
|
provider_base_url="https://sentry.prgs.cc",
|
|
provider_org="prgs",
|
|
provider_project="gitea-tools-mcp",
|
|
provider_issue_id="ISSUE-100",
|
|
provider_short_id="GITEA-TOOLS-1A",
|
|
provider_permalink="https://sentry.prgs.cc/organizations/prgs/issues/100/",
|
|
fingerprint="fp-abc",
|
|
title="TypeError: boom",
|
|
summary="TypeError: boom in handle_request",
|
|
first_seen="2026-07-01T00:00:00Z",
|
|
last_seen="2026-07-10T00:00:00Z",
|
|
event_count=3,
|
|
environment="production",
|
|
severity="error",
|
|
culprit="handle_request",
|
|
tags={"runtime": "python", "release": "1.2.3"},
|
|
)
|
|
base.update(kwargs)
|
|
return base
|
|
|
|
|
|
class IncidentBridgeTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3"))
|
|
self.mapping = _mapping()
|
|
self.created: list[dict] = []
|
|
|
|
def tearDown(self) -> None:
|
|
self._tmp.cleanup()
|
|
|
|
def _create_issue(self, title, body, labels, g_org, g_repo):
|
|
n = 9000 + len(self.created)
|
|
rec = {
|
|
"number": n,
|
|
"success": True,
|
|
"performed": True,
|
|
"title": title,
|
|
"body": body,
|
|
"labels": labels,
|
|
"org": g_org,
|
|
"repo": g_repo,
|
|
}
|
|
self.created.append(rec)
|
|
return rec
|
|
|
|
def test_redact_secrets(self) -> None:
|
|
dirty = "token=supersecret123 Authorization: Bearer abcdefghijklmnop"
|
|
clean = redact_text(dirty)
|
|
self.assertNotIn("supersecret", clean)
|
|
self.assertNotIn("abcdefghijklmnop", clean)
|
|
self.assertIn("[REDACTED]", clean)
|
|
tags = sanitize_tags(
|
|
{"token": "x", "runtime": "py", "password": "nope", "release": "1.0"}
|
|
)
|
|
self.assertEqual(tags, {"runtime": "py", "release": "1.0"})
|
|
|
|
def test_body_contains_no_secrets(self) -> None:
|
|
inc = normalize_incident(
|
|
_obs(
|
|
summary="fail password=hunter2 token: abc",
|
|
tags={"cookie": "sid=1", "runtime": "py"},
|
|
),
|
|
self.mapping,
|
|
)
|
|
body = build_gitea_issue_body(inc)
|
|
self.assertNotIn("hunter2", body)
|
|
self.assertNotIn("sid=1", body)
|
|
self.assertIn("provider_issue_id", body)
|
|
self.assertIn("not** assignable", body.lower())
|
|
|
|
def test_preview_no_mutation(self) -> None:
|
|
res = reconcile_incident(
|
|
self.db,
|
|
observation=_obs(),
|
|
mapping=self.mapping,
|
|
apply=False,
|
|
create_issue_fn=self._create_issue,
|
|
)
|
|
self.assertTrue(res["success"])
|
|
self.assertEqual(res["outcome"], OUTCOME_PREVIEW)
|
|
self.assertFalse(res["gitea_mutated"])
|
|
self.assertFalse(res["db_mutated"])
|
|
self.assertFalse(res["raw_incident_assignable"])
|
|
self.assertEqual(len(self.created), 0)
|
|
self.assertIsNone(
|
|
self.db.get_incident_link_by_provider(
|
|
provider="sentry",
|
|
provider_issue_id="ISSUE-100",
|
|
provider_base_url="https://sentry.prgs.cc",
|
|
provider_org="prgs",
|
|
provider_project="gitea-tools-mcp",
|
|
)
|
|
)
|
|
|
|
def test_apply_creates_issue_and_link(self) -> None:
|
|
res = reconcile_incident(
|
|
self.db,
|
|
observation=_obs(),
|
|
mapping=self.mapping,
|
|
apply=True,
|
|
create_issue_fn=self._create_issue,
|
|
)
|
|
self.assertTrue(res["success"])
|
|
self.assertEqual(res["outcome"], OUTCOME_CREATED)
|
|
self.assertTrue(res["gitea_mutated"])
|
|
self.assertTrue(res["db_mutated"])
|
|
self.assertEqual(len(self.created), 1)
|
|
self.assertEqual(res["gitea_issue"]["number"], self.created[0]["number"])
|
|
link = self.db.get_incident_link_by_provider(
|
|
provider="sentry",
|
|
provider_issue_id="ISSUE-100",
|
|
provider_base_url="https://sentry.prgs.cc",
|
|
provider_org="prgs",
|
|
provider_project="gitea-tools-mcp",
|
|
)
|
|
self.assertIsNotNone(link)
|
|
self.assertEqual(int(link["gitea_issue_number"]), self.created[0]["number"])
|
|
self.assertEqual(res["allocator_visible_as"]["kind"], "issue")
|
|
|
|
def test_duplicate_observation_reuses_issue(self) -> None:
|
|
first = reconcile_incident(
|
|
self.db,
|
|
observation=_obs(),
|
|
mapping=self.mapping,
|
|
apply=True,
|
|
create_issue_fn=self._create_issue,
|
|
)
|
|
second = reconcile_incident(
|
|
self.db,
|
|
observation=_obs(event_count=9, last_seen="2026-07-11T00:00:00Z"),
|
|
mapping=self.mapping,
|
|
apply=True,
|
|
create_issue_fn=self._create_issue,
|
|
)
|
|
self.assertEqual(first["outcome"], OUTCOME_CREATED)
|
|
self.assertEqual(second["outcome"], OUTCOME_UPDATED)
|
|
self.assertEqual(len(self.created), 1)
|
|
self.assertEqual(
|
|
second["gitea_issue"]["number"], first["gitea_issue"]["number"]
|
|
)
|
|
link = self.db.get_incident_link_by_provider(
|
|
provider="sentry",
|
|
provider_issue_id="ISSUE-100",
|
|
provider_base_url="https://sentry.prgs.cc",
|
|
provider_org="prgs",
|
|
provider_project="gitea-tools-mcp",
|
|
)
|
|
self.assertEqual(int(link["event_count"]), 9)
|
|
|
|
def test_explicit_link_existing_issue(self) -> None:
|
|
res = reconcile_incident(
|
|
self.db,
|
|
observation=_obs(provider_issue_id="ISSUE-200"),
|
|
mapping=self.mapping,
|
|
apply=True,
|
|
force_gitea_issue_number=555,
|
|
create_issue_fn=self._create_issue,
|
|
)
|
|
self.assertEqual(res["outcome"], OUTCOME_LINKED)
|
|
self.assertEqual(len(self.created), 0)
|
|
self.assertEqual(res["gitea_issue"]["number"], 555)
|
|
link = self.db.get_incident_link_for_gitea_issue(
|
|
gitea_org="Scaled-Tech-Consulting",
|
|
gitea_repo="Gitea-Tools",
|
|
gitea_issue_number=555,
|
|
)
|
|
self.assertIsNotNone(link)
|
|
|
|
def test_fingerprint_conflict_fails_closed(self) -> None:
|
|
reconcile_incident(
|
|
self.db,
|
|
observation=_obs(fingerprint="fp-a"),
|
|
mapping=self.mapping,
|
|
apply=True,
|
|
create_issue_fn=self._create_issue,
|
|
)
|
|
res = reconcile_incident(
|
|
self.db,
|
|
observation=_obs(fingerprint="fp-b"),
|
|
mapping=self.mapping,
|
|
apply=True,
|
|
create_issue_fn=self._create_issue,
|
|
)
|
|
self.assertEqual(res["outcome"], OUTCOME_BLOCKED)
|
|
self.assertIn("fingerprint conflict", res["reasons"][0].lower())
|
|
self.assertEqual(len(self.created), 1)
|
|
|
|
def test_ambiguous_mapping_fails_closed(self) -> None:
|
|
maps = [
|
|
_mapping(name="a", monitor_project="gitea-tools-mcp"),
|
|
_mapping(name="b", monitor_project="gitea-tools-mcp"),
|
|
]
|
|
res = reconcile_incident(
|
|
self.db,
|
|
observation=_obs(),
|
|
mappings=maps,
|
|
apply=False,
|
|
)
|
|
self.assertEqual(res["outcome"], OUTCOME_BLOCKED)
|
|
self.assertIn("ambiguous", res["reasons"][0].lower())
|
|
|
|
def test_missing_provider_issue_id_fails_closed(self) -> None:
|
|
res = reconcile_incident(
|
|
self.db,
|
|
observation=_obs(provider_issue_id=""),
|
|
mapping=self.mapping,
|
|
apply=False,
|
|
)
|
|
self.assertEqual(res["outcome"], OUTCOME_BLOCKED)
|
|
|
|
def test_raw_incident_not_work_kind(self) -> None:
|
|
self.assertNotIn("sentry_incident", WORK_KINDS)
|
|
with self.assertRaises(InvalidWorkKindError):
|
|
self.db.upsert_work_item(
|
|
remote="prgs",
|
|
org="o",
|
|
repo="r",
|
|
kind="sentry_incident",
|
|
number=1,
|
|
)
|
|
with self.assertRaises(Exception):
|
|
assert_not_raw_incident_work_item("glitchtip_incident")
|
|
|
|
def test_db_unavailable(self) -> None:
|
|
res = reconcile_incident(
|
|
None,
|
|
observation=_obs(),
|
|
mapping=self.mapping,
|
|
apply=True,
|
|
create_issue_fn=self._create_issue,
|
|
)
|
|
self.assertFalse(res["success"])
|
|
self.assertEqual(res["outcome"], OUTCOME_BLOCKED)
|
|
|
|
def test_structured_skip_reason_environment(self) -> None:
|
|
m = _mapping(environment_filters=("staging",))
|
|
res = reconcile_incident(
|
|
self.db,
|
|
observation=_obs(environment="production"),
|
|
mapping=m,
|
|
apply=True,
|
|
create_issue_fn=self._create_issue,
|
|
)
|
|
self.assertTrue(res["success"])
|
|
self.assertEqual(res["action"], "skip_environment_filter")
|
|
self.assertEqual(len(self.created), 0)
|
|
|
|
def test_load_mappings_json(self) -> None:
|
|
payload = json.dumps(
|
|
{
|
|
"projects": [
|
|
{
|
|
"name": "gt",
|
|
"provider": "glitchtip",
|
|
"monitor_base_url": "https://glitchtip.example",
|
|
"monitor_org": "org",
|
|
"monitor_project": "proj",
|
|
"gitea_org": "Scaled-Tech-Consulting",
|
|
"gitea_repo": "Gitea-Tools",
|
|
}
|
|
]
|
|
}
|
|
)
|
|
maps = load_project_mappings(mappings_json=payload)
|
|
self.assertEqual(len(maps), 1)
|
|
self.assertEqual(maps[0].provider, "glitchtip")
|
|
|
|
def test_glitchtip_provider_accepted(self) -> None:
|
|
m = _mapping(provider="glitchtip", monitor_base_url="https://glitchtip.example")
|
|
res = reconcile_incident(
|
|
self.db,
|
|
observation=_obs(
|
|
provider="glitchtip",
|
|
provider_base_url="https://glitchtip.example",
|
|
),
|
|
mapping=m,
|
|
apply=True,
|
|
create_issue_fn=self._create_issue,
|
|
)
|
|
self.assertEqual(res["outcome"], OUTCOME_CREATED)
|
|
self.assertEqual(res["incident"]["provider"], "glitchtip")
|
|
|
|
def test_allocator_only_sees_issue_kind(self) -> None:
|
|
res = reconcile_incident(
|
|
self.db,
|
|
observation=_obs(),
|
|
mapping=self.mapping,
|
|
apply=True,
|
|
create_issue_fn=self._create_issue,
|
|
)
|
|
vis = res["allocator_visible_as"]
|
|
self.assertEqual(vis["kind"], "issue")
|
|
self.assertIn(vis["kind"], WORK_KINDS)
|
|
self.assertFalse(res["raw_incident_assignable"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|