test_crons.py

import uuid
from unittest import mock

import pytest

import sentry_sdk

from sentry_sdk.crons import capture_checkin


@sentry_sdk.monitor(monitor_slug="abc123")
def _hello_world(name):
    return "Hello, {}".format(name)


@sentry_sdk.monitor(monitor_slug="def456")
def _break_world(name):
    1 / 0
    return "Hello, {}".format(name)


def _hello_world_contextmanager(name):
    with sentry_sdk.monitor(monitor_slug="abc123"):
        return "Hello, {}".format(name)


def _break_world_contextmanager(name):
    with sentry_sdk.monitor(monitor_slug="def456"):
        1 / 0
        return "Hello, {}".format(name)


@sentry_sdk.monitor(monitor_slug="abc123")
async def _hello_world_async(name):
    return "Hello, {}".format(name)


@sentry_sdk.monitor(monitor_slug="def456")
async def _break_world_async(name):
    1 / 0
    return "Hello, {}".format(name)


async def my_coroutine():
    return


async def _hello_world_contextmanager_async(name):
    with sentry_sdk.monitor(monitor_slug="abc123"):
        await my_coroutine()
        return "Hello, {}".format(name)


async def _break_world_contextmanager_async(name):
    with sentry_sdk.monitor(monitor_slug="def456"):
        await my_coroutine()
        1 / 0
        return "Hello, {}".format(name)


@sentry_sdk.monitor(monitor_slug="ghi789", monitor_config=None)
def _no_monitor_config():
    return


@sentry_sdk.monitor(
    monitor_slug="ghi789",
    monitor_config={
        "schedule": {"type": "crontab", "value": "0 0 * * *"},
        "failure_issue_threshold": 5,
    },
)
def _with_monitor_config():
    return


def test_decorator(sentry_init):
    sentry_init()

    with mock.patch(
        "sentry_sdk.crons.decorator.capture_checkin"
    ) as fake_capture_checkin:
        result = _hello_world("Grace")
        assert result == "Hello, Grace"

        # Check for initial checkin
        fake_capture_checkin.assert_has_calls(
            [
                mock.call(
                    monitor_slug="abc123", status="in_progress", monitor_config=None
                ),
            ]
        )

        # Check for final checkin
        assert fake_capture_checkin.call_args[1]["monitor_slug"] == "abc123"
        assert fake_capture_checkin.call_args[1]["status"] == "ok"
        assert fake_capture_checkin.call_args[1]["duration"]
        assert fake_capture_checkin.call_args[1]["check_in_id"]


def test_decorator_error(sentry_init):
    sentry_init()

    with mock.patch(
        "sentry_sdk.crons.decorator.capture_checkin"
    ) as fake_capture_checkin:
        with pytest.raises(ZeroDivisionError):
            result = _break_world("Grace")

        assert "result" not in locals()

        # Check for initial checkin
        fake_capture_checkin.assert_has_calls(
            [
                mock.call(
                    monitor_slug="def456", status="in_progress", monitor_config=None
                ),
            ]
        )

        # Check for final checkin
        assert fake_capture_checkin.call_args[1]["monitor_slug"] == "def456"
        assert fake_capture_checkin.call_args[1]["status"] == "error"
        assert fake_capture_checkin.call_args[1]["duration"]
        assert fake_capture_checkin.call_args[1]["check_in_id"]


def test_contextmanager(sentry_init):
    sentry_init()

    with mock.patch(
        "sentry_sdk.crons.decorator.capture_checkin"
    ) as fake_capture_checkin:
        result = _hello_world_contextmanager("Grace")
        assert result == "Hello, Grace"

        # Check for initial checkin
        fake_capture_checkin.assert_has_calls(
            [
                mock.call(
                    monitor_slug="abc123", status="in_progress", monitor_config=None
                ),
            ]
        )

        # Check for final checkin
        assert fake_capture_checkin.call_args[1]["monitor_slug"] == "abc123"
        assert fake_capture_checkin.call_args[1]["status"] == "ok"
        assert fake_capture_checkin.call_args[1]["duration"]
        assert fake_capture_checkin.call_args[1]["check_in_id"]


