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
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
|
from itertools import combinations
from unittest import mock
import pytest
from django.core.exceptions import ValidationError
from django.test import TestCase
from django_celery_beat.admin import PeriodicTaskAdmin
from django_celery_beat.models import (DAYS, ClockedSchedule, CrontabSchedule,
IntervalSchedule, PeriodicTask,
SolarSchedule)
@pytest.mark.django_db
class ActionsTests(TestCase):
@classmethod
def setUpTestData(cls):
super().setUpTestData()
cls.interval_schedule = IntervalSchedule.objects.create(every=10,
period=DAYS)
def test_toggle_action(self):
PeriodicTask.objects.create(name='name1', task='task1', enabled=False,
interval=self.interval_schedule)
PeriodicTask.objects.create(name='name2', task='task2', enabled=True,
interval=self.interval_schedule)
PeriodicTask.objects.create(name='name3', task='task3', enabled=False,
interval=self.interval_schedule)
qs = PeriodicTask.objects.all()
PeriodicTaskAdmin(PeriodicTask, None)._toggle_tasks_activity(qs)
e1 = PeriodicTask.objects.get(name='name1', task='task1').enabled
e2 = PeriodicTask.objects.get(name='name2', task='task2').enabled
e3 = PeriodicTask.objects.get(name='name3', task='task3').enabled
self.assertTrue(e1)
self.assertFalse(e2)
self.assertTrue(e3)
def test_toggle_action_all_enabled(self):
PeriodicTask.objects.create(name='name1', task='task1', enabled=True,
interval=self.interval_schedule)
PeriodicTask.objects.create(name='name2', task='task2', enabled=True,
interval=self.interval_schedule)
PeriodicTask.objects.create(name='name3', task='task3', enabled=True,
interval=self.interval_schedule)
qs = PeriodicTask.objects.all()
PeriodicTaskAdmin(PeriodicTask, None)._toggle_tasks_activity(qs)
e1 = PeriodicTask.objects.get(name='name1', task='task1').enabled
e2 = PeriodicTask.objects.get(name='name2', task='task2').enabled
e3 = PeriodicTask.objects.get(name='name3', task='task3').enabled
self.assertFalse(e1)
self.assertFalse(e2)
self.assertFalse(e3)
def test_toggle_action_all_disabled(self):
PeriodicTask.objects.create(name='name1', task='task1', enabled=False,
interval=self.interval_schedule)
PeriodicTask.objects.create(name='name2', task='task2', enabled=False,
interval=self.interval_schedule)
PeriodicTask.objects.create(name='name3', task='task3', enabled=False,
interval=self.interval_schedule)
qs = PeriodicTask.objects.all()
PeriodicTaskAdmin(PeriodicTask, None)._toggle_tasks_activity(qs)
e1 = PeriodicTask.objects.get(name='name1', task='task1').enabled
e2 = PeriodicTask.objects.get(name='name2', task='task2').enabled
e3 = PeriodicTask.objects.get(name='name3', task='task3').enabled
self.assertTrue(e1)
self.assertTrue(e2)
self.assertTrue(e3)
@pytest.mark.django_db
class ValidateUniqueTests(TestCase):
def test_validate_unique_raises_if_schedule_not_set(self):
with self.assertRaises(ValidationError) as cm:
PeriodicTask(name='task0').validate_unique()
self.assertEqual(
cm.exception.args[0],
'One of clocked, interval, crontab, or solar must be set.',
)
def test_validate_unique_raises_for_multiple_schedules(self):
schedules = [
('crontab', CrontabSchedule()),
('interval', IntervalSchedule()),
('solar', SolarSchedule()),
('clocked', ClockedSchedule())
]
expected_error_msg = (
'Only one of clocked, interval, crontab, or solar '
'must be set'
)
for i, options in enumerate(combinations(schedules, 2)):
name = f'task{i}'
options_dict = dict(options)
with self.assertRaises(ValidationError) as cm:
PeriodicTask(name=name, **options_dict).validate_unique()
errors = cm.exception.args[0]
self.assertEqual(errors.keys(), options_dict.keys())
for error_msg in errors.values():
self.assertEqual(error_msg, [expected_error_msg])
def test_validate_unique_not_raises(self):
PeriodicTask(crontab=CrontabSchedule()).validate_unique()
PeriodicTask(interval=IntervalSchedule()).validate_unique()
PeriodicTask(solar=SolarSchedule()).validate_unique()
PeriodicTask(clocked=ClockedSchedule(), one_off=True).validate_unique()
@pytest.mark.django_db
class DisableTasksTest(TestCase):
@classmethod
def setUpTestData(cls):
super().setUpTestData()
cls.interval_schedule = IntervalSchedule.objects.create(every=10,
period=DAYS)
@mock.patch('django_celery_beat.admin.PeriodicTaskAdmin.message_user')
def test_disable_tasks(self, mock_message_user):
PeriodicTask.objects.create(name='name1', task='task1', enabled=True,
interval=self.interval_schedule)
PeriodicTask.objects.create(name='name2', task='task2', enabled=True,
interval=self.interval_schedule)
qs = PeriodicTask.objects.all()
PeriodicTaskAdmin(PeriodicTask, None).disable_tasks(None, qs)
for periodic_task in qs:
self.assertFalse(periodic_task.enabled)
self.assertIsNone(periodic_task.last_run_at)
mock_message_user.assert_called_once()
|
