105 lines
3.4 KiB
Python
105 lines
3.4 KiB
Python
import importlib
|
|
|
|
from django.apps import apps
|
|
from django.test import TestCase
|
|
from django_celery_beat.models import CrontabSchedule, IntervalSchedule, PeriodicTask
|
|
|
|
schedule_migration = importlib.import_module(
|
|
"apps.parsers.migrations.0025_consolidate_inspections_schedule"
|
|
)
|
|
|
|
|
|
class DefaultParserScheduleTest(TestCase):
|
|
def test_inspections_have_one_canonical_sync_schedule(self):
|
|
crontab = CrontabSchedule.objects.create(
|
|
minute="0",
|
|
hour="0",
|
|
day_of_week="6",
|
|
day_of_month="*",
|
|
month_of_year="*",
|
|
timezone="Europe/Moscow",
|
|
)
|
|
interval = IntervalSchedule.objects.create(
|
|
every=7,
|
|
period=IntervalSchedule.DAYS,
|
|
)
|
|
PeriodicTask.objects.create(
|
|
name="parser:inspections:weekly-saturday-msk",
|
|
task="apps.parsers.tasks.parse_inspections",
|
|
crontab=crontab,
|
|
enabled=True,
|
|
)
|
|
PeriodicTask.objects.create(
|
|
name="parse-inspections-weekly",
|
|
task="apps.parsers.tasks.parse_inspections",
|
|
interval=interval,
|
|
enabled=True,
|
|
)
|
|
PeriodicTask.objects.create(
|
|
name="parser:sync_inspections:weekly-saturday-msk",
|
|
task="apps.parsers.tasks.parse_inspections",
|
|
args='["stale"]',
|
|
kwargs='{"year": 2020}',
|
|
crontab=crontab,
|
|
enabled=True,
|
|
)
|
|
|
|
schedule_migration.consolidate_inspections_schedule(apps, None)
|
|
schedule_migration.consolidate_inspections_schedule(apps, None)
|
|
|
|
schedule_names = {
|
|
"parser:inspections:weekly-saturday-msk",
|
|
"parser:sync_inspections:weekly-saturday-msk",
|
|
"parse-inspections-weekly",
|
|
}
|
|
|
|
schedules = list(
|
|
PeriodicTask.objects.filter(name__in=schedule_names)
|
|
.order_by("name")
|
|
.values_list("name", "task", "enabled", "args", "kwargs")
|
|
)
|
|
|
|
self.assertEqual(
|
|
schedules,
|
|
[
|
|
(
|
|
"parser:sync_inspections:weekly-saturday-msk",
|
|
"apps.parsers.tasks.sync_inspections",
|
|
True,
|
|
"[]",
|
|
"{}",
|
|
)
|
|
],
|
|
)
|
|
|
|
def test_fns_weekly_schedule_runs_bounded_api_sync(self):
|
|
fns_schedule_migration = importlib.import_module(
|
|
"apps.parsers.migrations.0026_update_fns_financial_schedule"
|
|
)
|
|
interval = IntervalSchedule.objects.create(
|
|
every=7,
|
|
period=IntervalSchedule.DAYS,
|
|
)
|
|
PeriodicTask.objects.create(
|
|
name="parser:fns_financial:weekly-saturday-msk",
|
|
task="apps.parsers.tasks.scan_fns_directory",
|
|
args='["stale"]',
|
|
kwargs='{"stale": true}',
|
|
interval=interval,
|
|
enabled=True,
|
|
)
|
|
|
|
fns_schedule_migration.update_fns_financial_schedule(apps, None)
|
|
fns_schedule_migration.update_fns_financial_schedule(apps, None)
|
|
|
|
task = PeriodicTask.objects.get(name="parser:fns_financial:weekly-saturday-msk")
|
|
self.assertEqual(
|
|
task.task,
|
|
"apps.parsers.tasks.sync_fns_financial_reports",
|
|
)
|
|
self.assertEqual(task.args, "[]")
|
|
self.assertEqual(task.kwargs, "{}")
|
|
self.assertTrue(task.enabled)
|
|
self.assertIsNotNone(task.crontab_id)
|
|
self.assertIsNone(task.interval_id)
|