labelbase/background_task/tests/test_tasks.py

# -*- coding: utf-8 -*-
import time
from datetime import timedelta, datetime
from mock import patch, Mock

from django.db.utils import OperationalError
from django.contrib.auth.models import User
from django.test import override_settings
from django.test.testcases import TransactionTestCase
from django.conf import settings
from django.utils import timezone

from background_task.exceptions import InvalidTaskError
from background_task.tasks import tasks, TaskSchedule, TaskProxy
from background_task.models import Task
from background_task.models import CompletedTask
from background_task import background
from background_task.settings import app_settings

_recorded = []


def mocked_run_task(name, args=None, kwargs=None):
    """
    We mock tasks.run_task to give other threads some time to update the database.

    Otherwise we run into a locked database.
    """
    val = tasks.run_task(name, args, kwargs)
    if app_settings.BACKGROUND_TASK_RUN_ASYNC:
        time.sleep(1)
    return val


def mocked_run_next_task(queue=None):
    """
    We mock tasks.mocked_run_next_task to give other threads some time to update the database.

    Otherwise we run into a locked database.
    """
    val = tasks.run_next_task(queue)
    if app_settings.BACKGROUND_TASK_RUN_ASYNC:
        time.sleep(1)
    return val

run_task = mocked_run_task
run_next_task = mocked_run_next_task


def empty_task():
    pass


def record_task(*arg, **kw):
    _recorded.append((arg, kw))


class TestBackgroundDecorator(TransactionTestCase):

    def test_get_proxy(self):
        proxy = tasks.background()(empty_task)
        self.assertNotEqual(proxy, empty_task)
        self.assertTrue(isinstance(proxy, TaskProxy))

        # and alternate form
        proxy = tasks.background(empty_task)
        self.assertNotEqual(proxy, empty_task)
        self.assertTrue(isinstance(proxy, TaskProxy))

    def test_default_name(self):
        proxy = tasks.background()(empty_task)
        self.assertEqual(proxy.name, 'background_task.tests.test_tasks.empty_task')

        proxy = tasks.background()(record_task)
        self.assertEqual(proxy.name, 'background_task.tests.test_tasks.record_task')

        proxy = tasks.background(empty_task)
        # print proxy
        self.assertTrue(isinstance(proxy, TaskProxy))
        self.assertEqual(proxy.name, 'background_task.tests.test_tasks.empty_task')

    def test_specified_name(self):
        proxy = tasks.background(name='mytask')(empty_task)
        self.assertEqual(proxy.name, 'mytask')

    def test_task_function(self):
        proxy = tasks.background()(empty_task)
        self.assertEqual(proxy.task_function, empty_task)

        proxy = tasks.background()(record_task)
        self.assertEqual(proxy.task_function, record_task)

    def test_default_schedule(self):
        proxy = tasks.background()(empty_task)
        self.assertEqual(TaskSchedule(), proxy.schedule)

    def test_schedule(self):
        proxy = tasks.background(schedule=10)(empty_task)
        self.assertEqual(TaskSchedule(run_at=10), proxy.schedule)

    def test_str(self):
        proxy = tasks.background()(empty_task)
        self.assertEqual(
            u'TaskProxy(background_task.tests.test_tasks.empty_task)',
            str(proxy)
        )

    def test_shortcut(self):
        '''check shortcut to decorator works'''
        proxy = background()(empty_task)
        self.failIfEqual(proxy, empty_task)
        self.assertEqual(proxy.task_function, empty_task)

    def test_launch_sync(self):
        ''' Check launch original function in synchronous mode '''
        @background
        def add(x, y):
            return x + y

        t = Task.objects.count()
        ct = CompletedTask.objects.count()
        answer = add.now(2, 3)
        self.assertEqual(answer, 5)
        self.assertEqual(Task.objects.count(), t, 'Task was created')
        self.assertEqual(CompletedTask.objects.count(), ct, 'Completed task was created')


