74 lines
2.8 KiB
Python
74 lines
2.8 KiB
Python
"""Tests for organization snapshot tasks and schedules."""
|
||
|
||
from importlib import import_module
|
||
|
||
from django.apps import apps as django_apps
|
||
from django.core.cache import cache
|
||
from django.test import TestCase
|
||
from django_celery_beat.models import PeriodicTask
|
||
from organizations.cache import get_organization_api_cache_version
|
||
from organizations.models import Organization
|
||
from organizations.tasks import refresh_all_organization_data_snapshots
|
||
|
||
from tests.apps.parsers.factories import IndustrialCertificateRecordFactory
|
||
|
||
|
||
class OrganizationSnapshotTasksTest(TestCase):
|
||
"""Checks Celery tasks that maintain API v2 organization snapshots."""
|
||
|
||
def test_refresh_all_task_rebuilds_snapshots_and_invalidates_api_cache(self):
|
||
organization = Organization.objects.create(
|
||
name='ООО "Снапшот"',
|
||
inn="7800000401",
|
||
ogrn="1027700144401",
|
||
)
|
||
IndustrialCertificateRecordFactory(
|
||
inn=organization.inn,
|
||
ogrn=organization.ogrn,
|
||
certificate_number="FULL-SNAPSHOT-CERT",
|
||
)
|
||
cache.set("unrelated:test", {"keep": True}, timeout=60)
|
||
cache_version_before = get_organization_api_cache_version()
|
||
|
||
result = refresh_all_organization_data_snapshots(batch_size=10)
|
||
|
||
self.assertEqual(result["processed"], 1)
|
||
self.assertEqual(result["created"], 1)
|
||
self.assertEqual(result["updated"], 0)
|
||
self.assertNotEqual(
|
||
get_organization_api_cache_version(),
|
||
cache_version_before,
|
||
)
|
||
self.assertEqual(cache.get("unrelated:test"), {"keep": True})
|
||
snapshot = organization.data_snapshot
|
||
self.assertEqual(
|
||
snapshot.data["industrial"][0]["certificate_number"],
|
||
"FULL-SNAPSHOT-CERT",
|
||
)
|
||
|
||
|
||
class OrganizationSnapshotScheduleMigrationTest(TestCase):
|
||
"""Checks data migration that schedules full snapshot refresh."""
|
||
|
||
def test_migration_seeds_daily_snapshot_refresh_periodic_task(self):
|
||
migration = import_module(
|
||
"organizations.migrations.0004_seed_daily_snapshot_refresh_schedule"
|
||
)
|
||
|
||
migration.seed_daily_snapshot_refresh_schedule(django_apps, None)
|
||
migration.seed_daily_snapshot_refresh_schedule(django_apps, None)
|
||
|
||
task = PeriodicTask.objects.get(
|
||
name=migration.DAILY_ORGANIZATION_SNAPSHOT_TASK_NAME
|
||
)
|
||
self.assertEqual(
|
||
task.task,
|
||
"organizations.tasks.refresh_all_organization_data_snapshots",
|
||
)
|
||
self.assertTrue(task.enabled)
|
||
self.assertEqual(task.args, "[]")
|
||
self.assertEqual(task.kwargs, '{"batch_size": 100}')
|
||
self.assertEqual(task.crontab.minute, "30")
|
||
self.assertEqual(task.crontab.hour, "4")
|
||
self.assertEqual(str(task.crontab.timezone), "Europe/Moscow")
|