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.task_tests.empty_task')

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

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

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

 def test__unicode__(self):
     proxy = tasks.background()(empty_task)
     self.assertEqual(u'TaskProxy(background_task.tests.task_tests.empty_task)',
                          unicode(proxy))

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

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

    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_specified_name(self):
     proxy = tasks.background(name='mytask')(empty_task)
     self.assertEqual(proxy.name, 'mytask')

 def test__unicode__(self):        proxy = tasks.background()(empty_task) self.assertEqual(u'TaskProxy(background_task.tests.empty_task)', unicode(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_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')

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

import unittestfrom django.test import TransactionTestCasefrom django.conf import settings
from datetime import timedelta, datetime
from background_task.tasks import tasks, TaskSchedule, TaskProxyfrom background_task.models import Task, datetime_nowfrom background_task import background
_recorded = []

def empty_task(): pass

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

class TestBackgroundDecorator(unittest.TestCase):
 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.empty_task')
        proxy = tasks.background()(record_task) self.assertEqual(proxy.name, 'background_task.tests.record_task')         proxy = tasks.background(empty_task) #print proxy self.assertTrue(isinstance(proxy, TaskProxy)) self.assertEqual(proxy.name, 'background_task.tests.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__unicode__(self):        proxy = tasks.background()(empty_task) self.assertEqual(u'TaskProxy(background_task.tests.empty_task)', unicode(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)

class TestTaskProxy(unittest.TestCase):
 def setUp(self): super(TestTaskProxy, self).setUp() self.proxy = tasks.background()(record_task)
 def test_run_task(self):        tasks.run_task(self.proxy.name, [], {}) self.assertEqual(((), {}), _recorded.pop())
        tasks.run_task(self.proxy.name, ['hi'], {}) self.assertEqual((('hi',), {}), _recorded.pop())
        tasks.run_task(self.proxy.name, [], {'kw': 1}) self.assertEqual(((), {'kw': 1}), _recorded.pop())

class TestTaskSchedule(unittest.TestCase):
 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 = datetime_now()            run_at = TaskSchedule(run_at=schedule).run_at self._within_one_second(run_at, now)
        now = datetime_now()        run_at = TaskSchedule(run_at=now).run_at self._within_one_second(run_at, now)

def background(*arg, **kw):
    from background_task.tasks import tasks
    return tasks.background(*arg, **kw)

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