From 0740c90148dc58308779aa4af54ff0e02cd3bc06 Mon Sep 17 00:00:00 2001 From: Ilan Steemers Date: Sat, 29 Aug 2015 18:54:07 +0200 Subject: [PATCH] adds pluggable brokers * added redis broker * added django_redis broker * added task acknowledgement --- django_q/brokers/__init__.py | 47 ++++++++++ django_q/brokers/django_redis.py | 10 +++ django_q/brokers/redis.py | 43 +++++++++ django_q/cluster.py | 45 +++++----- django_q/conf.py | 1 + django_q/monitor.py | 19 ++-- django_q/status.py | 40 ++++----- django_q/tasks.py | 16 ++-- django_q/tests/test_cluster.py | 145 ++++++++++++++++--------------- django_q/tests/test_config.py | 5 -- django_q/tests/test_monitor.py | 10 +-- django_q/tests/test_scheduler.py | 29 ++++--- 12 files changed, 247 insertions(+), 163 deletions(-) create mode 100644 django_q/brokers/__init__.py create mode 100644 django_q/brokers/django_redis.py create mode 100644 django_q/brokers/redis.py diff --git a/django_q/brokers/__init__.py b/django_q/brokers/__init__.py new file mode 100644 index 0000000..23ebf21 --- /dev/null +++ b/django_q/brokers/__init__.py @@ -0,0 +1,47 @@ +from django_q.conf import Conf + + +class Broker(object): + def __init__(self, list_key=Conf.Q_LIST): + self.connection = self.get_connection() + self.list_key = list_key + + def enqueue(self, task): + pass + + def dequeue(self): + pass + + def queue_size(self): + pass + + def delete_queue(self, list_key=None): + pass + + def acknowledge(self, ack_id): + pass + + def ping(self): + pass + + def set(self, key, value, timeout): + pass + + def get(self, key): + pass + + def get_pattern(self, pattern): + pass + + @staticmethod + def get_connection(): + return 0 + + +def get_broker(list_key=Conf.Q_LIST): + if Conf.REDIS: + from brokers import redis + return redis.Redis(list_key=list_key) + elif Conf.DJANGO_REDIS: + from brokers import django_redis + return django_redis.DjangoRedis(list_key=list_key) diff --git a/django_q/brokers/django_redis.py b/django_q/brokers/django_redis.py new file mode 100644 index 0000000..35b67bf --- /dev/null +++ b/django_q/brokers/django_redis.py @@ -0,0 +1,10 @@ +import django_redis +from django_q.brokers import redis +from django_q.conf import Conf + + +class DjangoRedis(redis.Redis): + + @staticmethod + def get_connection(): + return django_redis.get_redis_connection(Conf.DJANGO_REDIS) \ No newline at end of file diff --git a/django_q/brokers/redis.py b/django_q/brokers/redis.py new file mode 100644 index 0000000..86e0143 --- /dev/null +++ b/django_q/brokers/redis.py @@ -0,0 +1,43 @@ +import redis +from django_q.brokers import Broker +from django_q.conf import Conf, logger + + +class Redis(Broker): + def enqueue(self, task): + return self.connection.rpush(self.list_key, task) + + def dequeue(self): + task = self.connection.blpop(self.list_key, 1) + if task: + return None, task[1] + + def queue_size(self): + return self.connection.llen(self.list_key) + + def delete_queue(self, list_key=None): + list_key = list_key if list_key else self.list_key + return self.connection.delete(list_key) + + def ping(self): + try: + return self.connection.ping() + except Exception as e: + logger.error('Can not connect to Redis server.') + raise e + + def set(self, key, value, timeout): + self.connection.set(key, value, timeout) + + def get(self, key): + if self.connection.exists(key): + return self.connection.get(key) + + def get_pattern(self, pattern): + keys = self.connection.keys(pattern=pattern) + if keys: + return self.connection.mget(keys) + + @staticmethod + def get_connection(): + return redis.StrictRedis(**Conf.REDIS) diff --git a/django_q/cluster.py b/django_q/cluster.py index 1e03148..c28c4ad 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -30,19 +30,20 @@ from django import db import signing import tasks -from django_q.conf import Conf, redis_client, logger, psutil, get_ppid +from django_q.conf import Conf, logger, psutil, get_ppid from django_q.models import Task, Success, Schedule -from django_q.status import Stat, Status, ping_redis +from django_q.status import Stat, Status +from django_q.brokers import get_broker class Cluster(object): - def __init__(self, list_key=Conf.Q_LIST): + def __init__(self, broker=get_broker()): + self.broker = broker self.sentinel = None self.stop_event = None self.start_event = None self.pid = current_process().pid self.host = socket.gethostname() - self.list_key = list_key self.timeout = Conf.TIMEOUT signal.signal(signal.SIGTERM, self.sig_handler) signal.signal(signal.SIGINT, self.sig_handler) @@ -58,7 +59,7 @@ class Cluster(object): self.stop_event = Event() self.start_event = Event() self.sentinel = Process(target=Sentinel, - args=(self.stop_event, self.start_event, self.list_key, self.timeout)) + args=(self.stop_event, self.start_event, self.broker, self.timeout)) self.sentinel.start() logger.info(_('Q Cluster-{} starting.').format(self.pid)) while not self.start_event.is_set(): @@ -105,15 +106,14 @@ class Cluster(object): class Sentinel(object): - def __init__(self, stop_event, start_event, list_key=Conf.Q_LIST, timeout=Conf.TIMEOUT, start=True): + def __init__(self, stop_event, start_event, broker=get_broker(), timeout=Conf.TIMEOUT, start=True): # Make sure we catch signals for the pool signal.signal(signal.SIGINT, signal.SIG_IGN) signal.signal(signal.SIGTERM, signal.SIG_DFL) self.pid = current_process().pid self.parent_pid = get_ppid() self.name = current_process().name - self.list_key = list_key - self.r = redis_client + self.broker = broker self.reincarnations = 0 self.tob = timezone.now() self.stop_event = stop_event @@ -130,7 +130,7 @@ class Sentinel(object): self.start() def start(self): - ping_redis(self.r) + self.broker.ping() self.spawn_cluster() self.guard() @@ -165,7 +165,7 @@ class Sentinel(object): return p def spawn_pusher(self): - return self.spawn_process(pusher, self.task_queue, self.event_out, self.list_key) + return self.spawn_process(pusher, self.task_queue, self.event_out, self.broker) def spawn_worker(self): self.spawn_process(worker, self.task_queue, self.result_queue, Value('f', -1), self.timeout) @@ -216,7 +216,7 @@ class Sentinel(object): self.start_event.set() Stat(self).save() logger.info(_('Q Cluster-{} running.').format(self.parent_pid)) - scheduler(list_key=self.list_key) + scheduler(broker=self.broker) counter = 0 cycle = 0.5 # guard loop sleep in seconds # Guard loop. Runs at least once @@ -240,7 +240,7 @@ class Sentinel(object): counter += cycle if counter == 30: counter = 0 - scheduler(list_key=self.list_key) + scheduler(broker=self.broker) # Save current status Stat(self).save() sleep(cycle) @@ -287,38 +287,38 @@ class Sentinel(object): Stat(self).save() -def pusher(task_queue, event, list_key=Conf.Q_LIST): +def pusher(task_queue, event, broker=get_broker()): """ Pulls tasks of the Redis List and puts them in the task queue :type task_queue: multiprocessing.Queue :type event: multiprocessing.Event - :type list_key: str """ logger.info(_('{} pushing tasks at {}').format(current_process().name, current_process().pid)) - r = redis_client while True: try: - task = r.blpop(list_key, 1) + task = broker.dequeue() except Exception as e: logger.error(e) - # redis probably crashed. Let the sentinel handle it. + # broker probably crashed. Let the sentinel handle it. sleep(10) break if task: + ack_id = task[0] # unpack the task try: task = signing.SignedPackage.loads(task[1]) except (TypeError, signing.BadSignature) as e: logger.error(e) continue + task['ack_id'] = ack_id task_queue.put(task) - logger.debug(_('queueing from {}').format(list_key)) + logger.debug(_('queueing from {}').format(broker.list_key)) if event.is_set(): break logger.info(_("{} stopped pushing tasks").format(current_process().name)) -def monitor(result_queue): +def monitor(result_queue, broker=get_broker()): """ Gets finished tasks from the result queue and saves them to Django :type result_queue: multiprocessing.Queue @@ -327,6 +327,9 @@ def monitor(result_queue): logger.info(_("{} monitoring at {}").format(name, current_process().pid)) db.close_old_connections() for task in iter(result_queue.get, 'STOP'): + ack_id = task.pop('ack_id', False) + if ack_id: + broker.acknowledge(ack_id) save_task(task) if task['success']: logger.info(_("Processed [{}]").format(task['name'])) @@ -410,7 +413,7 @@ def save_task(task): logger.error(e) -def scheduler(list_key=Conf.Q_LIST): +def scheduler(broker=get_broker()): """ Creates a task from a schedule at the scheduled time and schedules next run """ @@ -456,7 +459,7 @@ def scheduler(list_key=Conf.Q_LIST): s.next_run = next_run.datetime s.repeats += -1 # send it to the cluster - q_options['list_key'] = list_key + q_options['broker'] = broker q_options['group'] = s.name or s.id kwargs['q_options'] = q_options s.task = tasks.async(s.func, *args, **kwargs) diff --git a/django_q/conf.py b/django_q/conf.py index 16a00b1..c31834c 100644 --- a/django_q/conf.py +++ b/django_q/conf.py @@ -46,6 +46,7 @@ class Conf(object): # Maximum number of tasks that each cluster can work on QUEUE_LIMIT = conf.get('queue_limit', None) + # Number of workers in the pool. Default is cpu count if implemented, otherwise 4. WORKERS = conf.get('workers', False) if not WORKERS: diff --git a/django_q/monitor.py b/django_q/monitor.py index 9acc8b0..c978607 100644 --- a/django_q/monitor.py +++ b/django_q/monitor.py @@ -11,13 +11,14 @@ from django.utils.translation import ugettext as _ # local from django_q.conf import Conf, redis_client -from django_q.status import Stat, ping_redis +from django_q.status import Stat +from django_q.brokers import get_broker from django_q import models -def monitor(run_once=False, r=redis_client): +def monitor(run_once=False, broker=get_broker()): term = Terminal() - ping_redis(r) + broker.ping() with term.fullscreen(), term.hidden_cursor(), term.cbreak(): val = None start_width = int(term.width / 8) @@ -36,7 +37,7 @@ def monitor(run_once=False, r=redis_client): print(term.move(0, 6 * col_width) + term.black_on_green(term.center(_('RC'), width=col_width - 1))) print(term.move(0, 7 * col_width) + term.black_on_green(term.center(_('Up'), width=col_width - 1))) i = 2 - stats = Stat.get_all(r=r) + stats = Stat.get_all(broker=broker) print(term.clear_eos()) for stat in stats: status = stat.status @@ -79,15 +80,15 @@ def monitor(run_once=False, r=redis_client): i += 1 # for testing if run_once: - return Stat.get_all(r=r) + return Stat.get_all(broker=broker) print(term.move(i + 2, 0) + term.center(_('[Press q to quit]'))) val = term.inkey(timeout=1) -def info(r=redis_client): +def info(broker=get_broker()): term = Terminal() - ping_redis(r) - stat = Stat.get_all(r) + broker.ping() + stat = Stat.get_all(broker=broker) # general stats clusters = len(stat) workers = 0 @@ -141,7 +142,7 @@ def info(r=redis_client): ) print(term.cyan(_('Queued')) + term.move_x(1 * col_width) + - term.white(str(r.llen(Conf.Q_LIST))) + + term.white(str(broker.queue_size())) + term.move_x(2 * col_width) + term.cyan(_('Successes')) + term.move_x(3 * col_width) + diff --git a/django_q/status.py b/django_q/status.py index ef98148..63139c2 100644 --- a/django_q/status.py +++ b/django_q/status.py @@ -1,6 +1,7 @@ import socket from django.utils import timezone -from django_q.conf import Conf, logger, redis_client +from django_q.brokers import get_broker +from django_q.conf import Conf, logger import signing @@ -27,7 +28,7 @@ class Stat(Status): def __init__(self, sentinel): super(Stat, self).__init__(sentinel.parent_pid or sentinel.pid) - self.r = sentinel.r + self.broker = sentinel.broker or get_broker() self.tob = sentinel.tob self.reincarnations = sentinel.reincarnations self.sentinel = sentinel.pid @@ -63,7 +64,7 @@ class Stat(Status): def save(self): try: - self.r.set(self.key, signing.SignedPackage.dumps(self, True), 3) + self.broker.set(self.key, signing.SignedPackage.dumps(self, True), 3) except Exception as e: logger.error(e) @@ -71,15 +72,14 @@ class Stat(Status): return self.done_q_size + self.task_q_size == 0 @staticmethod - def get(cluster_id, r=redis_client): + def get(cluster_id, broker=get_broker()): """ gets the current status for the cluster :param cluster_id: id of the cluster :return: Stat or Status """ - key = Stat.get_key(cluster_id) - if r.exists(key): - pack = r.get(key) + pack = broker.get(Stat.get_key(cluster_id)) + if pack: try: return signing.SignedPackage.loads(pack) except signing.BadSignature: @@ -87,33 +87,23 @@ class Stat(Status): return Status(cluster_id) @staticmethod - def get_all(r=redis_client): + def get_all(broker=get_broker()): """ Get the status for all currently running clusters with the same prefix and secret key. :return: list of type Stat """ stats = [] - keys = r.keys(pattern='{}:*'.format(Conf.Q_STAT)) - if keys: - packs = r.mget(keys) - for pack in packs: - try: - stats.append(signing.SignedPackage.loads(pack)) - except signing.BadSignature: - continue + packs = broker.get_pattern('{}:*'.format(Conf.Q_STAT)) or [] + for pack in packs: + try: + stats.append(signing.SignedPackage.loads(pack)) + except signing.BadSignature: + continue return stats def __getstate__(self): # Don't pickle the redis connection state = dict(self.__dict__) - del state['r'] + del state['broker'] return state - - -def ping_redis(r): - try: - r.ping() - except Exception as e: - logger.error('Can not connect to Redis server.') - raise e diff --git a/django_q/tasks.py b/django_q/tasks.py index b9c7360..599078f 100644 --- a/django_q/tasks.py +++ b/django_q/tasks.py @@ -7,9 +7,10 @@ from django.utils import timezone # local import signing import cluster -from django_q.conf import Conf, redis_client, logger +from django_q.conf import Conf, logger from django_q.models import Schedule, Task from django_q.humanhash import uuid +from django_q.brokers import get_broker def async(func, *args, **kwargs): @@ -17,8 +18,7 @@ def async(func, *args, **kwargs): # get options from q_options dict or direct from kwargs options = kwargs.pop('q_options', kwargs) hook = options.pop('hook', None) - list_key = options.pop('list_key', Conf.Q_LIST) - redis = options.pop('redis', redis_client) + broker = options.pop('broker', get_broker()) sync = options.pop('sync', False) group = options.pop('group', None) save = options.pop('save', None) @@ -42,7 +42,7 @@ def async(func, *args, **kwargs): if sync or Conf.SYNC: return _sync(pack) # push it - redis.rpush(list_key, pack) + broker.enqueue(pack) logger.debug('Pushed {}'.format(tag)) return task['id'] @@ -152,17 +152,17 @@ def delete_group(group_id, tasks=False): return Task.delete_group(group_id, tasks) -def queue_size(list_key=Conf.Q_LIST, r=redis_client): +def queue_size(broker=get_broker()): """ Returns the current queue size. Note that this doesn't count any tasks currently being processed by workers. - :param list_key: optional redis key - :param r: optional redis connection + :param list_key: optional list key + :param broker: optional broker :return: current queue size :rtype: int """ - return r.llen(list_key) + return broker.queue_size() def _sync(pack): diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index 7e5a407..f35c3c3 100644 --- a/django_q/tests/test_cluster.py +++ b/django_q/tests/test_cluster.py @@ -13,8 +13,9 @@ from django_q.cluster import Cluster, Sentinel, pusher, worker, monitor from django_q.humanhash import DEFAULT_WORDLIST from django_q.tasks import fetch, fetch_group, async, result, result_group, count_group, delete_group, queue_size from django_q.models import Task, Success -from django_q.conf import Conf, redis_client +from django_q.conf import Conf from django_q.status import Stat +from django_q.brokers import get_broker from .tasks import multiply @@ -27,25 +28,25 @@ class WordClass(object): @pytest.fixture -def r(): - return redis_client +def broker(): + return get_broker() -def test_redis_connection(r): - assert r.ping() is True +def test_redis_connection(broker): + assert broker.ping() is True @pytest.mark.django_db -def test_sync(r): - task = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, redis=r, sync=True) +def test_sync(broker): + task = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, broker=broker, sync=True) assert result(task) == 1506 @pytest.mark.django_db -def test_cluster_initial(r): - list_key = 'initial_test:q' - r.delete(list_key) - c = Cluster(list_key=list_key) +def test_cluster_initial(broker): + broker.list_key = 'initial_test:q' + broker.delete_queue() + c = Cluster(broker=broker) assert c.sentinel is None assert c.stat.status == Conf.STOPPED assert c.start() > 0 @@ -59,7 +60,7 @@ def test_cluster_initial(r): assert c.stop() is True assert c.sentinel.is_alive() is False assert c.has_stopped - r.delete(list_key) + broker.delete_queue() @pytest.mark.django_db @@ -67,17 +68,17 @@ def test_sentinel(): start_event = Event() stop_event = Event() stop_event.set() - s = Sentinel(stop_event, start_event, list_key='sentinel_test:q') + s = Sentinel(stop_event, start_event, broker=get_broker('sentinel_test:q')) assert start_event.is_set() assert s.status() == Conf.STOPPED @pytest.mark.django_db -def test_cluster(r): - list_key = 'cluster_test:q' - r.delete(list_key) - task = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, list_key=list_key) - assert queue_size(list_key=list_key, r=r) == 1 +def test_cluster(broker): + broker.list_key = 'cluster_test:q' + broker.delete_queue() + task = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, broker=broker) + assert broker.queue_size() == 1 task_queue = Queue() assert task_queue.qsize() == 0 result_queue = Queue() @@ -85,9 +86,9 @@ def test_cluster(r): event = Event() event.set() # Test push - pusher(task_queue, event, list_key=list_key) + pusher(task_queue, event, broker=broker) assert task_queue.qsize() == 1 - assert queue_size(list_key=list_key, r=r) == 0 + assert queue_size(broker=broker) == 0 # Test work task_queue.put('STOP') worker(task_queue, result_queue, Value('f', -1)) @@ -99,36 +100,36 @@ def test_cluster(r): assert result_queue.qsize() == 0 # check result assert result(task) == 1506 - r.delete(list_key) + broker.delete_queue() @pytest.mark.django_db -def test_async(r, admin_user): - list_key = 'cluster_test:q' - r.delete(list_key) +def test_async(broker, admin_user): + broker.list_key = 'cluster_test:q' + broker.delete_queue() a = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, hook='django_q.tests.test_cluster.assert_result', - list_key=list_key, redis=r) + broker=broker) b = async('django_q.tests.tasks.count_letters2', WordClass(), hook='django_q.tests.test_cluster.assert_result', - list_key=list_key, redis=r) + broker=broker) # unknown argument c = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, 'oneargumentoomany', - hook='django_q.tests.test_cluster.assert_bad_result', list_key=list_key, redis=r) + hook='django_q.tests.test_cluster.assert_bad_result', broker=broker) # unknown function d = async('django_q.tests.tasks.does_not_exist', WordClass(), hook='django_q.tests.test_cluster.assert_bad_result', - list_key=list_key, redis=r) + broker=broker) # function without result - e = async('django_q.tests.tasks.countdown', 100000, list_key=list_key, redis=r) + e = async('django_q.tests.tasks.countdown', 100000, broker=broker) # function as instance - f = async(multiply, 753, 2, hook=assert_result, list_key=list_key, redis=r) + f = async(multiply, 753, 2, hook=assert_result, broker=broker) # model as argument - g = async('django_q.tests.tasks.get_task_name', Task(name='John'), list_key=list_key, redis=r) + g = async('django_q.tests.tasks.get_task_name', Task(name='John'), broker=broker) # args,kwargs, group and broken hook - h = async('django_q.tests.tasks.word_multiply', 2, word='django', hook='fail.me', list_key=list_key, redis=r) + h = async('django_q.tests.tasks.word_multiply', 2, word='django', hook='fail.me', broker=broker) # args unpickle test - j = async('django_q.tests.tasks.get_user_id', admin_user, list_key=list_key, group='test_j', redis=r) + j = async('django_q.tests.tasks.get_user_id', admin_user, broker=broker, group='test_j') # q_options and save opt_out test k = async('django_q.tests.tasks.get_user_id', admin_user, - q_options={'list_key': list_key, 'group': 'test_k', 'redis': r, 'save': False, 'timeout': 90}) + q_options={'broker': broker, 'group': 'test_k', 'save': False, 'timeout': 90}) # check if everything has a task id assert isinstance(a, str) assert isinstance(b, str) @@ -142,14 +143,14 @@ def test_async(r, admin_user): assert isinstance(k, str) # run the cluster to execute the tasks task_count = 10 - assert queue_size(list_key=list_key, r=r) == task_count + assert broker.queue_size() == task_count task_queue = Queue() stop_event = Event() stop_event.set() # push the tasks for i in range(task_count): - pusher(task_queue, stop_event, list_key=list_key) - assert queue_size(list_key=list_key, r=r) == 0 + pusher(task_queue, stop_event, broker=broker) + assert broker.queue_size() == 0 assert task_queue.qsize() == task_count task_queue.put('STOP') # let a worker handle them @@ -218,64 +219,64 @@ def test_async(r, admin_user): assert delete_group('test_j', tasks=True) is None # task k should not have been saved assert fetch(k) is None - r.delete(list_key) + broker.delete_queue() @pytest.mark.django_db -def test_timeout(r): +def test_timeout(broker): # set up the Sentinel - list_key = 'timeout_test:q' - async('django_q.tests.tasks.count_forever', list_key=list_key) + broker.list_key = 'timeout_test:q' + async('django_q.tests.tasks.count_forever',broker=broker) start_event = Event() stop_event = Event() # Set a timer to stop the Sentinel threading.Timer(3, stop_event.set).start() - s = Sentinel(stop_event, start_event, list_key=list_key, timeout=1) + s = Sentinel(stop_event, start_event, broker=broker, timeout=1) assert start_event.is_set() assert s.status() == Conf.STOPPED assert s.reincarnations == 1 - r.delete(list_key) + broker.delete_queue() @pytest.mark.django_db -def test_timeout(r): +def test_timeout(broker): # set up the Sentinel - list_key = 'timeout_test:q' - async('django_q.tests.tasks.count_forever', list_key=list_key) + broker.list_key = 'timeout_test:q' + async('django_q.tests.tasks.count_forever', broker=broker) start_event = Event() stop_event = Event() # Set a timer to stop the Sentinel threading.Timer(3, stop_event.set).start() - s = Sentinel(stop_event, start_event, list_key=list_key, timeout=1) + s = Sentinel(stop_event, start_event,broker=broker, timeout=1) assert start_event.is_set() assert s.status() == Conf.STOPPED assert s.reincarnations == 1 - r.delete(list_key) + broker.delete_queue() @pytest.mark.django_db -def test_timeout_override(r): +def test_timeout_override(broker): # set up the Sentinel - list_key = 'timeout_override_test:q' - async('django_q.tests.tasks.count_forever', list_key=list_key, timeout=1) + broker.list_key = 'timeout_override_test:q' + async('django_q.tests.tasks.count_forever', broker=broker, timeout=1) start_event = Event() stop_event = Event() # Set a timer to stop the Sentinel threading.Timer(3, stop_event.set).start() - s = Sentinel(stop_event, start_event, list_key=list_key, timeout=10) + s = Sentinel(stop_event, start_event, broker=broker, timeout=10) assert start_event.is_set() assert s.status() == Conf.STOPPED assert s.reincarnations == 1 - r.delete(list_key) + broker.delete_queue() @pytest.mark.django_db -def test_recycle(r): +def test_recycle(broker): # set up the Sentinel - list_key = 'test_recycle_test:q' - async('django_q.tests.tasks.multiply', 2, 2, list_key=list_key, redis=r) - async('django_q.tests.tasks.multiply', 2, 2, list_key=list_key, redis=r) - async('django_q.tests.tasks.multiply', 2, 2, list_key=list_key, redis=r) + broker.list_key = 'test_recycle_test:q' + async('django_q.tests.tasks.multiply', 2, 2, broker=broker) + async('django_q.tests.tasks.multiply', 2, 2, broker=broker) + async('django_q.tests.tasks.multiply', 2, 2, broker=broker) start_event = Event() stop_event = Event() # override settings @@ -283,17 +284,17 @@ def test_recycle(r): Conf.WORKERS = 1 # set a timer to stop the Sentinel threading.Timer(3, stop_event.set).start() - s = Sentinel(stop_event, start_event, list_key=list_key) + s = Sentinel(stop_event, start_event,broker=broker) assert start_event.is_set() assert s.status() == Conf.STOPPED assert s.reincarnations == 1 - async('django_q.tests.tasks.multiply', 2, 2, list_key=list_key, redis=r) - async('django_q.tests.tasks.multiply', 2, 2, list_key=list_key, redis=r) + async('django_q.tests.tasks.multiply', 2, 2, broker=broker) + async('django_q.tests.tasks.multiply', 2, 2, broker=broker) task_queue = Queue() result_queue = Queue() # push two tasks - pusher(task_queue, stop_event, list_key=list_key) - pusher(task_queue, stop_event, list_key=list_key) + pusher(task_queue, stop_event, broker=broker) + pusher(task_queue, stop_event, broker=broker) # worker should exit on recycle worker(task_queue, result_queue, Value('f', -1)) # check if the work has been done @@ -304,30 +305,30 @@ def test_recycle(r): # run monitor monitor(result_queue) assert Success.objects.count() == Conf.SAVE_LIMIT - r.delete(list_key) + broker.delete_queue() @pytest.mark.django_db -def test_bad_secret(r, monkeypatch): - list_key = 'test_bad_secret' - async('math.copysign', 1, -1, list_key=list_key) +def test_bad_secret(broker, monkeypatch): + broker.list_key='test_bad_secret:q' + async('math.copysign', 1, -1, broker=broker) stop_event = Event() stop_event.set() start_event = Event() - s = Sentinel(stop_event, start_event, list_key=list_key, start=False) + s = Sentinel(stop_event, start_event, broker=broker, start=False) Stat(s).save() # change the SECRET monkeypatch.setattr(Conf, "SECRET_KEY", "OOPS") - stat = Stat.get_all(r) + stat = Stat.get_all() assert len(stat) == 0 - assert Stat.get(s.parent_pid, r) is None + assert Stat.get(s.parent_pid) is None task_queue = Queue() - pusher(task_queue, stop_event, list_key=list_key) + pusher(task_queue, stop_event, broker=broker) result_queue = Queue() task_queue.put('STOP') worker(task_queue, result_queue, Value('f', -1), ) assert result_queue.qsize() == 0 - r.delete(list_key) + broker.delete_queue() @pytest.mark.django_db diff --git a/django_q/tests/test_config.py b/django_q/tests/test_config.py index de5ab71..10ade3d 100644 --- a/django_q/tests/test_config.py +++ b/django_q/tests/test_config.py @@ -3,11 +3,6 @@ import pytest from django_q import conf -@pytest.fixture -def r(): - return conf.redis_client - - def test_django_redis(): conf.Conf.DJANGO_REDIS = None assert conf.redis_client.ping() is True diff --git a/django_q/tests/test_monitor.py b/django_q/tests/test_monitor.py index 4f94b75..e1a3ac0 100644 --- a/django_q/tests/test_monitor.py +++ b/django_q/tests/test_monitor.py @@ -1,10 +1,9 @@ import pytest -import redis from django_q import async from django_q.cluster import Cluster from django_q.monitor import monitor, info -from django_q.status import Stat, ping_redis +from django_q.status import Stat @pytest.mark.django_db @@ -37,10 +36,3 @@ def test_info(): def do_sync(): async('django_q.tests.tasks.countdown', 1, sync=True, save=True) - - -@pytest.mark.django_db -def test_ping_redis(): - r = redis.StrictRedis(port=6388) - with pytest.raises(Exception): - ping_redis(r) diff --git a/django_q/tests/test_scheduler.py b/django_q/tests/test_scheduler.py index 7709477..2ee337a 100644 --- a/django_q/tests/test_scheduler.py +++ b/django_q/tests/test_scheduler.py @@ -6,20 +6,21 @@ import arrow from django.utils import timezone -from django_q.conf import redis_client, Conf +from django_q.brokers import get_broker +from django_q.conf import Conf from django_q.cluster import pusher, worker, monitor, scheduler from django_q.tasks import Schedule, fetch, schedule as create_schedule, queue_size @pytest.fixture -def r(): - return redis_client +def broker(): + return get_broker() @pytest.mark.django_db -def test_scheduler(r): - list_key = 'scheduler_test:q' - r.delete(list_key) +def test_scheduler(broker): + broker.list_key = 'scheduler_test:q' + broker.delete_queue() schedule = create_schedule('math.copysign', 1, -1, name='test math', @@ -28,15 +29,15 @@ def test_scheduler(r): repeats=1) assert schedule.last_run() is None # run scheduler - scheduler(list_key=list_key) + scheduler(broker=broker) # set up the workflow task_queue = Queue() stop_event = Event() stop_event.set() # push it - pusher(task_queue, stop_event, list_key=list_key) + pusher(task_queue, stop_event, broker=broker) assert task_queue.qsize() == 1 - assert queue_size(list_key=list_key, r=r) == 0 + assert broker.queue_size() == 0 task_queue.put('STOP') # let a worker handle them result_queue = Queue() @@ -91,7 +92,7 @@ def test_scheduler(r): ) assert schedule is not None assert schedule.last_run() is None - scheduler(list_key=list_key) + scheduler(broker=broker) # via model Schedule.objects.create(func='django_q.tests.tasks.word_multiply', args='2', @@ -99,7 +100,7 @@ def test_scheduler(r): schedule_type=Schedule.DAILY ) # scheduler - scheduler(list_key=list_key) + scheduler(broker=broker) # ONCE schedule should be deleted assert Schedule.objects.filter(pk=once_schedule.pk).exists() is False # Catch up On @@ -112,13 +113,13 @@ def test_scheduler(r): next_run=timezone.now() - timedelta(hours=12), repeats=-1 ) - scheduler(list_key=list_key) + scheduler(broker=broker) schedule = Schedule.objects.get(pk=schedule.pk) assert schedule.next_run < now # Catch up off Conf.CATCH_UP = False - scheduler(list_key=list_key) + scheduler(broker=broker) schedule = Schedule.objects.get(pk=schedule.pk) assert schedule.next_run > now # Done - r.delete(list_key) + broker.delete_queue()