19
19
20
20
from unittest import mock
21
21
22
- import pytest
23
-
22
+ from airflow import DAG
24
23
from airflow .providers .pagerduty .hooks .pagerduty_events import PagerdutyEventsHook
25
24
from airflow .providers .pagerduty .notifications .pagerduty import (
26
25
PagerdutyNotifier ,
27
26
send_pagerduty_notification ,
28
27
)
29
- from airflow .providers .standard .operators .empty import EmptyOperator
30
-
31
- pytestmark = pytest .mark .db_test
32
-
33
28
34
29
PAGERDUTY_API_DEFAULT_CONN_ID = PagerdutyEventsHook .default_conn_name
35
30
36
31
37
32
class TestPagerdutyNotifier :
38
33
@mock .patch ("airflow.providers.pagerduty.notifications.pagerduty.PagerdutyEventsHook" )
39
- def test_notifier (self , mock_pagerduty_event_hook , dag_maker ):
40
- with dag_maker ("test_notifier" ) as dag :
41
- EmptyOperator (task_id = "task1" )
34
+ def test_notifier (self , mock_pagerduty_event_hook ):
35
+ dag = DAG ("test_notifier" )
42
36
notifier = send_pagerduty_notification (summary = "DISK at 99%" , severity = "critical" , action = "trigger" )
43
37
notifier ({"dag" : dag })
44
38
mock_pagerduty_event_hook .return_value .send_event .assert_called_once_with (
@@ -56,9 +50,8 @@ def test_notifier(self, mock_pagerduty_event_hook, dag_maker):
56
50
)
57
51
58
52
@mock .patch ("airflow.providers.pagerduty.notifications.pagerduty.PagerdutyEventsHook" )
59
- def test_notifier_with_notifier_class (self , mock_pagerduty_event_hook , dag_maker ):
60
- with dag_maker ("test_notifier" ) as dag :
61
- EmptyOperator (task_id = "task1" )
53
+ def test_notifier_with_notifier_class (self , mock_pagerduty_event_hook ):
54
+ dag = DAG ("test_notifier" )
62
55
notifier = PagerdutyNotifier (summary = "DISK at 99%" , severity = "critical" , action = "trigger" )
63
56
notifier ({"dag" : dag })
64
57
mock_pagerduty_event_hook .return_value .send_event .assert_called_once_with (
@@ -76,9 +69,8 @@ def test_notifier_with_notifier_class(self, mock_pagerduty_event_hook, dag_maker
76
69
)
77
70
78
71
@mock .patch ("airflow.providers.pagerduty.notifications.pagerduty.PagerdutyEventsHook" )
79
- def test_notifier_templated (self , mock_pagerduty_event_hook , dag_maker ):
80
- with dag_maker ("test_notifier" ) as dag :
81
- EmptyOperator (task_id = "task1" )
72
+ def test_notifier_templated (self , mock_pagerduty_event_hook ):
73
+ dag = DAG ("test_notifier" )
82
74
83
75
notifier = PagerdutyNotifier (
84
76
summary = "DISK at 99% {{dag.dag_id}}" ,
0 commit comments