class TestTaskProxy(TransactionTestCase):

    def setUp(self):
        super(TestTaskProxy, self).setUp()
        self.proxy = tasks.background()(record_task)

    def test_run_task(self):
        run_task(self.proxy.name, [], {})
        self.assertEqual(((), {}), _recorded.pop())

        run_task(self.proxy.name, ['hi'], {})
        self.assertEqual((('hi',), {}), _recorded.pop())

        run_task(self.proxy.name, [], {'kw': 1})
        self.assertEqual(((), {'kw': 1}), _recorded.pop())


class TestTaskSchedule(TransactionTestCase):

    def test_priority(self):
        self.assertEqual(0, TaskSchedule().priority)
        self.assertEqual(0, TaskSchedule(priority=0).priority)
        self.assertEqual(1, TaskSchedule(priority=1).priority)
        self.assertEqual(2, TaskSchedule(priority=2).priority)

    def _within_one_second(self, d1, d2):
        self.failUnless(isinstance(d1, datetime))
        self.failUnless(isinstance(d2, datetime))
        self.failUnless(abs(d1 - d2) <= timedelta(seconds=1))

    def test_run_at(self):
        for schedule in [None, 0, timedelta(seconds=0)]:
            now = timezone.now()
            run_at = TaskSchedule(run_at=schedule).run_at
            self._within_one_second(run_at, now)

        now = timezone.now()
        run_at = TaskSchedule(run_at=now).run_at
        self._within_one_second(run_at, now)

        fixed_dt = timezone.now() + timedelta(seconds=60)
        run_at = TaskSchedule(run_at=fixed_dt).run_at
        self._within_one_second(run_at, fixed_dt)

        run_at = TaskSchedule(run_at=90).run_at
        self._within_one_second(run_at, timezone.now() + timedelta(seconds=90))

        run_at = TaskSchedule(run_at=timedelta(seconds=35)).run_at
        self._within_one_second(run_at, timezone.now() + timedelta(seconds=35))

    def test_create(self):
        fixed_dt = timezone.now() + timedelta(seconds=10)
        schedule = TaskSchedule.create({'run_at': fixed_dt})
        self.assertEqual(schedule.run_at, fixed_dt)
        self.assertEqual(0, schedule.priority)
        self.assertEqual(TaskSchedule.SCHEDULE, schedule.action)

        schedule = {'run_at': fixed_dt, 'priority': 2,
                    'action': TaskSchedule.RESCHEDULE_EXISTING}
        schedule = TaskSchedule.create(schedule)
        self.assertEqual(schedule.run_at, fixed_dt)
        self.assertEqual(2, schedule.priority)
        self.assertEqual(TaskSchedule.RESCHEDULE_EXISTING, schedule.action)

        schedule = TaskSchedule.create(0)
        self._within_one_second(schedule.run_at, timezone.now())

        schedule = TaskSchedule.create(10)
        self._within_one_second(schedule.run_at,
                                timezone.now() + timedelta(seconds=10))

        schedule = TaskSchedule.create(TaskSchedule(run_at=fixed_dt))
        self.assertEqual(schedule.run_at, fixed_dt)
        self.assertEqual(0, schedule.priority)
        self.assertEqual(TaskSchedule.SCHEDULE, schedule.action)

    def test_merge(self):
        default = TaskSchedule(run_at=10, priority=2,
                               action=TaskSchedule.RESCHEDULE_EXISTING)
        schedule = TaskSchedule.create(20).merge(default)

        self._within_one_second(timezone.now() + timedelta(seconds=20),
                                schedule.run_at)
        self.assertEqual(2, schedule.priority)
        self.assertEqual(TaskSchedule.RESCHEDULE_EXISTING, schedule.action)

        schedule = TaskSchedule.create({'priority': 0}).merge(default)
        self._within_one_second(timezone.now() + timedelta(seconds=10),
                                schedule.run_at)
        self.assertEqual(0, schedule.priority)
        self.assertEqual(TaskSchedule.RESCHEDULE_EXISTING, schedule.action)

        action = TaskSchedule.CHECK_EXISTING
        schedule = TaskSchedule.create({'action': action}).merge(default)
        self._within_one_second(timezone.now() + timedelta(seconds=10),
                                schedule.run_at)
        self.assertEqual(2, schedule.priority)
        self.assertEqual(action, schedule.action)

    def test_repr(self):
        self.assertEqual('TaskSchedule(run_at=10, priority=0)',
                            repr(TaskSchedule(run_at=10, priority=0)))


