83 lines
2.5 KiB
Python
83 lines
2.5 KiB
Python
"""Tests for core Celery tasks"""
|
|
|
|
|
|
from apps.core.tasks import (
|
|
BaseTask,
|
|
IdempotentTask,
|
|
PeriodicTask,
|
|
TimedTask,
|
|
TransactionalTask,
|
|
)
|
|
from celery import Task
|
|
from django.test import TestCase
|
|
|
|
|
|
class BaseTaskTest(TestCase):
|
|
"""Tests for BaseTask"""
|
|
|
|
def test_inherits_from_celery_task(self):
|
|
"""Test BaseTask inherits from Celery Task"""
|
|
self.assertTrue(issubclass(BaseTask, Task))
|
|
|
|
def test_has_default_retry_settings(self):
|
|
"""Test BaseTask has default retry settings"""
|
|
self.assertEqual(BaseTask.max_retries, 3)
|
|
self.assertTrue(BaseTask.retry_backoff)
|
|
self.assertEqual(BaseTask.retry_backoff_max, 600)
|
|
|
|
def test_acks_late_enabled(self):
|
|
"""Test acks_late is enabled"""
|
|
self.assertTrue(BaseTask.acks_late)
|
|
|
|
def test_reject_on_worker_lost(self):
|
|
"""Test reject_on_worker_lost is enabled"""
|
|
self.assertTrue(BaseTask.reject_on_worker_lost)
|
|
|
|
|
|
class TransactionalTaskTest(TestCase):
|
|
"""Tests for TransactionalTask"""
|
|
|
|
def test_inherits_from_base_task(self):
|
|
"""Test TransactionalTask inherits from BaseTask"""
|
|
self.assertTrue(issubclass(TransactionalTask, BaseTask))
|
|
|
|
|
|
class IdempotentTaskTest(TestCase):
|
|
"""Tests for IdempotentTask"""
|
|
|
|
def test_inherits_from_base_task(self):
|
|
"""Test IdempotentTask inherits from BaseTask"""
|
|
self.assertTrue(issubclass(IdempotentTask, BaseTask))
|
|
|
|
def test_has_lock_timeout(self):
|
|
"""Test IdempotentTask has lock_timeout attribute"""
|
|
self.assertEqual(IdempotentTask.lock_timeout, 3600)
|
|
|
|
|
|
class TimedTaskTest(TestCase):
|
|
"""Tests for TimedTask"""
|
|
|
|
def test_inherits_from_base_task(self):
|
|
"""Test TimedTask inherits from BaseTask"""
|
|
self.assertTrue(issubclass(TimedTask, BaseTask))
|
|
|
|
def test_has_slow_threshold(self):
|
|
"""Test TimedTask has slow_threshold attribute"""
|
|
self.assertEqual(TimedTask.slow_threshold, 60)
|
|
|
|
|
|
class PeriodicTaskTest(TestCase):
|
|
"""Tests for PeriodicTask"""
|
|
|
|
def test_inherits_from_timed_task(self):
|
|
"""Test PeriodicTask inherits from TimedTask"""
|
|
self.assertTrue(issubclass(PeriodicTask, TimedTask))
|
|
|
|
def test_max_retries_is_one(self):
|
|
"""Test max_retries is 1 for periodic tasks"""
|
|
self.assertEqual(PeriodicTask.max_retries, 1)
|
|
|
|
def test_autoretry_for_is_empty(self):
|
|
"""Test autoretry_for is empty for periodic tasks"""
|
|
self.assertEqual(PeriodicTask.autoretry_for, ())
|