def test_contextmanager_error(sentry_init):
    sentry_init()

    with mock.patch(
        "sentry_sdk.crons.decorator.capture_checkin"
    ) as fake_capture_checkin:
        with pytest.raises(ZeroDivisionError):
            result = _break_world_contextmanager("Grace")

        assert "result" not in locals()

        # Check for initial checkin
        fake_capture_checkin.assert_has_calls(
            [
                mock.call(
                    monitor_slug="def456", status="in_progress", monitor_config=None
                ),
            ]
        )

        # Check for final checkin
        assert fake_capture_checkin.call_args[1]["monitor_slug"] == "def456"
        assert fake_capture_checkin.call_args[1]["status"] == "error"
        assert fake_capture_checkin.call_args[1]["duration"]
        assert fake_capture_checkin.call_args[1]["check_in_id"]


def test_capture_checkin_simple(sentry_init):
    sentry_init()

    check_in_id = capture_checkin(
        monitor_slug="abc123",
        check_in_id="112233",
        status=None,
        duration=None,
    )
    assert check_in_id == "112233"


def test_sample_rate_doesnt_affect_crons(sentry_init, capture_envelopes):
    sentry_init(sample_rate=0)
    envelopes = capture_envelopes()

    capture_checkin(check_in_id="112233")

    assert len(envelopes) == 1

    check_in = envelopes[0].items[0].payload.json
    assert check_in["check_in_id"] == "112233"


def test_capture_checkin_new_id(sentry_init):
    sentry_init()

    with mock.patch("uuid.uuid4") as mock_uuid:
        mock_uuid.return_value = uuid.UUID("a8098c1a-f86e-11da-bd1a-00112444be1e")
        check_in_id = capture_checkin(
            monitor_slug="abc123",
            check_in_id=None,
            status=None,
            duration=None,
        )

        assert check_in_id == "a8098c1af86e11dabd1a00112444be1e"


def test_end_to_end(sentry_init, capture_envelopes):
    sentry_init()
    envelopes = capture_envelopes()

    capture_checkin(
        monitor_slug="abc123",
        check_in_id="112233",
        duration=123,
        status="ok",
    )

    check_in = envelopes[0].items[0].payload.json

    # Check for final checkin
    assert check_in["check_in_id"] == "112233"
    assert check_in["monitor_slug"] == "abc123"
    assert check_in["status"] == "ok"
    assert check_in["duration"] == 123


def test_monitor_config(sentry_init, capture_envelopes):
    sentry_init()
    envelopes = capture_envelopes()

    monitor_config = {
        "schedule": {"type": "crontab", "value": "0 0 * * *"},
        "failure_issue_threshold": 5,
        "recovery_threshold": 5,
    }

    capture_checkin(monitor_slug="abc123", monitor_config=monitor_config)
    check_in = envelopes[0].items[0].payload.json

    # Check for final checkin
    assert check_in["monitor_slug"] == "abc123"
    assert check_in["monitor_config"] == monitor_config

    # Without passing a monitor_config the field is not in the checkin
    capture_checkin(monitor_slug="abc123")
    check_in = envelopes[1].items[0].payload.json

    assert check_in["monitor_slug"] == "abc123"
    assert "monitor_config" not in check_in


def test_decorator_monitor_config(sentry_init, capture_envelopes):
    sentry_init()
    envelopes = capture_envelopes()

    _with_monitor_config()

    assert len(envelopes) == 2

    for check_in_envelope in envelopes:
        assert len(check_in_envelope.items) == 1
        check_in = check_in_envelope.items[0].payload.json

        assert check_in["monitor_slug"] == "ghi789"
        assert check_in["monitor_config"] == {
            "schedule": {"type": "crontab", "value": "0 0 * * *"},
            "failure_issue_threshold": 5,
        }


def test_decorator_no_monitor_config(sentry_init, capture_envelopes):
    sentry_init()
    envelopes = capture_envelopes()

    _no_monitor_config()

    assert len(envelopes) == 2

    for check_in_envelope in envelopes:
        assert len(check_in_envelope.items) == 1
        check_in = check_in_envelope.items[0].payload.json

        assert check_in["monitor_slug"] == "ghi789"
        assert "monitor_config" not in check_in


def test_capture_checkin_sdk_not_initialized():
    # Tests that the capture_checkin does not raise an error when Sentry SDK is not initialized.
    # sentry_init() is intentionally omitted.
    check_in_id = capture_checkin(
        monitor_slug="abc123",
        check_in_id="112233",
        status=None,
        duration=None,
    )
    assert check_in_id == "112233"