class TestSchedulingTasks(TransactionTestCase):

    def test_background_gets_scheduled(self):
        self.result = None

        @tasks.background(name='test_background_gets_scheduled')
        def set_result(result):
            self.result = result

        # calling set_result should now actually create a record in the db
        set_result(1)

        all_tasks = Task.objects.all()
        self.assertEqual(1, all_tasks.count())
        task = all_tasks[0]
        self.assertEqual('test_background_gets_scheduled', task.task_name)
        self.assertEqual('[[1], {}]', task.task_params)

    def test_reschedule_existing(self):

        reschedule_existing = TaskSchedule.RESCHEDULE_EXISTING

        @tasks.background(name='test_reschedule_existing',
                         schedule=TaskSchedule(action=reschedule_existing))
        def reschedule_fn():
            pass

        # this should only end up with one task
        # and it should be scheduled for the later time
        reschedule_fn()
        reschedule_fn(schedule=90)

        all_tasks = Task.objects.all()
        self.assertEqual(1, all_tasks.count())
        task = all_tasks[0]
        self.assertEqual('test_reschedule_existing', task.task_name)

        # check task is scheduled for later on
        now = timezone.now()
        self.failUnless(now + timedelta(seconds=89) < task.run_at)
        self.failUnless(now + timedelta(seconds=91) > task.run_at)

    def test_check_existing(self):

        check_existing = TaskSchedule.CHECK_EXISTING

        @tasks.background(name='test_check_existing',
                         schedule=TaskSchedule(action=check_existing))
        def check_fn():
            pass

        # this should only end up with the first call
        # scheduled
        check_fn()
        check_fn(schedule=90)

        all_tasks = Task.objects.all()
        self.assertEqual(1, all_tasks.count())
        task = all_tasks[0]
        self.assertEqual('test_check_existing', task.task_name)

        # check new task is scheduled for the earlier time
        now = timezone.now()
        self.failUnless(now - timedelta(seconds=1) < task.run_at)
        self.failUnless(now + timedelta(seconds=1) > task.run_at)


class TestTaskRunner(TransactionTestCase):

    def setUp(self):
        super(TestTaskRunner, self).setUp()
        self.runner = tasks._runner

    def test_get_task_to_run_no_tasks(self):
        self.failIf(self.runner.get_task_to_run(tasks))

    def test_get_task_to_run(self):
        task = Task.objects.new_task('mytask', (1), {})
        task.save()
        self.failUnless(task.locked_by is None)
        self.failUnless(task.locked_at is None)

        locked_task = self.runner.get_task_to_run(tasks)
        self.failIf(locked_task is None)
        self.failIf(locked_task.locked_by is None)
        self.assertEqual(self.runner.worker_name, locked_task.locked_by)
        self.failIf(locked_task.locked_at is None)
        self.assertEqual('mytask', locked_task.task_name)


