Test django-rq about django-rq HOT 12 CLOSED

@job
def send_async_mail(connection=None):
    send_mail('foo', 'bar', '[email protected]', ['[email protected]'], fail_silently=False, connection=connection)


def send_custom_mail():
     send_async_mail.delay(connection=get_connection())

# in jobs.py for example

@job
def send_async_signal():
    print my_custom_signal.receivers  # I got the one created below by mock_signal_receiver
    print my_custom_signal.receivers[0][1]().call_count  # I have 0, which is right
    my_custom_signal.send(sender=Foo)
    print my_custom_signal.receivers[0][1]().call_count  # I have 1, which is right too, and 'should' be the case at the end of my testcase (see below). 

def send_custom_signal():
     send_async_signal.delay()

# in tests.py

from django_rq import get_worker
from mock_django import mock_signal_receiver
from jobs import send_custom_signal

class AsyncSendSignalTestCase(unittest.TestCase):
    def test_async_signal(self):
        with mock_signal_receiver(my_custom_signal) as receiver:
            print my_custom_signal.receivers  # I have exactly the same receiver as the first print in send_sync_signal()
            send_custom_signal()
            print receiver.call_count  # I get 0, and it's right, because the signal is sent in an asynchronous job that hasn't been yet treated.
            get_worker().work(burst=True)
            print receiver.call_count  # I get 0, and not 1 as I would have expected.
            print my_custom_signal.receivers[0][1]().call_count  # I also get 0 here (in send_async_signal, the value was 1)

# in jobs.py for example

@job
def send_async_signal():
    my_custom_signal.send(sender=Foo)

def send_custom_signal():
     send_async_signal.delay()

# in tests.py

from django_rq import get_worker
from django.dispatch import receiver
from jobs import send_custom_signal


def count_calls(fn):
    def _counting(*args, **kwargs):
        _counting.calls += 1
        return fn(*args, **kwargs)
    _counting.calls = 0
    return _counting

@receiver(my_custom_signal)
@count_calls
def my_custom_signal_callback(sender, **kwargs):
    print 'well, callback reached'

class AsyncSendSignalTestCase(unittest.TestCase):
    def test_async_signal(self):
            send_custom_signal()
            print my_custom_signal_callback.calls  # 0, which is right
            get_worker().work(burst=True)
            print my_custom_signal_callback.calls  # 0, which is wrong, it should return 1 

def long_running_function(*args, **kwargs):
    # do stuff

my_custom_signal.connect(long_running_function)

@job
def send_async_signal():
    my_custom_signal.send(sender=Foo)

def send_custom_signal():
     send_async_signal.delay()

class AsyncSendSignalTestCase(unittest.TestCase):
    def test_async_signal(self):
            send_custom_signal()
            print my_custom_signal_callback.calls  # 0, which is right
            get_worker().work(burst=True)
            print my_custom_signal_callback.calls  # 0, which is wrong, it should return 1 

@job
def long_running_function(*args, **kwargs):
    # do stuff


class MyTestCase(unittest.TestCase):
    def test_async_job(self):
            long_running_function.delay()
            get_worker().work(burst=True)
            # Check that work gets done

from django_rq import job
from django.dispatch import Signal

from mock_django import mock_signal_receiver

# set up a Signal
bar = Signal()


# the async function
@job
def async_foo():
    bar.send(sender='baz')


def foo():
    async_foo.delay()


# the sync function for demonstration purpose
def sync_foo():
    bar.send(sender='baz')


class DjangoRQSignalTestCase(unittest.TestCase):

    def test_async_signal(self):
        with mock_signal_receiver(bar) as receiver:
            foo()
            get_worker().work(burst=True)
            self.assertEquals(receiver.call_count, 1)

    def test_sync_signal(self):
        with mock_signal_receiver(bar) as receiver:
            sync_foo()
            self.assertEquals(receiver.call_count, 1)

from django.utils import unittest
from django_rq import job 
from django.dispatch import Signal

# set up a Signal
bar = Signal()

globvar = 0

# the async function
@job
def async_foo():
    global globvar
    globvar = 1

def foo():
    async_foo.delay()

# the sync function for demonstration purpose
def sync_foo():
    global globvar
    globvar = 1

class DjangoRQSignalTestCase(unittest.TestCase):

    def test_async_signal(self):
        foo()
        get_worker().work(burst=True)
        self.assertEquals(globvar, 1)  # this should pass, but globvar is 0

    def test_sync_signal(self):
        sync_foo()
        self.assertEquals(globvar, 1)  # this pass

from rq.decorators import job as _rq_job

from .queues import get_queue


def job(func_or_queue, connection=None, *args, **kwargs):
    """
    The same as RQ's job decorator, but it works automatically works out
    the ``connection`` argument from RQ_QUEUES.

    And also, it allows simplified ``@job`` syntax to put job into
    default queue.
    """
    async = kwargs.pop('async', False)

    if callable(func_or_queue):
        func = func_or_queue
        queue = 'default'
    else:
        func = None
        queue = func_or_queue

    if isinstance(queue, basestring):
        try:
            queue = get_queue(queue, async=async)
            if connection is None:
                connection = queue.connection
        except KeyError:
            pass

    decorator = _rq_job(queue, connection=connection, *args, **kwargs)
    if func:
        return decorator(func)
    return decorator