procurement-ingest/tests/test_sentry_init.py

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

from unittest.mock import patch

import pytest
from sentry_sdk.integrations.aws_lambda import AwsLambdaIntegration

from tests.support import load_lambda_module


@pytest.fixture
def sentry_mod():
    return load_lambda_module("shared", "sentry_init")


def _reexec(mod, 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:
        mod.__spec__.loader.exec_module(mod)
        return mocked


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


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


def test_set_dsn_inits_lambda_integration(sentry_mod, monkeypatch):
    mocked = _reexec(sentry_mod, 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


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


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


def test_before_send_drops_body_prompt_and_secret_keys(sentry_mod):
    event = {
        "request": {
            "body": "<email>raw mime</email>",
            "data": {"prompt": "EXTRACT"},
            "method": "POST",
        },
        "extra": {
            "raw_email": "From: attacker",
            "bedrock_prompt": "ignore previous",
            "hmac_secret": "aabbcc",
            "po_number": "123",
        },
    }
    out = sentry_mod._before_send(event, {})
    assert "body" not in out["request"]
    assert "data" not in out["request"]
    assert out["request"]["method"] == "POST"
    assert "raw_email" not in out["extra"]
    assert "bedrock_prompt" not in out["extra"]
    assert "hmac_secret" not in out["extra"]
    assert out["extra"]["po_number"] == "123"


def test_before_send_drops_exception_and_thread_frame_locals(sentry_mod):
    event = {
        "exception": {
            "values": [
                {
                    "stacktrace": {
                        "frames": [
                            {
                                "function": "handler",
                                "vars": {
                                    "raw_email": "From: attacker",
                                    "SecretString": "aabbcc",
                                },
                            }
                        ]
                    }
                }
            ]
        },
        "threads": {
            "values": [
                {
                    "stacktrace": {
                        "frames": [
                            {"function": "extract_with_claude", "vars": {"prompt": "EXTRACT"}}
                        ]
                    }
                }
            ]
        },
        "stacktrace": {
            "frames": [{"function": "save_work_order", "vars": {"item": {"wo": 1}}}]
        },
    }
    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"