class TestTaskModel(TransactionTestCase):

    def test_lock_uncontested(self):
        task = Task.objects.new_task('mytask')
        task.save()
        self.failUnless(task.locked_by is None)
        self.failUnless(task.locked_at is None)

        locked_task = task.lock('mylock')
        self.assertEqual('mylock', locked_task.locked_by)
        self.failIf(locked_task.locked_at is None)
        self.assertEqual(task.pk, locked_task.pk)

    def test_lock_contested(self):
        # locking should actually look at db, not object
        # in memory
        task = Task.objects.new_task('mytask')
        task.save()
        self.failIf(task.lock('mylock') is None)

        self.failUnless(task.lock('otherlock') is None)

    def test_lock_expired(self):
        task = Task.objects.new_task('mytask')
        task.save()
        locked_task = task.lock('mylock')

        # force expire the lock
        expire_by = timedelta(seconds=(app_settings.BACKGROUND_TASK_MAX_RUN_TIME + 2))
        locked_task.locked_at = locked_task.locked_at - expire_by
        locked_task.save()

        # now try to get the lock again
        self.failIf(task.lock('otherlock') is None)

    def test_str(self):
        task = Task.objects.new_task('mytask')
        self.assertEqual(u'mytask', str(task))
        task = Task.objects.new_task('mytask', verbose_name="My Task")
        self.assertEqual(u'My Task', str(task))

    def test_creator(self):
        user = User.objects.create_user(username='bob', email='bob@example.com', password='12345')
        task = Task.objects.new_task('mytask', creator=user)
        task.save()
        self.assertEqual(task.creator, user)

    def test_repeat(self):
        repeat_until = timezone.now() + timedelta(days=1)
        task = Task.objects.new_task('mytask', repeat=Task.HOURLY, repeat_until=repeat_until)
        task.save()
        self.assertEqual(task.repeat, Task.HOURLY)
        self.assertEqual(task.repeat_until, repeat_until)

    def test_create_completed_task(self):
        task = Task.objects.new_task(
            task_name='mytask',
            args=[1],
            kwargs={'q': 's'},
            priority=1,
            queue='myqueue',
            verbose_name='My Task',
            creator=User.objects.create_user(username='bob', email='bob@example.com', password='12345'),
        )
        task.save()
        completed_task = task.create_completed_task()
        self.assertEqual(completed_task.task_name, task.task_name)
        self.assertEqual(completed_task.task_params, task.task_params)
        self.assertEqual(completed_task.priority, task.priority)
        self.assertEqual(completed_task.queue, task.queue)
        self.assertEqual(completed_task.verbose_name, task.verbose_name)
        self.assertEqual(completed_task.creator, task.creator)
        self.assertEqual(completed_task.repeat, task.repeat)
        self.assertEqual(completed_task.repeat_until, task.repeat_until)


