Tests now passing again!

celery/tests/test_backends/test_database.py

@@ -3,7 +3,6 @@ from celery.backends.database import Backend
 from celery.utils import gen_unique_id
 from celery.task import PeriodicTask
 from celery import registry
-from celery.models import PeriodicTaskMeta
 from datetime import datetime, timedelta
 
 
@@ -24,17 +23,6 @@ registry.tasks.register(MyPeriodicTask)
 
 class TestDatabaseBackend(unittest.TestCase):
 
-    def test_run_periodic_tasks(self):
-        #obj, created = PeriodicTaskMeta.objects.get_or_create(
-        #                    name=MyPeriodicTask.name,
-        #                    defaults={"last_run_at": datetime.now() -
-        #                        timedelta(days=-4)})
-        #if not created:
-        #    obj.last_run_at = datetime.now() - timedelta(days=4)
-        #    obj.save()
-        b = Backend()
-        b.run_periodic_tasks()
-
     def test_backend(self):
         b = Backend()
         tid = gen_unique_id()

celery/tests/test_models.py

@@ -1,16 +1,9 @@
 import unittest
 from datetime import datetime, timedelta
 from celery.models import TaskMeta
-from celery.task import PeriodicTask
-from celery.registry import tasks
 from celery.utils import gen_unique_id
 
 
-class TestPeriodicTask(PeriodicTask):
-    name = "celery.unittest.test_models.test_periodic_task"
-    run_every = timedelta(minutes=30)
-
-
 class TestModels(unittest.TestCase):
 
     def createTaskMeta(self):
@@ -18,11 +11,6 @@ class TestModels(unittest.TestCase):
         taskmeta, created = TaskMeta.objects.get_or_create(task_id=id)
         return taskmeta
 
-    def createPeriodicTaskMeta(self, name):
-        ptaskmeta, created = PeriodicTaskMeta.objects.get_or_create(name=name,
-                defaults={"last_run_at": datetime.now()})
-        return ptaskmeta
-
     def test_taskmeta(self):
         m1 = self.createTaskMeta()
         m2 = self.createTaskMeta()
@@ -50,17 +38,3 @@ class TestModels(unittest.TestCase):
 
         TaskMeta.objects.delete_expired()
         self.assertFalse(m1 in TaskMeta.objects.all())
-
-    def test_periodic_taskmeta(self):
-        tasks.register(TestPeriodicTask)
-        p = self.createPeriodicTaskMeta(TestPeriodicTask.name)
-        # check that repr works.
-        self.assertTrue(unicode(p).startswith("<PeriodicTask:"))
-        self.assertFalse(p in PeriodicTaskMeta.objects.get_waiting_tasks())
-        p.last_run_at = datetime.now() - (TestPeriodicTask.run_every +
-                timedelta(seconds=10))
-        p.save()
-        self.assertTrue(p in PeriodicTaskMeta.objects.get_waiting_tasks())
-        self.assertTrue(isinstance(p.task, TestPeriodicTask))
-
-        p.delay()

celery/tests/test_worker.py

@@ -9,6 +9,7 @@ from carrot.backends.base import BaseMessage
 from celery import registry
 from celery.serialization import pickle
 from celery.utils import gen_unique_id
+from celery.scheduler import Scheduler
 from datetime import datetime, timedelta
 
 
@@ -84,12 +85,12 @@ class TestAMQPListener(unittest.TestCase):
 
     def setUp(self):
         self.bucket_queue = Queue()
-        self.hold_queue = Queue()
+        self.eta_scheduler = Scheduler(self.bucket_queue)
         self.logger = get_logger()
         self.logger.setLevel(0)
 
     def test_connection(self):
-        l = AMQPListener(self.bucket_queue, self.hold_queue, self.logger)
+        l = AMQPListener(self.bucket_queue, self.eta_scheduler, self.logger)
 
         c = l.reset_connection()
         self.assertTrue(isinstance(l.amqp_connection, BrokerConnection))
@@ -106,7 +107,7 @@ class TestAMQPListener(unittest.TestCase):
         self.assertTrue(l.task_consumer is None)
 
     def test_receieve_message(self):
-        l = AMQPListener(self.bucket_queue, self.hold_queue, self.logger)
+        l = AMQPListener(self.bucket_queue, self.eta_scheduler, self.logger)
         backend = MockBackend()
         m = create_message(backend, task="c.u.foo", args=[2, 4, 8], kwargs={})
 
@@ -116,30 +117,29 @@ class TestAMQPListener(unittest.TestCase):
         self.assertTrue(isinstance(in_bucket, TaskWrapper))
         self.assertEquals(in_bucket.task_name, "c.u.foo")
         self.assertEquals(in_bucket.execute(), 2 * 4 * 8)
