Replace global Conf mangling with monkeypatch

The test suite, especially for the brokers, was heavily dependant on the
success of the tests preceeding it to pass. This commit should eliminate
that dependency and allow tests to fail without affecting others.
It also removes the need to cleanup globals manually after a test.
This commit is contained in:
Harm Geerts
2016-11-29 18:08:52 +01:00
parent 82f3cbdf28
commit cf37efb1eb
6 changed files with 55 additions and 99 deletions
+10 -18
View File
@@ -29,13 +29,8 @@ class WordClass(object):
@pytest.fixture
def broker():
Conf.DISQUE_NODES = None
Conf.IRON_MQ = None
Conf.SQS = None
Conf.ORM = None
Conf.MONGO = None
Conf.DJANGO_REDIS = 'default'
def broker(monkeypatch):
monkeypatch.setattr(Conf, 'DJANGO_REDIS', 'default')
return get_broker()
@@ -281,7 +276,7 @@ def test_timeout_override(broker):
@pytest.mark.django_db
def test_recycle(broker):
def test_recycle(broker, monkeypatch):
# set up the Sentinel
broker.list_key = 'test_recycle_test:q'
async('django_q.tests.tasks.multiply', 2, 2, broker=broker)
@@ -290,8 +285,8 @@ def test_recycle(broker):
start_event = Event()
stop_event = Event()
# override settings
Conf.RECYCLE = 2
Conf.WORKERS = 1
monkeypatch.setattr(Conf, 'RECYCLE', 2)
monkeypatch.setattr(Conf, 'WORKERS', 1)
# set a timer to stop the Sentinel
threading.Timer(3, stop_event.set).start()
s = Sentinel(stop_event, start_event, broker=broker)
@@ -310,7 +305,7 @@ def test_recycle(broker):
# check if the work has been done
assert result_queue.qsize() == 2
# save_limit test
Conf.SAVE_LIMIT = 1
monkeypatch.setattr(Conf, 'SAVE_LIMIT', 1)
result_queue.put('STOP')
# run monitor
monitor(result_queue)
@@ -361,14 +356,14 @@ def test_update_failed(broker):
sleep(0.5)
# second save - no success
old_stopped = task['stopped']
task['stopped']=timezone.now()
task['stopped'] = timezone.now()
save_task(task, broker)
saved_task = Task.objects.get(id=task['id'])
assert saved_task.stopped > old_stopped
# third save - success
task['stopped']=timezone.now()
task['result']='result'
task['success']=True
task['stopped'] = timezone.now()
task['result'] = 'result'
task['success'] = True
save_task(task, broker)
saved_task = Task.objects.get(id=task['id'])
assert saved_task.success is True
@@ -383,9 +378,6 @@ def test_update_failed(broker):
assert saved_task.result == 'result'
@pytest.mark.django_db
def assert_result(task):
assert task is not None