class TestTasks(TransactionTestCase):

    def setUp(self):
        super(TestTasks, self).setUp()

        @tasks.background(name='set_fields')
        def set_fields(**fields):
            for key, value in fields.items():
                setattr(self, key, value)

        @tasks.background(name='throws_error')
        def throws_error():
            raise RuntimeError("an error")

        self.set_fields = set_fields
        self.throws_error = throws_error

    def test_run_next_task_nothing_scheduled(self):
        self.failIf(run_next_task())

    def test_run_next_task_one_task_scheduled(self):
        self.set_fields(worked=True)
        self.failIf(hasattr(self, 'worked'))

        self.failUnless(run_next_task())

        self.failUnless(hasattr(self, 'worked'))
        self.failUnless(self.worked)

    def test_run_next_task_several_tasks_scheduled(self):
        self.set_fields(one='1')
        self.set_fields(two='2')
        self.set_fields(three='3')

        for i in range(3):
            self.failUnless(run_next_task())

        self.failIf(run_next_task())  # everything should have been run

        for field, value in [('one', '1'), ('two', '2'), ('three', '3')]:
            self.failUnless(hasattr(self, field))
            self.assertEqual(value, getattr(self, field))

    def test_run_next_task_error_handling(self):
        self.throws_error()

        all_tasks = Task.objects.all()
        self.assertEqual(1, all_tasks.count())
        original_task = all_tasks[0]

        # should run, but trigger error
        self.failUnless(run_next_task())

        all_tasks = Task.objects.all()
        self.assertEqual(1, all_tasks.count())

        failed_task = all_tasks[0]
        # should have an error recorded
        self.failIfEqual('', failed_task.last_error)
        self.failUnless(failed_task.failed_at is None)
        self.assertEqual(1, failed_task.attempts)

        # should have been rescheduled for the future
        # and no longer locked
        self.failUnless(failed_task.run_at > original_task.run_at)
        self.failUnless(failed_task.locked_by is None)
        self.failUnless(failed_task.locked_at is None)

    def test_run_next_task_does_not_run_locked(self):
        self.set_fields(locked=True)
        self.failIf(hasattr(self, 'locked'))

        all_tasks = Task.objects.all()
        self.assertEqual(1, all_tasks.count())
        original_task = all_tasks[0]
        original_task.lock('lockname')

        self.failIf(run_next_task())

        self.failIf(hasattr(self, 'locked'))
        all_tasks = Task.objects.all()
        self.assertEqual(1, all_tasks.count())

    def test_run_next_task_unlocks_after_MAX_RUN_TIME(self):
        self.set_fields(lock_overridden=True)

        all_tasks = Task.objects.all()
        self.assertEqual(1, all_tasks.count())
        original_task = all_tasks[0]
        locked_task = original_task.lock('lockname')

        self.failIf(run_next_task())

        self.failIf(hasattr(self, 'lock_overridden'))

        # put lot time into past
        expire_by = timedelta(seconds=(app_settings.BACKGROUND_TASK_MAX_RUN_TIME + 2))
        locked_task.locked_at = locked_task.locked_at - expire_by
        locked_task.save()

        # so now we should be able to override the lock
        # and run the task
        self.failUnless(run_next_task())
        self.assertEqual(0, Task.objects.count())

        self.failUnless(hasattr(self, 'lock_overridden'))
        self.failUnless(self.lock_overridden)

    def test_default_schedule_used_for_run_at(self):

        @tasks.background(name='default_schedule_used_for_run_at', schedule=60)
        def default_schedule_used_for_time():
            pass

        now = timezone.now()
        default_schedule_used_for_time()

        all_tasks = Task.objects.all()
        self.assertEqual(1, all_tasks.count())
        task = all_tasks[0]

        self.failUnless(now < task.run_at)
        self.failUnless((task.run_at - now) <= timedelta(seconds=61))
        self.failUnless((task.run_at - now) >= timedelta(seconds=59))

    def test_default_schedule_used_for_priority(self):

        @tasks.background(name='default_schedule_used_for_priority',
                          schedule={'priority': 2})
        def default_schedule_used_for_priority():
            pass

        default_schedule_used_for_priority()

        all_tasks = Task.objects.all()
        self.assertEqual(1, all_tasks.count())
        task = all_tasks[0]
        self.assertEqual(2, task.priority)

    def test_non_default_schedule_used(self):
        default_run_at = timezone.now() + timedelta(seconds=90)

        @tasks.background(name='non_default_schedule_used',
                          schedule={'run_at': default_run_at, 'priority': 2})
        def default_schedule_used_for_priority():
            pass

        run_at = timezone.now().replace(microsecond=0) + timedelta(seconds=60)
        default_schedule_used_for_priority(schedule=run_at)

        all_tasks = Task.objects.all()
        self.assertEqual(1, all_tasks.count())
        task = all_tasks[0]
        self.assertEqual(run_at, task.run_at)

    def test_failed_at_set_after_MAX_ATTEMPTS(self):
        @tasks.background(name='test_failed_at_set_after_MAX_ATTEMPTS')
        def failed_at_set_after_MAX_ATTEMPTS():
            raise RuntimeError('failed')

        failed_at_set_after_MAX_ATTEMPTS()

        available = Task.objects.find_available()
        self.assertEqual(1, available.count())
        task = available[0]

        self.failUnless(task.failed_at is None)

        task.attempts = app_settings.BACKGROUND_TASK_MAX_ATTEMPTS
        task.save()

        # task should be scheduled to run now
        # but will be marked as failed straight away
        self.failUnless(run_next_task())

        available = Task.objects.find_available()
        self.assertEqual(0, available.count())

        all_tasks = Task.objects.all()
        self.assertEqual(0, all_tasks.count())
        self.assertEqual(1, CompletedTask.objects.count())
        completed_task = CompletedTask.objects.all()[0]
        self.failIf(completed_task.failed_at is None)

    def test_run_task_return_value(self):
        return_value = self.set_fields(test='test')
        self.assertEqual(Task.objects.count(), 1)
        task = Task.objects.first()
        self.assertEqual(return_value, task)
        self.assertEqual(return_value.pk, task.pk)

    def test_verbose_name_param(self):
        verbose_name = 'My Task'
        task = self.set_fields(test='test1', verbose_name=verbose_name)
        self.assertEqual(task.verbose_name, verbose_name)

    def test_creator_param(self):
        user = User.objects.create_user(username='bob', email='bob@example.com', password='12345')
        task = self.set_fields(test='test2', creator=user)
        self.assertEqual(task.creator, user)