-        self.assertRaises(Empty, self.hold_queue.get_nowait)
+        self.assertTrue(self.eta_scheduler.empty())
 
     def test_receieve_message_not_registered(self):
-        l = AMQPListener(self.bucket_queue, self.hold_queue, self.logger)
+        l = AMQPListener(self.bucket_queue, self.eta_scheduler, self.logger)
         backend = MockBackend()
         m = create_message(backend, task="x.X.31x", args=[2, 4, 8], kwargs={})
 
         self.assertFalse(l.receive_message(m.decode(), m))
         self.assertRaises(Empty, self.bucket_queue.get_nowait)
-        self.assertRaises(Empty, self.hold_queue.get_nowait)
+        self.assertTrue(self.eta_scheduler.empty())
 
     def test_receieve_message_eta(self):
-        l = AMQPListener(self.bucket_queue, self.hold_queue, self.logger)
+        l = AMQPListener(self.bucket_queue, self.eta_scheduler, self.logger)
         backend = MockBackend()
         m = create_message(backend, task="c.u.foo", args=[2, 4, 8], kwargs={},
                            eta=datetime.now() + timedelta(days=1))
 
         l.receive_message(m.decode(), m)
 
-        in_hold = self.hold_queue.get_nowait()
-        self.assertEquals(len(in_hold), 3)
-        task, eta, on_accept = in_hold
+        in_hold = self.eta_scheduler.queue[0]
+        self.assertEquals(len(in_hold), 4)
+        eta, priority, task, on_accept = in_hold
         self.assertTrue(isinstance(task, TaskWrapper))
-        self.assertTrue(isinstance(eta, datetime))
         self.assertTrue(callable(on_accept))
         self.assertEquals(task.task_name, "c.u.foo")
         self.assertEquals(task.execute(), 2 * 4 * 8)
@@ -156,8 +156,8 @@ class TestWorkController(unittest.TestCase):
     def test_attrs(self):
         worker = self.worker
         self.assertTrue(isinstance(worker.bucket_queue, Queue))
-        self.assertTrue(isinstance(worker.hold_queue, Queue))
-        self.assertTrue(worker.periodic_work_controller)
+        self.assertTrue(isinstance(worker.eta_scheduler, Scheduler))
+        self.assertTrue(worker.schedule_controller)
         self.assertTrue(worker.pool)
         self.assertTrue(worker.amqp_listener)
         self.assertTrue(worker.mediator)

celery/tests/test_worker_controllers.py

@@ -4,7 +4,7 @@ import multiprocessing
 from Queue import Queue, Empty
 from datetime import datetime, timedelta
 
-from celery.worker.controllers import Mediator, PeriodicWorkController
+from celery.worker.controllers import Mediator
 from celery.worker.controllers import BackgroundThread
 
 
@@ -71,45 +71,3 @@ class TestMediator(unittest.TestCase):
         m.on_iteration()
 
         self.assertEquals(got["value"], "George Constanza")
-
-
-class TestPeriodicWorkController(unittest.TestCase):
-
-    def test_process_hold_queue(self):
-        bucket_queue = Queue()
-        hold_queue = Queue()
-        m = PeriodicWorkController(bucket_queue, hold_queue)
-        m.process_hold_queue()
-
-        scratchpad = {}
-
-        def on_accept():
-            scratchpad["accepted"] = True
-
-        hold_queue.put((MockTask("task1"),
-                        datetime.now() - timedelta(days=1),
-                        on_accept))
-
-        m.process_hold_queue()
-        self.assertRaises(Empty, hold_queue.get_nowait)
-        self.assertTrue(scratchpad.get("accepted"))
-        self.assertEquals(bucket_queue.get_nowait().value, "task1")
-        tomorrow = datetime.now() + timedelta(days=1)
-        hold_queue.put((MockTask("task2"), tomorrow, on_accept))
-        m.process_hold_queue()
-        self.assertRaises(Empty, bucket_queue.get_nowait)
-        value, eta, on_accept = hold_queue.get_nowait()
-        self.assertEquals(value.value, "task2")
-        self.assertEquals(eta, tomorrow)
-
-    def test_run_periodic_tasks(self):
-        bucket_queue = Queue()
-        hold_queue = Queue()
-        m = PeriodicWorkController(bucket_queue, hold_queue)
-        m.run_periodic_tasks()
-
-    def test_on_iteration(self):
-        bucket_queue = Queue()
-        hold_queue = Queue()
-        m = PeriodicWorkController(bucket_queue, hold_queue)
-        m.on_iteration()