def test_scope_data_in_checkin(sentry_init, capture_envelopes):
    sentry_init()
    envelopes = capture_envelopes()

    valid_keys = [
        # Mandatory event keys
        "type",
        "event_id",
        "timestamp",
        "platform",
        # Optional event keys
        "release",
        "environment",
        "server_name",
        "sdk",
        # Mandatory check-in specific keys
        "check_in_id",
        "monitor_slug",
        "status",
        # Optional check-in specific keys
        "duration",
        "monitor_config",
        "contexts",  # an event processor adds this
    ]

    # Add some data to the scope
    sentry_sdk.add_breadcrumb(message="test breadcrumb")
    sentry_sdk.set_context("test_context", {"test_key": "test_value"})
    sentry_sdk.set_extra("test_extra", "test_value")
    sentry_sdk.set_level("warning")
    sentry_sdk.set_tag("test_tag", "test_value")

    capture_checkin(
        monitor_slug="abc123",
        check_in_id="112233",
        status="ok",
        duration=123,
    )

    (envelope,) = envelopes
    check_in_event = envelope.items[0].payload.json

    invalid_keys = []
    for key in check_in_event.keys():
        if key not in valid_keys:
            invalid_keys.append(key)

    assert len(invalid_keys) == 0, "Unexpected keys found in checkin: {}".format(
        invalid_keys
    )


@pytest.mark.asyncio
async def test_decorator_async(sentry_init):
    sentry_init()

    with mock.patch(
        "sentry_sdk.crons.decorator.capture_checkin"
    ) as fake_capture_checkin:
        result = await _hello_world_async("Grace")
        assert result == "Hello, Grace"

        # Check for initial checkin
        fake_capture_checkin.assert_has_calls(
            [
                mock.call(
                    monitor_slug="abc123", status="in_progress", monitor_config=None
                ),
            ]
        )

        # Check for final checkin
        assert fake_capture_checkin.call_args[1]["monitor_slug"] == "abc123"
        assert fake_capture_checkin.call_args[1]["status"] == "ok"
        assert fake_capture_checkin.call_args[1]["duration"]
        assert fake_capture_checkin.call_args[1]["check_in_id"]


@pytest.mark.asyncio
async def test_decorator_error_async(sentry_init):
    sentry_init()

    with mock.patch(
        "sentry_sdk.crons.decorator.capture_checkin"
    ) as fake_capture_checkin:
        with pytest.raises(ZeroDivisionError):
            result = await _break_world_async("Grace")

        assert "result" not in locals()

        # Check for initial checkin
        fake_capture_checkin.assert_has_calls(
            [
                mock.call(
                    monitor_slug="def456", status="in_progress", monitor_config=None
                ),
            ]
        )

        # Check for final checkin
        assert fake_capture_checkin.call_args[1]["monitor_slug"] == "def456"
        assert fake_capture_checkin.call_args[1]["status"] == "error"
        assert fake_capture_checkin.call_args[1]["duration"]
        assert fake_capture_checkin.call_args[1]["check_in_id"]


@pytest.mark.asyncio
async def test_contextmanager_async(sentry_init):
    sentry_init()

    with mock.patch(
        "sentry_sdk.crons.decorator.capture_checkin"
    ) as fake_capture_checkin:
        result = await _hello_world_contextmanager_async("Grace")
        assert result == "Hello, Grace"

        # Check for initial checkin
        fake_capture_checkin.assert_has_calls(
            [
                mock.call(
                    monitor_slug="abc123", status="in_progress", monitor_config=None
                ),
            ]
        )

        # Check for final checkin
        assert fake_capture_checkin.call_args[1]["monitor_slug"] == "abc123"
        assert fake_capture_checkin.call_args[1]["status"] == "ok"
        assert fake_capture_checkin.call_args[1]["duration"]
        assert fake_capture_checkin.call_args[1]["check_in_id"]


@pytest.mark.asyncio
async def test_contextmanager_error_async(sentry_init):
    sentry_init()

    with mock.patch(
        "sentry_sdk.crons.decorator.capture_checkin"
    ) as fake_capture_checkin:
        with pytest.raises(ZeroDivisionError):
            result = await _break_world_contextmanager_async("Grace")

        assert "result" not in locals()

        # Check for initial checkin
        fake_capture_checkin.assert_has_calls(
            [
                mock.call(
                    monitor_slug="def456", status="in_progress", monitor_config=None
                ),
            ]
        )

        # Check for final checkin
        assert fake_capture_checkin.call_args[1]["monitor_slug"] == "def456"
        assert fake_capture_checkin.call_args[1]["status"] == "error"
        assert fake_capture_checkin.call_args[1]["duration"]
        assert fake_capture_checkin.call_args[1]["check_in_id"]