class MaxAttemptsTestCase(TransactionTestCase):

    def setUp(self):
        @tasks.background(name='failing task')
        def failing_task():
            raise Exception("error")
            # return 0 / 0
        self.failing_task = failing_task
        self.task1 = self.failing_task()
        self.task2 = self.failing_task()
        self.task1_id = self.task1.id
        self.task2_id = self.task2.id

    @override_settings(MAX_ATTEMPTS=1)
    def test_max_attempts_one(self):
        self.assertEqual(settings.MAX_ATTEMPTS, 1)
        self.assertEqual(Task.objects.count(), 2)

        run_next_task()
        self.assertEqual(Task.objects.count(), 1)
        self.assertEqual(Task.objects.all()[0].id, self.task2_id)
        self.assertEqual(CompletedTask.objects.count(), 1)
        completed_task = CompletedTask.objects.all()[0]
        self.assertEqual(completed_task.attempts, 1)
        self.assertEqual(completed_task.task_name, self.task1.task_name)
        self.assertEqual(completed_task.task_params, self.task1.task_params)
        self.assertIsNotNone(completed_task.last_error)
        self.assertIsNotNone(completed_task.failed_at)

        run_next_task()
        self.assertEqual(Task.objects.count(), 0)
        self.assertEqual(CompletedTask.objects.count(), 2)

    @override_settings(MAX_ATTEMPTS=2)
    def test_max_attempts_two(self):
        self.assertEqual(settings.MAX_ATTEMPTS, 2)
        run_next_task()
        self.assertEqual(Task.objects.count(), 2)
        self.assertEqual(CompletedTask.objects.count(), 0)


class InvalidTaskTestCase(TransactionTestCase):

    class SomeInvalidTaskError(InvalidTaskError):
        pass

    def setUp(self):
        @tasks.background(name='failing task')
        def failing_task():
            raise self.SomeInvalidTaskError("invalid")

        self.failing_task = failing_task
        self.task1 = self.failing_task()
        self.task1_id = self.task1.id

    @override_settings(MAX_ATTEMPTS=2)
    def test_invalid_task(self):
        self.assertEqual(settings.MAX_ATTEMPTS, 2)
        run_next_task()
        self.assertEqual(Task.objects.count(), 0)
        self.assertEqual(CompletedTask.objects.count(), 1)


class ArgumentsWithDictTestCase(TransactionTestCase):
    def setUp(self):
        @tasks.background(name='failing task')
        def task(d):
            pass
        self.task = task

    def test_task_with_dictionary_in_args(self):
        self.assertEqual(Task.objects.count(), 0)
        d = {22222: 2, 11111: 1}
        self.task(d)
        self.assertEqual(Task.objects.count(), 1)
        run_next_task()
        self.assertEqual(Task.objects.count(), 0)


completed_named_queue_tasks = []


@background(queue='named_queue')
def named_queue_task(message):
    completed_named_queue_tasks.append(message)


class NamedQueueTestCase(TransactionTestCase):

    def test_process_queue(self):
        named_queue_task('test1')
        run_next_task(queue='named_queue')
        self.assertIn('test1', completed_named_queue_tasks, msg='Task should be processed')

    def test_process_all_tasks(self):
        named_queue_task('test2')
        run_next_task()
        self.assertIn('test2', completed_named_queue_tasks, msg='Task should be processed')

    def test_process_other_queue(self):
        named_queue_task('test3')
        run_next_task(queue='other_named_queue')
        self.assertNotIn('test3', completed_named_queue_tasks, msg='Task should be ignored')
        run_next_task()


