1
- import mock
2
-
3
1
import pytest
4
2
5
3
pytest .importorskip ("celery" )
16
14
from sentry_sdk .crons import MonitorStatus
17
15
from celery .schedules import crontab , schedule
18
16
17
+ try :
18
+ from unittest import mock # python 3.3 and above
19
+ from unittest .mock import MagicMock
20
+ except ImportError :
21
+ import mock # python < 3.3
22
+ from mock import MagicMock
23
+
19
24
20
25
def test_get_headers ():
21
- fake_task = mock . MagicMock ()
26
+ fake_task = MagicMock ()
22
27
fake_task .request = {
23
28
"bla" : "blub" ,
24
29
"foo" : "bar" ,
@@ -69,7 +74,7 @@ def test_get_humanized_interval(seconds, expected_tuple):
69
74
70
75
71
76
def test_crons_task_success ():
72
- fake_task = mock . MagicMock ()
77
+ fake_task = MagicMock ()
73
78
fake_task .request = {
74
79
"headers" : {
75
80
"sentry-monitor-slug" : "test123" ,
@@ -113,7 +118,7 @@ def test_crons_task_success():
113
118
114
119
115
120
def test_crons_task_failure ():
116
- fake_task = mock . MagicMock ()
121
+ fake_task = MagicMock ()
117
122
fake_task .request = {
118
123
"headers" : {
119
124
"sentry-monitor-slug" : "test123" ,
@@ -157,7 +162,7 @@ def test_crons_task_failure():
157
162
158
163
159
164
def test_crons_task_retry ():
160
- fake_task = mock . MagicMock ()
165
+ fake_task = MagicMock ()
161
166
fake_task .request = {
162
167
"headers" : {
163
168
"sentry-monitor-slug" : "test123" ,
@@ -201,8 +206,8 @@ def test_crons_task_retry():
201
206
202
207
203
208
def test_get_monitor_config ():
204
- app = mock . MagicMock ()
205
- app .conf = mock . MagicMock ()
209
+ app = MagicMock ()
210
+ app .conf = MagicMock ()
206
211
app .conf .timezone = "Europe/Vienna"
207
212
208
213
celery_schedule = crontab (day_of_month = "3" , hour = "12" , minute = "*/10" )
@@ -229,14 +234,14 @@ def test_get_monitor_config():
229
234
"timezone" : "Europe/Vienna" ,
230
235
}
231
236
232
- unknown_celery_schedule = mock . MagicMock ()
237
+ unknown_celery_schedule = MagicMock ()
233
238
monitor_config = _get_monitor_config (unknown_celery_schedule , app )
234
239
assert monitor_config == {}
235
240
236
241
237
242
def test_get_monitor_config_default_timezone ():
238
- app = mock . MagicMock ()
239
- app .conf = mock . MagicMock ()
243
+ app = MagicMock ()
244
+ app .conf = MagicMock ()
240
245
app .conf .timezone = None
241
246
242
247
celery_schedule = crontab (day_of_month = "3" , hour = "12" , minute = "*/10" )
@@ -259,18 +264,18 @@ def test_exclude_beat_tasks_option(
259
264
"""
260
265
Test excluding Celery Beat tasks from automatic instrumentation.
261
266
"""
262
- fake_apply_entry = mock . MagicMock ()
267
+ fake_apply_entry = MagicMock ()
263
268
264
- fake_scheduler = mock . MagicMock ()
269
+ fake_scheduler = MagicMock ()
265
270
fake_scheduler .apply_entry = fake_apply_entry
266
271
267
- fake_integration = mock . MagicMock ()
272
+ fake_integration = MagicMock ()
268
273
fake_integration .exclude_beat_tasks = exclude_beat_tasks
269
274
270
- fake_schedule_entry = mock . MagicMock ()
275
+ fake_schedule_entry = MagicMock ()
271
276
fake_schedule_entry .name = task_name
272
277
273
- fake_get_monitor_config = mock . MagicMock ()
278
+ fake_get_monitor_config = MagicMock ()
274
279
275
280
with mock .patch (
276
281
"sentry_sdk.integrations.celery.Scheduler" , fake_scheduler
@@ -290,10 +295,10 @@ def test_exclude_beat_tasks_option(
290
295
291
296
if task_in_excluded_beat_tasks :
292
297
# Only the original Scheduler.apply_entry() is called, _get_monitor_config is NOT called.
293
- fake_apply_entry .assert_called_once ()
298
+ assert fake_apply_entry .call_count == 1
294
299
_get_monitor_config .assert_not_called ()
295
300
296
301
else :
297
302
# The original Scheduler.apply_entry() is called, AND _get_monitor_config is called.
298
- fake_apply_entry .assert_called_once ()
299
- _get_monitor_config .assert_called_once ()
303
+ assert fake_apply_entry .call_count == 1
304
+ assert _get_monitor_config .call_count == 1
0 commit comments