sqlmesh/tests/core/test_notification_target.py at fb8d8a503eca4925663142971ea68db92e9acc09 · SQLMesh/sqlmesh · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
from __future__ import annotations

import typing as t
from unittest import mock

import pytest

from sqlmesh.core.notification_target import (
    ConsoleNotificationTarget,
    NotificationEvent,
    NotificationStatus,
    NotificationTargetManager,
)


@pytest.fixture
def notification_target_manager_with_spy(mocker) -> tuple[NotificationTargetManager, t.Callable]:
    console_notification_target_send_spy = mocker.spy(ConsoleNotificationTarget, "send")
    console_notification_target = ConsoleNotificationTarget()
    test_user_console_notification_target = ConsoleNotificationTarget(
        notify_on={NotificationEvent.APPLY_START}
    )
    notification_target_manager = NotificationTargetManager(
        notification_targets={
            NotificationEvent.APPLY_START: {console_notification_target},
            NotificationEvent.APPLY_END: {console_notification_target},
        },
        user_notification_targets={
            "test_user": {test_user_console_notification_target},
        },
    )
    return notification_target_manager, console_notification_target_send_spy


def test_notify(notification_target_manager_with_spy):
    notification_target_manager, spy = notification_target_manager_with_spy
    notification_target_manager.notify(NotificationEvent.APPLY_START, "prod", "a-plan-id")
    spy.assert_called_once_with(
        mock.ANY,
        NotificationStatus.INFO,
        "Plan `a-plan-id` apply started for environment `prod`.",
    )

    spy.reset_mock()
    notification_target_manager.notify(NotificationEvent.APPLY_END, "prod", "a-plan-id")
    spy.assert_called_once_with(
        mock.ANY,
        NotificationStatus.SUCCESS,
        "Plan `a-plan-id` apply finished for environment `prod`.",
    )

    # No notification target configured for APPLY_FAILURE event
    spy.reset_mock()
    notification_target_manager.notify(
        NotificationEvent.APPLY_FAILURE, "prod", "a-plan-id", ValueError()
    )
    spy.assert_not_called()


def test_notify_user(notification_target_manager_with_spy):
    notification_target_manager, spy = notification_target_manager_with_spy
    notification_target_manager.notify_user(
        NotificationEvent.APPLY_START, "test_user", "prod", "a-plan-id"
    )
    spy.assert_called_once_with(
        mock.ANY,
        NotificationStatus.INFO,
        "Plan `a-plan-id` apply started for environment `prod`.",
    )

    # No notification target configured for APPLY_END event for test_user
    spy.reset_mock()
    notification_target_manager.notify_user(
        NotificationEvent.APPLY_END, "test_user", "prod", "a-plan-id"
    )
    spy.assert_not_called()