class RepetitionTestCase(TransactionTestCase):

    def setUp(self):
        @tasks.background()
        def my_task(*args, **kwargs):
            pass
        self.my_task = my_task

    def test_repeat(self):
        repeat_until = timezone.now() + timedelta(weeks=1)
        old_task = self.my_task(
            'test-repeat',
            foo='bar',
            repeat=Task.HOURLY,
            repeat_until=repeat_until,
            verbose_name="Test repeat",
        )
        self.assertEqual(old_task.repeat, Task.HOURLY)
        self.assertEqual(old_task.repeat_until, repeat_until)
        tasks.run_next_task()
        time.sleep(0.5)

        self.assertEqual(Task.objects.filter(repeat=Task.HOURLY).count(), 1)
        new_task = Task.objects.get(repeat=Task.HOURLY)
        self.assertNotEqual(new_task.id, old_task.id)
        self.assertEqual(new_task.task_name, old_task.task_name)
        self.assertEqual(new_task.params(), old_task.params())
        self.assertEqual(new_task.task_hash, old_task.task_hash)
        self.assertEqual(new_task.verbose_name, old_task.verbose_name)
        self.assertEqual((new_task.run_at - old_task.run_at), timedelta(hours=1))
        self.assertEqual(new_task.repeat_until, old_task.repeat_until)

    def test_repetition_in_future(self):
        repeat_until = timezone.now() + timedelta(weeks=1)
        old_task = self.my_task(
            'test-repetition',
            repeat=Task.HOURLY,
            repeat_until=repeat_until,
            verbose_name="Test repetition in future",
        )
        old_task.run_at = timezone.now() - timedelta(weeks=1)  # task is one week old
        old_task.save()
        tasks.run_next_task()
        time.sleep(0.5)

        self.assertEqual(Task.objects.filter(repeat=Task.HOURLY).count(), 1)
        new_task = Task.objects.get(repeat=Task.HOURLY)
        self.assertNotEqual(new_task.id, old_task.id)
        # new task skipped exactly one week of downtime in the past, keeps period
        self.assertEqual((new_task.run_at - old_task.run_at), timedelta(weeks=1, hours=1))
        # new task will be executed in the future
        self.assertTrue(new_task.run_at > timezone.now())
        # new task will be executed in less than one hour
        self.assertTrue((new_task.run_at - timezone.now()) <= timedelta(hours=1))


class QuerySetManagerTestCase(TransactionTestCase):

    def setUp(self):
        @tasks.background()
        def succeeding_task():
            return 0/1

        @tasks.background()
        def failing_task():
            return 0/0

        self.user1 = User.objects.create_user(username='bob', email='bob@example.com', password='12345')
        self.user2 = User.objects.create_user(username='bob2', email='bob@example.com', password='12345')
        self.task_all = succeeding_task()
        self.task_user = succeeding_task(creator=self.user1)
        self.failing_task_all = failing_task()
        self.failing_task_user = failing_task(creator=self.user1)

    @override_settings(MAX_ATTEMPTS=1)
    def test_task_manager(self):
        self.assertEqual(len(Task.objects.all()), 4)
        self.assertEqual(len(Task.objects.created_by(self.user1)), 2)
        self.assertEqual(len(Task.objects.created_by(self.user2)), 0)
        for i in range(4):
            run_next_task()
        self.assertEqual(len(Task.objects.all()), 0)
        self.assertEqual(len(Task.objects.created_by(self.user1)), 0)
        self.assertEqual(len(Task.objects.created_by(self.user2)), 0)

    @override_settings(MAX_ATTEMPTS=1)
    def test_completed_task_manager(self):
        self.assertEqual(len(CompletedTask.objects.created_by(self.user1)), 0)
        self.assertEqual(len(CompletedTask.objects.created_by(self.user2)), 0)
        self.assertEqual(len(CompletedTask.objects.failed()), 0)
        self.assertEqual(len(CompletedTask.objects.created_by(self.user1).failed()), 0)
        self.assertEqual(len(CompletedTask.objects.failed(within=timedelta(hours=1))), 0)
        self.assertEqual(len(CompletedTask.objects.succeeded()), 0)
        self.assertEqual(len(CompletedTask.objects.created_by(self.user1).succeeded()), 0)
        self.assertEqual(len(CompletedTask.objects.succeeded(within=timedelta(hours=1))), 0)
        for i in range(4):
            run_next_task()
        self.assertEqual(len(CompletedTask.objects.created_by(self.user1)), 2)
        self.assertEqual(len(CompletedTask.objects.created_by(self.user2)), 0)
        self.assertEqual(len(CompletedTask.objects.failed()), 2)
        self.assertEqual(len(CompletedTask.objects.created_by(self.user1).failed()), 1)
        self.assertEqual(len(CompletedTask.objects.failed(within=timedelta(hours=1))), 2)
        self.assertEqual(len(CompletedTask.objects.succeeded()), 2)
        self.assertEqual(len(CompletedTask.objects.created_by(self.user1).succeeded()), 1)
        self.assertEqual(len(CompletedTask.objects.succeeded(within=timedelta(hours=1))), 2)


