"""Tests for optional self-hosted Sentry observability (#606). Covers the pure module (config, redaction, event/check-in builders) and the runtime capture paths using a fake ``sentry_sdk``, so nothing ever touches the network. Critically proves the feature is a no-op when disabled or DSN-less (acceptance criterion 1) and that secrets/paths/session-state are never sent (criterion 5). """ import sys import contextlib from pathlib import Path import pytest sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) import sentry_observability as so # noqa: E402 # ── Fake SDK ──────────────────────────────────────────────────────────────── class _FakeScope: def __init__(self): self.tags = {} self.extras = {} def set_tag(self, k, v): self.tags[k] = v def set_extra(self, k, v): self.extras[k] = v class FakeSentrySDK: """Minimal stand-in exposing the SDK surface sentry_observability uses.""" def __init__(self): self.init_kwargs = None self.events = [] self.exceptions = [] self.checkins = [] self.last_scope = None def init(self, **kwargs): self.init_kwargs = kwargs def capture_event(self, event): self.events.append(event) def capture_exception(self, exc): self.exceptions.append(exc) def capture_checkin(self, **payload): self.checkins.append(payload) @contextlib.contextmanager def push_scope(self): self.last_scope = _FakeScope() yield self.last_scope @pytest.fixture(autouse=True) def _reset_state(): """Isolate module init state between tests.""" so.reset_for_tests() yield so.reset_for_tests() @pytest.fixture def fake_sdk(monkeypatch): sdk = FakeSentrySDK() monkeypatch.setattr(so, "sentry_sdk", sdk) return sdk # ── Config / gating ───────────────────────────────────────────────────────── def test_disabled_by_default_empty_env(): cfg = so.load_config(env={}) assert cfg.enabled is False assert cfg.active is False def test_enabled_flag_without_dsn_is_not_active(): cfg = so.load_config(env={"MCP_SENTRY_ENABLED": "1"}) assert cfg.enabled is True assert cfg.dsn is None assert cfg.active is False # DSN required def test_dsn_without_enabled_flag_is_not_active(): cfg = so.load_config(env={"SENTRY_DSN": "https://k@sentry.prgs.cc/1"}) assert cfg.active is False def test_active_requires_enabled_and_dsn(): cfg = so.load_config( env={"MCP_SENTRY_ENABLED": "true", "SENTRY_DSN": "https://k@sentry.prgs.cc/1"} ) assert cfg.active is True def test_truthy_variants(): for val in ("1", "true", "YES", "On"): cfg = so.load_config(env={"MCP_SENTRY_ENABLED": val}) assert cfg.enabled is True for val in ("0", "false", "no", "", "off"): cfg = so.load_config(env={"MCP_SENTRY_ENABLED": val}) assert cfg.enabled is False def test_traces_sample_rate_parsed_and_clamped(): assert so.load_config(env={"MCP_SENTRY_TRACES_SAMPLE_RATE": "0.25"}).traces_sample_rate == 0.25 assert so.load_config(env={"MCP_SENTRY_TRACES_SAMPLE_RATE": "5"}).traces_sample_rate == 1.0 assert so.load_config(env={"MCP_SENTRY_TRACES_SAMPLE_RATE": "-1"}).traces_sample_rate == 0.0 assert so.load_config(env={"MCP_SENTRY_TRACES_SAMPLE_RATE": "junk"}).traces_sample_rate == 0.0 def test_safe_summary_has_no_dsn_value(): cfg = so.load_config( env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://secret@sentry.prgs.cc/1"} ) summary = cfg.safe_summary() assert summary["dsn_present"] is True assert "secret" not in repr(summary) assert "dsn" not in summary # only presence, never the value # ── init_sentry ───────────────────────────────────────────────────────────── def test_init_noop_when_disabled(fake_sdk): status = so.init_sentry(so.load_config(env={})) assert status["initialized"] is False assert fake_sdk.init_kwargs is None # no SDK init assert so.is_initialized() is False def test_init_noop_when_enabled_but_missing_dsn(fake_sdk): status = so.init_sentry(so.load_config(env={"MCP_SENTRY_ENABLED": "1"})) assert status["initialized"] is False assert fake_sdk.init_kwargs is None def test_init_reports_missing_sdk(monkeypatch): monkeypatch.setattr(so, "sentry_sdk", None) status = so.init_sentry( so.load_config( env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://k@sentry.prgs.cc/1"} ) ) assert status["initialized"] is False assert "not installed" in status["reason"] def test_init_configures_sdk_with_scrubber(fake_sdk): status = so.init_sentry( so.load_config( env={ "MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://k@sentry.prgs.cc/1", "SENTRY_ENVIRONMENT": "prod", "MCP_SENTRY_TRACES_SAMPLE_RATE": "0.1", } ) ) assert status["initialized"] is True assert so.is_initialized() is True kw = fake_sdk.init_kwargs assert kw["dsn"] == "https://k@sentry.prgs.cc/1" assert kw["environment"] == "prod" assert kw["traces_sample_rate"] == 0.1 assert kw["before_send"] is so.scrub_event assert kw["send_default_pii"] is False def test_init_enable_logs_wires_log_scrubber(fake_sdk): so.init_sentry( so.load_config( env={ "MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://k@sentry.prgs.cc/1", "MCP_SENTRY_ENABLE_LOGS": "1", } ) ) exp = fake_sdk.init_kwargs["_experiments"] assert exp["enable_logs"] is True assert exp["before_send_log"] is so.scrub_event def test_init_never_raises_on_sdk_failure(monkeypatch): class Boom: def init(self, **kwargs): raise RuntimeError("sentry down") monkeypatch.setattr(so, "sentry_sdk", Boom()) status = so.init_sentry( so.load_config( env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://k@sentry.prgs.cc/1"} ) ) assert status["initialized"] is False assert "init failed" in status["reason"] # ── Redaction (fail closed) ───────────────────────────────────────────────── def test_build_tags_allowlist_only(): tags = so.build_tags(role="author", secret_thing="leak", pid=123) assert tags["role"] == "author" assert tags["pid"] == "123" assert "secret_thing" not in tags def test_build_tags_hashes_session_id(): tags = so.build_tags(session_id="prgs-author-20479-cf9ac178") assert "session_id" not in tags assert "session_id_hash" in tags assert tags["session_id_hash"] != "prgs-author-20479-cf9ac178" assert len(tags["session_id_hash"]) == 12 def test_build_tags_collapses_worktree_path(): tags = so.build_tags( worktree_path="/Users/x/Development/Gitea-Tools/branches/issue-606-sentry-observability" ) assert "worktree_path" not in tags assert tags["worktree_category"] == "author" def test_build_tags_drops_value_that_scrubs_to_redacted(): # A tag value that is itself a token gets scrubbed then dropped. tags = so.build_tags(capability="token abcdef1234567890") assert "capability" not in tags def test_sanitize_path_categories(): assert so.sanitize_path("/repo/branches/review-pr-654") == "reviewer" assert so.sanitize_path("/repo/branches/merge-pr-1") == "merger" assert so.sanitize_path("/repo/branches/reconcile-pr-1") == "reconciler" assert so.sanitize_path("/repo/branches/feat-issue-606") == "author" assert so.sanitize_path("/x/y/Gitea-Tools") == "root" def test_redact_value_scrubs_secrets_and_paths(): out = so.redact_value( { "token": "abc123", "note": "Authorization: Bearer sk_live_abcdefgh12345", "path": "/Users/jasonwalker/Development/Gitea-Tools/secret", "dsn": "https://k@sentry.prgs.cc/1", "safe": "hello", } ) assert out["token"] == so.REDACTED assert out["dsn"] == so.REDACTED assert "sk_live" not in out["note"] assert so.REDACTED_PATH in out["path"] assert "/Users/" not in out["path"] assert out["safe"] == "hello" def test_redact_value_forbidden_prompt_and_session_state(): out = so.redact_value( {"prompt": "full body", "session_state": "{...}", "keep": "ok"} ) assert out["prompt"] == so.REDACTED assert out["session_state"] == so.REDACTED assert out["keep"] == "ok" def test_scrub_event_redacts_nested_and_drops_server_name(): event = { "server_name": "some-host", "message": "boom", "extra": {"token": "leak", "ok": "1"}, } scrubbed = so.scrub_event(event) assert "server_name" not in scrubbed assert scrubbed["extra"]["token"] == so.REDACTED assert scrubbed["extra"]["ok"] == "1" def test_scrub_event_drops_non_dict(): assert so.scrub_event("not a dict") is None assert so.scrub_event(None) is None # ── Event builders ────────────────────────────────────────────────────────── def test_build_blocker_event_structure_and_next_action(): event = so.build_blocker_event( "active_foreign_lease", message="blocked by foreign lease", next_action="wait or adopt via allocator", level="warning", tags={"pr_number": 606, "session_id": "s-123"}, ) assert event["tags"]["blocker_type"] == "active_foreign_lease" assert event["tags"]["pr_number"] == "606" assert "session_id" not in event["tags"] assert event["tags"]["session_id_hash"] assert event["extra"]["canonical_next_action"] == "wait or adopt via allocator" assert event["fingerprint"] == ["workflow-blocker", "active_foreign_lease"] def test_build_blocker_event_invalid_level_defaults_warning(): event = so.build_blocker_event("x", level="nonsense") assert event["level"] == "warning" def test_build_checkin_payload_maps_slugs(): for key, slug in so.MONITOR_SLUGS.items(): payload = so.build_checkin_payload(key, "ok") assert payload["monitor_slug"] == slug assert payload["status"] == "ok" def test_build_checkin_payload_explicit_slug_passthrough(): payload = so.build_checkin_payload("custom-slug", "in_progress", duration=1.5) assert payload["monitor_slug"] == "custom-slug" assert payload["duration"] == 1.5 def test_build_checkin_payload_rejects_bad_status(): with pytest.raises(ValueError): so.build_checkin_payload("allocator_health", "bogus") def test_all_six_monitors_registered(): assert set(so.MONITOR_SLUGS) == { "stale_lease_scan", "terminal_lock_scan", "allocator_health", "namespace_health", "dashboard_freshness", "reconciler_cleanup", } # ── Capture paths (fail open) ─────────────────────────────────────────────── def test_capture_workflow_blocker_noop_when_disabled(fake_sdk): # not initialised event = so.capture_workflow_blocker("some_blocker", message="x") assert isinstance(event, dict) # still returns redacted event assert fake_sdk.events == [] # but nothing sent def test_capture_workflow_blocker_sends_when_initialised(fake_sdk): so.init_sentry( so.load_config( env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://k@sentry.prgs.cc/1"} ) ) so.capture_workflow_blocker("terminal_lock_occupied", message="held") assert len(fake_sdk.events) == 1 assert fake_sdk.events[0]["tags"]["blocker_type"] == "terminal_lock_occupied" def test_capture_exception_noop_when_disabled(fake_sdk): assert so.capture_exception(ValueError("x")) is False assert fake_sdk.exceptions == [] def test_capture_exception_sends_scrubbed_tags(fake_sdk): so.init_sentry( so.load_config( env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://k@sentry.prgs.cc/1"} ) ) ok = so.capture_exception( RuntimeError("bad"), tags={"mutation_tool": "gitea_merge_pr", "leaky": "x"} ) assert ok is True assert len(fake_sdk.exceptions) == 1 assert fake_sdk.last_scope.tags["mutation_tool"] == "gitea_merge_pr" assert "leaky" not in fake_sdk.last_scope.tags def test_capture_exception_never_raises(monkeypatch, fake_sdk): so.init_sentry( so.load_config( env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://k@sentry.prgs.cc/1"} ) ) def boom(exc): raise RuntimeError("sdk exploded") monkeypatch.setattr(fake_sdk, "capture_exception", boom) # Must swallow the SDK failure (fail open). assert so.capture_exception(ValueError("y")) is False def test_monitor_checkin_noop_when_disabled(fake_sdk): payload = so.monitor_checkin("allocator_health", "ok") assert payload["monitor_slug"] == "gitea-mcp-allocator-health" assert fake_sdk.checkins == [] # not sent while disabled def test_monitor_checkin_sends_when_initialised(fake_sdk): so.init_sentry( so.load_config( env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://k@sentry.prgs.cc/1"} ) ) so.monitor_checkin("stale_lease_scan", "ok") assert fake_sdk.checkins == [ {"monitor_slug": "gitea-mcp-stale-lease-scan", "status": "ok"} ] def test_monitor_checkin_invalid_status_returns_none(fake_sdk): assert so.monitor_checkin("allocator_health", "bogus") is None assert fake_sdk.checkins == []