retools tests

gistfile1.py

#
# doula.queue
#
from retools.queue import QueueManager


def add_job(job=None):
    p = job.redis.pipeline()
    import pdb; pdb.set_trace();


def add_result(job=None, result=None):
    p = job.redis.pipeline()


def add_failure(job=None, exc=None):
    p = job.redis.pipeline()


class Queue(object):

    def __init__(self):
        self.qm = QueueManager()
        self.qm.subscriber('job_prerun', handler='doula.queue:add_job')
        self.qm.subscriber('job_postrun', handler='doula.queue:add_result')
        self.qm.subscriber('job_failure', handler='doula.queue:add_failure')

    def this(self, job, **kwargs):
        self.qm.enqueue(job, **kwargs)

    def get(self, env, service):
        pass

#
# doula.tests.test_queue
#
import unittest
from doula.queue import Queue


def slow_function(num=0):
    if num < 0:
        raise Exception

    previous = 1
    for i in range(1, num):
        previous = previous + i
    return previous


class QueueTests(unittest.TestCase):
    def setUp(self):
        pass

    def tearDown(self):
        pass

    def test_this(self):
        q = Queue()
        q.this('doula.tests.test_queue:slow_function', num=10000000)