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
|
from itertools import combinations
from django.test import TestCase
from django_celery_beat.admin import PeriodicTaskAdmin
from django_celery_beat.models import \
DAYS, \
PeriodicTask, \
CrontabSchedule, \
IntervalSchedule, \
SolarSchedule, \
ClockedSchedule
from django.core.exceptions import ValidationError
class ActionsTests(TestCase):
@classmethod
def setUpTestData(cls):
super(ActionsTests, cls).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)
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.assertEquals(
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 = 'task{}'.format(i)
options_dict = dict(options)
with self.assertRaises(ValidationError) as cm:
PeriodicTask(name=name, **options_dict).validate_unique()
errors = cm.exception.args[0]
self.assertEquals(errors.keys(), options_dict.keys())
for error_msg in errors.values():
self.assertEquals(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()
|