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
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
|
import datetime
import os
try:
from zoneinfo import ZoneInfo, available_timezones
except ImportError:
from backports.zoneinfo import available_timezones, ZoneInfo
import pytest
from celery import schedules
from django.apps import apps
from django.conf import settings
from django.db.migrations.autodetector import MigrationAutodetector
from django.db.migrations.loader import MigrationLoader
from django.db.migrations.questioner import NonInteractiveMigrationQuestioner
from django.db.migrations.state import ProjectState
from django.test import TestCase, override_settings
from django.utils import timezone
from django_celery_beat import migrations as beat_migrations
from django_celery_beat.models import (DAYS, ClockedSchedule, CrontabSchedule,
IntervalSchedule, PeriodicTasks,
SolarSchedule,
crontab_schedule_celery_timezone)
from t.proj.models import O2OToPeriodicTasks
class MigrationTests(TestCase):
def test_no_future_duplicate_migration_numbers(self):
"""Verify no duplicate migration numbers.
Migration files with the same number can cause issues with
backward migrations, so avoid them.
"""
path = os.path.dirname(beat_migrations.__file__)
files = [f[:4] for f in os.listdir(path) if f.endswith('.py')]
expected_duplicates = [
(3, '0006'),
]
duplicates_extra = sum(count - 1 for count, _ in expected_duplicates)
duplicates_numbers = [number for _, number in expected_duplicates]
self.assertEqual(
len(files), len(set(files)) + duplicates_extra,
msg=('Detected migration files with the same migration number'
' (besides {})'.format(' and '.join(duplicates_numbers))))
def test_models_match_migrations(self):
"""Make sure that no model changes exist.
This logic is taken from django's makemigrations.py file.
Here just detect if model changes exist that require
a migration, and if so we fail.
"""
app_labels = ['django_celery_beat']
loader = MigrationLoader(None, ignore_no_migrations=True)
questioner = NonInteractiveMigrationQuestioner(
specified_apps=app_labels, dry_run=False)
autodetector = MigrationAutodetector(
loader.project_state(),
ProjectState.from_apps(apps),
questioner,
)
changes = autodetector.changes(
graph=loader.graph,
trim_to_apps=app_labels,
convert_apps=app_labels,
migration_name='fake_name',
)
self.assertTrue(
not changes,
msg='Model changes exist that do not have a migration')
class TestDuplicatesMixin:
def _test_duplicate_schedules(self, cls, kwargs, schedule=None):
sched1 = cls.objects.create(**kwargs)
cls.objects.create(**kwargs)
self.assertEqual(cls.objects.filter(**kwargs).count(), 2)
# try to create a duplicate schedule from a celery schedule
if schedule is None:
schedule = sched1.schedule
sched3 = cls.from_schedule(schedule)
# the schedule should be the first of the 2 previous duplicates
self.assertEqual(sched3, sched1)
# and the duplicates should not be deleted !
self.assertEqual(cls.objects.filter(**kwargs).count(), 2)
class CrontabScheduleTestCase(TestCase, TestDuplicatesMixin):
FIRST_VALID_TIMEZONE = available_timezones().pop()
def test_default_timezone_without_settings_config(self):
assert crontab_schedule_celery_timezone() == "UTC"
@override_settings(CELERY_TIMEZONE=FIRST_VALID_TIMEZONE)
def test_default_timezone_with_settings_config(self):
assert crontab_schedule_celery_timezone() == self.FIRST_VALID_TIMEZONE
def test_duplicate_schedules(self):
# See: https://github.com/celery/django-celery-beat/issues/322
kwargs = {
"minute": "*",
"hour": "4",
"day_of_week": "*",
"day_of_month": "*",
"month_of_year": "*",
"day_of_week": "*",
}
schedule = schedules.crontab(hour="4")
self._test_duplicate_schedules(CrontabSchedule, kwargs, schedule)
class SolarScheduleTestCase(TestCase):
EVENT_CHOICES = SolarSchedule._meta.get_field("event").choices
def test_celery_solar_schedules_sorted(self):
assert all(
self.EVENT_CHOICES[i] <= self.EVENT_CHOICES[i + 1]
for i in range(len(self.EVENT_CHOICES) - 1)
), "SolarSchedule event choices are unsorted"
def test_celery_solar_schedules_included_as_event_choices(self):
"""Make sure that all Celery solar schedules are included
in SolarSchedule `event` field choices, keeping synchronized
Celery solar events with `django-celery-beat` supported solar
events.
This test is necessary because Celery solar schedules are
hardcoded at models so that Django can discover their translations.
"""
event_choices_values = [value for value, tr in self.EVENT_CHOICES]
for solar_event in schedules.solar._all_events:
assert solar_event in event_choices_values
for event_choice in event_choices_values:
assert event_choice in schedules.solar._all_events
class IntervalScheduleTestCase(TestCase, TestDuplicatesMixin):
def test_duplicate_schedules(self):
kwargs = {'every': 1, 'period': IntervalSchedule.SECONDS}
schedule = schedules.schedule(run_every=1.0)
self._test_duplicate_schedules(IntervalSchedule, kwargs, schedule)
class ClockedScheduleTestCase(TestCase, TestDuplicatesMixin):
def test_duplicate_schedules(self):
kwargs = {'clocked_time': timezone.now()}
self._test_duplicate_schedules(ClockedSchedule, kwargs)
# IMPORTANT: we must have a valid timezone (not UTC) for accurate testing
@override_settings(TIME_ZONE='Africa/Cairo')
def test_timezone_format(self):
"""Ensure scheduled time is not shown in UTC when timezone is used"""
tz_info = datetime.datetime.now(ZoneInfo(settings.TIME_ZONE))
schedule, created = ClockedSchedule.objects.get_or_create(
clocked_time=tz_info)
# testnig str(schedule) calls make_aware() internally
assert str(schedule.clocked_time) == str(schedule)
@pytest.mark.django_db()
class OneToOneRelTestCase(TestCase):
"""
Make sure that when OneToOne relation Model changed,
the `PeriodicTasks.last_update` will be update.
"""
@classmethod
def setUpTestData(cls):
super().setUpTestData()
cls.interval_schedule = IntervalSchedule.objects.create(
every=10, period=DAYS
)
def test_trigger_update_when_saved(self):
o2o_to_periodic_tasks = O2OToPeriodicTasks.objects.create(
name='name1',
task='task1',
enabled=True,
interval=self.interval_schedule
)
not_changed_dt = PeriodicTasks.last_change()
o2o_to_periodic_tasks.enabled = True # Change something on instance.
o2o_to_periodic_tasks.save()
has_changed_dt = PeriodicTasks.last_change()
self.assertTrue(
not_changed_dt != has_changed_dt,
'The `PeriodicTasks.last_update` has not be update.'
)
# Check the `PeriodicTasks` does be updated.
def test_trigger_update_when_deleted(self):
o2o_to_periodic_tasks = O2OToPeriodicTasks.objects.create(
name='name1',
task='task1',
enabled=True,
interval=self.interval_schedule
)
not_changed_dt = PeriodicTasks.last_change()
o2o_to_periodic_tasks.delete()
has_changed_dt = PeriodicTasks.last_change()
self.assertTrue(
not_changed_dt != has_changed_dt,
'The `PeriodicTasks.last_update` has not be update.'
)
# Check the `PeriodicTasks` does be updated.
|