class PriorityTestCase(TransactionTestCase):

    def setUp(self):
        @tasks.background()
        def mytask():
            pass

        run_at = timezone.now() - timedelta(minutes=1)

        self.high_priority_task = mytask(priority=99, schedule=run_at)
        self.low_priority_task = mytask(priority=-1, schedule=run_at)

    def test_priority(self):
        self.assertEqual(self.high_priority_task.priority, 99)
        self.assertEqual(self.low_priority_task.priority, -1)

        available = Task.objects.find_available()
        self.assertEqual(available.count(), 2)
        self.assertEqual(available.first(), self.high_priority_task)
        # Using a list here. QuerySet.last() is prohibited after slicing (new in Django 2.0)
        self.assertEqual(list(available)[-1], self.low_priority_task)

        self.assertFalse(CompletedTask.objects.filter(priority=self.high_priority_task.priority).exists())
        self.assertFalse(CompletedTask.objects.filter(priority=self.low_priority_task.priority).exists())
        run_next_task()
        self.assertTrue(CompletedTask.objects.filter(priority=self.high_priority_task.priority).exists())
        self.assertFalse(CompletedTask.objects.filter(priority=self.low_priority_task.priority).exists())
        run_next_task()
        self.assertTrue(CompletedTask.objects.filter(priority=self.high_priority_task.priority).exists())
        self.assertTrue(CompletedTask.objects.filter(priority=self.low_priority_task.priority).exists())


class LoggingTestCase(TransactionTestCase):

    def setUp(self):
        @tasks.background()
        def succeeding_task():
            return 0/1

        @tasks.background()
        def failing_task():
            return 0/0

        self.succeeding_task = succeeding_task
        self.failing_task = failing_task

    @patch('background_task.tasks.logger')
    def test_success_logging(self, mock_logger):
        self.succeeding_task()
        run_next_task()
        self.assertFalse(mock_logger.warning.called)
        self.assertFalse(mock_logger.error.called)
        self.assertFalse(mock_logger.critical.called)

    @patch('background_task.tasks.logger')
    def test_error_logging(self, mock_logger):
        self.failing_task()
        run_next_task()
        self.assertFalse(mock_logger.warning.called)
        self.assertTrue(mock_logger.error.called)
        self.assertFalse(mock_logger.critical.called)


class DatabaseOutageTestCase(TransactionTestCase):

    def setUp(self):
        @tasks.background()
        def my_task(*args, **kwargs):
            pass
        self.my_task = my_task

    @patch('background_task.tasks.logger')
    def test_dropped_db_connection(self, mock_logger):
        self.my_task() # Entered into database successfully

        cursor_wrapper = Mock()
        cursor_wrapper.side_effect = OperationalError

        # Force django cursor to throw OperationalError as if connection were dropped
        with patch("django.db.backends.utils.CursorWrapper", cursor_wrapper) as patched_method:
            run_next_task()

        self.assertTrue(mock_logger.warning.called)
        self.assertFalse(mock_logger.error.called)
        self.assertFalse(mock_logger.critical.called)