afterhours-shift-manager/tests/shared/test_sentry_init.py

"""sentry_init: DSN no-op, init options, and before_send scrub."""

import importlib
import sys
from types import ModuleType
from unittest.mock import patch

from sentry_sdk.integrations.aws_lambda import AwsLambdaIntegration

import shared.sentry_init as sentry_mod


def _reexec(monkeypatch, dsn=None):
    if dsn is None:
        monkeypatch.delenv("SENTRY_DSN", raising=False)
    else:
        monkeypatch.setenv("SENTRY_DSN", dsn)
    with patch("sentry_sdk.init") as mocked:
        importlib.reload(sentry_mod)
        return mocked


def test_unset_dsn_does_not_init(monkeypatch):
    mocked = _reexec(monkeypatch, dsn=None)
    mocked.assert_not_called()


def test_empty_dsn_does_not_init(monkeypatch):
    mocked = _reexec(monkeypatch, dsn="")
    mocked.assert_not_called()


def test_set_dsn_inits_lambda_integration(monkeypatch):
    monkeypatch.setenv("AWS_LAMBDA_FUNCTION_NAME", "afterhours-shift-manager")
    mocked = _reexec(monkeypatch, dsn="https://key@o1.ingest.sentry.io/1")
    mocked.assert_called_once()
    kwargs = mocked.call_args.kwargs
    assert kwargs["dsn"] == "https://key@o1.ingest.sentry.io/1"
    assert kwargs["send_default_pii"] is False
    assert kwargs["include_local_variables"] is False
    assert kwargs["enable_logs"] is False
    assert kwargs["traces_sample_rate"] == 0.0
    assert kwargs["before_send"] is sentry_mod._before_send
    integrations = kwargs["integrations"]
    assert len(integrations) == 1
    assert isinstance(integrations[0], AwsLambdaIntegration)
    assert integrations[0].timeout_warning is True
    assert "release" not in kwargs


def test_build_info_sha_sets_sentry_release(monkeypatch):
    monkeypatch.setenv("AWS_LAMBDA_FUNCTION_NAME", "afterhours-shift-manager")
    monkeypatch.setenv("SENTRY_DSN", "https://key@o1.ingest.sentry.io/1")
    fake = ModuleType("shared.build_info")
    fake.GIT_SHA = "abc123def"
    monkeypatch.setitem(sys.modules, "shared.build_info", fake)
    with patch("sentry_sdk.init") as mocked:
        importlib.reload(sentry_mod)
        kwargs = mocked.call_args.kwargs
    assert kwargs["release"] == "abc123def"


def test_before_send_strips_auth_and_sigv4_headers():
    event = {
        "request": {
            "headers": {
                "Authorization": "Bearer secret",
                "X-Auth-Token": "tok",
                "X-Amz-Date": "20260101T000000Z",
                "Content-Type": "application/json",
            },
            "url": "https://example.invalid/oncall",
        }
    }
    out = sentry_mod._before_send(event, {})
    assert out["request"]["headers"] == {"Content-Type": "application/json"}
    assert out["request"]["url"] == "https://example.invalid/oncall"


def test_before_send_strips_slack_signature_header():
    event = {
        "request": {
            "headers": {
                "X-Slack-Signature": "v0=abc",
                "X-Slack-Request-Timestamp": "123",
                "Content-Type": "application/json",
            }
        }
    }
    out = sentry_mod._before_send(event, {})
    assert out["request"]["headers"] == {
        "X-Slack-Request-Timestamp": "123",
        "Content-Type": "application/json",
    }


def test_before_send_strips_list_headers():
    event = {
        "request": {
            "headers": [
                ("Authorization", "Bearer secret"),
                ("X-Slack-Signature", "v0=abc"),
                ("Content-Type", "application/json"),
            ]
        }
    }
    out = sentry_mod._before_send(event, {})
    assert out["request"]["headers"] == [("Content-Type", "application/json")]


def test_before_send_drops_body_and_secret_keys():
    event = {
        "request": {
            "body": "token=xoxb-secret&command=/oncall",
            "data": {"signing_secret": "abc"},
            "method": "POST",
        },
        "extra": {
            "slack_signing_secret": "abc",
            "tcx_password": "hunter2",
            "bot_token": "xoxb-secret",
            "hmac_secret": "aabbcc",
            "shift_date": "2026-08-29",
        },
    }
    out = sentry_mod._before_send(event, {})
    assert "body" not in out["request"]
    assert "data" not in out["request"]
    assert out["request"]["method"] == "POST"
    assert "slack_signing_secret" not in out["extra"]
    assert "tcx_password" not in out["extra"]
    assert "bot_token" not in out["extra"]
    assert "hmac_secret" not in out["extra"]
    assert out["extra"]["shift_date"] == "2026-08-29"


def test_before_send_drops_exception_and_thread_frame_locals():
    event = {
        "exception": {
            "values": [
                {
                    "stacktrace": {
                        "frames": [
                            {
                                "function": "handler",
                                "vars": {
                                    "signing_secret": "abc",
                                    "SecretString": "aabbcc",
                                },
                            }
                        ]
                    }
                }
            ]
        },
        "threads": {
            "values": [
                {
                    "stacktrace": {
                        "frames": [
                            {
                                "function": "_authenticate_user",
                                "vars": {"password": "hunter2"},
                            }
                        ]
                    }
                }
            ]
        },
        "stacktrace": {
            "frames": [{"function": "get_secret", "vars": {"item": {"token": "x"}}}]
        },
    }
    out = sentry_mod._before_send(event, {})
    assert "vars" not in out["exception"]["values"][0]["stacktrace"]["frames"][0]
    assert "vars" not in out["threads"]["values"][0]["stacktrace"]["frames"][0]
    assert "vars" not in out["stacktrace"]["frames"][0]
    assert (
        out["exception"]["values"][0]["stacktrace"]["frames"][0]["function"]
        == "handler"
    )