fork/django-q2
1
0
Fork 0
mirror of https://github.com/django-q2/django-q2.git synced 2026-10-09 14:10:55 +08:00
Code Issues Packages Projects Releases Wiki Activity

Test for no nodes config

Browse Source
This commit is contained in:
Ilan Steemers
2020-06-30 21:35:57 +02:00
parent d46d08e94e
commit b17fd24682
1 changed files with 102 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
django_q/tests/test_brokers.py
+102 -84
View File
@@ -11,72 +11,72 @@ from django_q.humanhash import uuid
def test_broker(monkeypatch): def test_broker(monkeypatch):
broker = Broker() broker = Broker()
broker.enqueue('test') broker.enqueue("test")
broker.dequeue() broker.dequeue()
broker.queue_size() broker.queue_size()
broker.lock_size() broker.lock_size()
broker.purge_queue() broker.purge_queue()
broker.delete('id') broker.delete("id")
broker.delete_queue() broker.delete_queue()
broker.acknowledge('test') broker.acknowledge("test")
broker.ping() broker.ping()
broker.info() broker.info()
# stats # stats
assert broker.get_stat('test_1') is None assert broker.get_stat("test_1") is None
broker.set_stat('test_1', 'test', 3) broker.set_stat("test_1", "test", 3)
assert broker.get_stat('test_1') == 'test' assert broker.get_stat("test_1") == "test"
assert broker.get_stats('test:*')[0] == 'test' assert broker.get_stats("test:*")[0] == "test"
# stats with no cache # stats with no cache
monkeypatch.setattr(Conf, 'CACHE', 'not_configured') monkeypatch.setattr(Conf, "CACHE", "not_configured")
broker.cache = broker.get_cache() broker.cache = broker.get_cache()
assert broker.get_stat('test_1') is None assert broker.get_stat("test_1") is None
broker.set_stat('test_1', 'test', 3) broker.set_stat("test_1", "test", 3)
assert broker.get_stat('test_1') is None assert broker.get_stat("test_1") is None
assert broker.get_stats('test:*') is None assert broker.get_stats("test:*") is None
def test_redis(monkeypatch): def test_redis(monkeypatch):
monkeypatch.setattr(Conf, 'DJANGO_REDIS', None) monkeypatch.setattr(Conf, "DJANGO_REDIS", None)
broker = get_broker() broker = get_broker()
assert broker.ping() is True assert broker.ping() is True
assert broker.info() is not None assert broker.info() is not None
monkeypatch.setattr(Conf, 'REDIS', {'host': '127.0.0.1', 'port': 7799}) monkeypatch.setattr(Conf, "REDIS", {"host": "127.0.0.1", "port": 7799})
broker = get_broker() broker = get_broker()
with pytest.raises(Exception): with pytest.raises(Exception):
broker.ping() broker.ping()
monkeypatch.setattr(Conf, 'REDIS', 'redis://127.0.0.1:7799') monkeypatch.setattr(Conf, "REDIS", "redis://127.0.0.1:7799")
broker = get_broker() broker = get_broker()
with pytest.raises(Exception): with pytest.raises(Exception):
broker.ping() broker.ping()
def test_custom(monkeypatch): def test_custom(monkeypatch):
monkeypatch.setattr(Conf, 'BROKER_CLASS', 'brokers.redis_broker.Redis') monkeypatch.setattr(Conf, "BROKER_CLASS", "brokers.redis_broker.Redis")
broker = get_broker() broker = get_broker()
assert broker.ping() is True assert broker.ping() is True
assert broker.info() is not None assert broker.info() is not None
assert broker.__class__.__name__ == 'Redis' assert broker.__class__.__name__ == "Redis"
def test_disque(monkeypatch): def test_disque(monkeypatch):
monkeypatch.setattr(Conf, 'DISQUE_NODES', ['127.0.0.1:7711']) monkeypatch.setattr(Conf, "DISQUE_NODES", ["127.0.0.1:7711"])
# check broker # check broker
broker = get_broker(list_key='disque_test') broker = get_broker(list_key="disque_test")
assert broker.ping() is True assert broker.ping() is True
assert broker.info() is not None assert broker.info() is not None
# clear before we start # clear before we start
broker.delete_queue() broker.delete_queue()
# async_task # async_task
broker.enqueue('test') broker.enqueue("test")
assert broker.queue_size() == 1 assert broker.queue_size() == 1
# dequeue # dequeue
task = broker.dequeue()[0] task = broker.dequeue()[0]
assert task[1] == 'test' assert task[1] == "test"
broker.acknowledge(task[0]) broker.acknowledge(task[0])
assert broker.queue_size() == 0 assert broker.queue_size() == 0
# Retry test # Retry test
monkeypatch.setattr(Conf, 'RETRY', 1) monkeypatch.setattr(Conf, "RETRY", 1)
broker.enqueue('test') broker.enqueue("test")
assert broker.queue_size() == 1 assert broker.queue_size() == 1
broker.dequeue() broker.dequeue()
assert broker.queue_size() == 0 assert broker.queue_size() == 0
@@ -88,17 +88,17 @@ def test_disque(monkeypatch):
sleep(1.5) sleep(1.5)
assert broker.queue_size() == 0 assert broker.queue_size() == 0
# delete job # delete job
task_id = broker.enqueue('test') task_id = broker.enqueue("test")
broker.delete(task_id) broker.delete(task_id)
assert broker.dequeue() is None assert broker.dequeue() is None
# fail # fail
task_id = broker.enqueue('test') task_id = broker.enqueue("test")
broker.fail(task_id) broker.fail(task_id)
# bulk test # bulk test
for i in range(5): for i in range(5):
broker.enqueue('test') broker.enqueue("test")
monkeypatch.setattr(Conf, 'BULK', 5) monkeypatch.setattr(Conf, "BULK", 5)
monkeypatch.setattr(Conf, 'DISQUE_FASTACK', True) monkeypatch.setattr(Conf, "DISQUE_FASTACK", True)
tasks = broker.dequeue() tasks = broker.dequeue()
for task in tasks: for task in tasks:
assert task is not None assert task is not None
@@ -106,35 +106,46 @@ def test_disque(monkeypatch):
# test duplicate acknowledge # test duplicate acknowledge
broker.acknowledge(task[0]) broker.acknowledge(task[0])
# delete queue # delete queue
broker.enqueue('test') broker.enqueue("test")
broker.enqueue('test') broker.enqueue("test")
broker.delete_queue() broker.delete_queue()
assert broker.queue_size() == 0 assert broker.queue_size() == 0
# connection test # connection test
monkeypatch.setattr(Conf, 'DISQUE_NODES', ['127.0.0.1:7798', '127.0.0.1:7799']) monkeypatch.setattr(Conf, "DISQUE_NODES", ["127.0.0.1:7798", "127.0.0.1:7799"])
with pytest.raises(redis.exceptions.ConnectionError):
broker.get_connection()
# connection test with no nodes
monkeypatch.setattr(Conf, "DISQUE_NODES", None)
with pytest.raises(redis.exceptions.ConnectionError): with pytest.raises(redis.exceptions.ConnectionError):
broker.get_connection() broker.get_connection()
@pytest.mark.skipif(not os.getenv('IRON_MQ_TOKEN'), @pytest.mark.skipif(
reason="requires IronMQ credentials") not os.getenv("IRON_MQ_TOKEN"), reason="requires IronMQ credentials"
)
def test_ironmq(monkeypatch): def test_ironmq(monkeypatch):
monkeypatch.setattr(Conf, 'IRON_MQ', {'token': os.getenv('IRON_MQ_TOKEN'), monkeypatch.setattr(
'project_id': os.getenv('IRON_MQ_PROJECT_ID')}) Conf,
"IRON_MQ",
{
"token": os.getenv("IRON_MQ_TOKEN"),
"project_id": os.getenv("IRON_MQ_PROJECT_ID"),
},
)
# check broker # check broker
broker = get_broker(list_key=uuid()[0]) broker = get_broker(list_key=uuid()[0])
assert broker.ping() is True assert broker.ping() is True
assert broker.info() is not None assert broker.info() is not None
# initialize the queue # initialize the queue
broker.enqueue('test') broker.enqueue("test")
# clear before we start # clear before we start
broker.purge_queue() broker.purge_queue()
assert broker.queue_size() == 0 assert broker.queue_size() == 0
# async_task # async_task
broker.enqueue('test') broker.enqueue("test")
# dequeue # dequeue
task = broker.dequeue()[0] task = broker.dequeue()[0]
assert task[1] == 'test' assert task[1] == "test"
broker.acknowledge(task[0]) broker.acknowledge(task[0])
assert broker.dequeue() is None assert broker.dequeue() is None
# Retry test # Retry test
@@ -148,16 +159,16 @@ def test_ironmq(monkeypatch):
# broker.acknowledge(task[0]) # broker.acknowledge(task[0])
# sleep(3) # sleep(3)
# delete job # delete job
task_id = broker.enqueue('test') task_id = broker.enqueue("test")
broker.delete(task_id) broker.delete(task_id)
assert broker.dequeue() is None assert broker.dequeue() is None
# fail # fail
task_id = broker.enqueue('test') task_id = broker.enqueue("test")
broker.fail(task_id) broker.fail(task_id)
# bulk test # bulk test
for i in range(5): for i in range(5):
broker.enqueue('test') broker.enqueue("test")
monkeypatch.setattr(Conf, 'BULK', 5) monkeypatch.setattr(Conf, "BULK", 5)
tasks = broker.dequeue() tasks = broker.dequeue()
for task in tasks: for task in tasks:
assert task is not None assert task is not None
@@ -165,62 +176,69 @@ def test_ironmq(monkeypatch):
# duplicate acknowledge # duplicate acknowledge
broker.acknowledge(task[0]) broker.acknowledge(task[0])
# delete queue # delete queue
broker.enqueue('test') broker.enqueue("test")
broker.enqueue('test') broker.enqueue("test")
broker.purge_queue() broker.purge_queue()
assert broker.dequeue() is None assert broker.dequeue() is None
broker.delete_queue() broker.delete_queue()
@pytest.mark.skipif(not os.getenv('AWS_ACCESS_KEY_ID'), @pytest.mark.skipif(
reason="requires AWS credentials") not os.getenv("AWS_ACCESS_KEY_ID"), reason="requires AWS credentials"
)
def canceled_sqs(monkeypatch): def canceled_sqs(monkeypatch):
monkeypatch.setattr(Conf, 'SQS', {'aws_region': os.getenv('AWS_REGION'), monkeypatch.setattr(
'aws_access_key_id': os.getenv('AWS_ACCESS_KEY_ID'), Conf,
'aws_secret_access_key': os.getenv('AWS_SECRET_ACCESS_KEY')}) "SQS",
{
"aws_region": os.getenv("AWS_REGION"),
"aws_access_key_id": os.getenv("AWS_ACCESS_KEY_ID"),
"aws_secret_access_key": os.getenv("AWS_SECRET_ACCESS_KEY"),
},
)
# check broker # check broker
broker = get_broker(list_key=uuid()[0]) broker = get_broker(list_key=uuid()[0])
assert broker.ping() is True assert broker.ping() is True
assert broker.info() is not None assert broker.info() is not None
assert broker.queue_size() == 0 assert broker.queue_size() == 0
# async_task # async_task
broker.enqueue('test') broker.enqueue("test")
# dequeue # dequeue
task = broker.dequeue()[0] task = broker.dequeue()[0]
assert task[1] == 'test' assert task[1] == "test"
broker.acknowledge(task[0]) broker.acknowledge(task[0])
assert broker.dequeue() is None assert broker.dequeue() is None
# Retry test # Retry test
monkeypatch.setattr(Conf, 'RETRY', 1) monkeypatch.setattr(Conf, "RETRY", 1)
broker.enqueue('test') broker.enqueue("test")
sleep(2) sleep(2)
# Sometimes SQS is not linear # Sometimes SQS is not linear
task = broker.dequeue() task = broker.dequeue()
if not task: if not task:
pytest.skip('SQS being weird') pytest.skip("SQS being weird")
task = task[0] task = task[0]
assert len(task) > 0 assert len(task) > 0
broker.acknowledge(task[0]) broker.acknowledge(task[0])
sleep(2) sleep(2)
# delete job # delete job
monkeypatch.setattr(Conf, 'RETRY', 60) monkeypatch.setattr(Conf, "RETRY", 60)
broker.enqueue('test') broker.enqueue("test")
sleep(1) sleep(1)
task = broker.dequeue() task = broker.dequeue()
if not task: if not task:
pytest.skip('SQS being weird') pytest.skip("SQS being weird")
task_id = task[0][0] task_id = task[0][0]
broker.delete(task_id) broker.delete(task_id)
assert broker.dequeue() is None assert broker.dequeue() is None
# fail # fail
broker.enqueue('test') broker.enqueue("test")
while task is None: while task is None:
task = broker.dequeue()[0] task = broker.dequeue()[0]
broker.fail(task[0]) broker.fail(task[0])
# bulk test # bulk test
for i in range(10): for i in range(10):
broker.enqueue('test') broker.enqueue("test")
monkeypatch.setattr(Conf, 'BULK', 12) monkeypatch.setattr(Conf, "BULK", 12)
tasks = broker.dequeue() tasks = broker.dequeue()
for task in tasks: for task in tasks:
assert task is not None assert task is not None
@@ -229,31 +247,31 @@ def canceled_sqs(monkeypatch):
broker.acknowledge(task[0]) broker.acknowledge(task[0])
assert broker.lock_size() == 0 assert broker.lock_size() == 0
# delete queue # delete queue
broker.enqueue('test') broker.enqueue("test")
broker.purge_queue() broker.purge_queue()
broker.delete_queue() broker.delete_queue()
@pytest.mark.django_db @pytest.mark.django_db
def test_orm(monkeypatch): def test_orm(monkeypatch):
monkeypatch.setattr(Conf, 'ORM', 'default') monkeypatch.setattr(Conf, "ORM", "default")
# check broker # check broker
broker = get_broker(list_key='orm_test') broker = get_broker(list_key="orm_test")
assert broker.ping() is True assert broker.ping() is True
assert broker.info() is not None assert broker.info() is not None
# clear before we start # clear before we start
broker.delete_queue() broker.delete_queue()
# async_task # async_task
broker.enqueue('test') broker.enqueue("test")
assert broker.queue_size() == 1 assert broker.queue_size() == 1
# dequeue # dequeue
task = broker.dequeue()[0] task = broker.dequeue()[0]
assert task[1] == 'test' assert task[1] == "test"
broker.acknowledge(task[0]) broker.acknowledge(task[0])
assert broker.queue_size() == 0 assert broker.queue_size() == 0
# Retry test # Retry test
monkeypatch.setattr(Conf, 'RETRY', 1) monkeypatch.setattr(Conf, "RETRY", 1)
broker.enqueue('test') broker.enqueue("test")
assert broker.queue_size() == 1 assert broker.queue_size() == 1
broker.dequeue() broker.dequeue()
assert broker.queue_size() == 0 assert broker.queue_size() == 0
@@ -265,16 +283,16 @@ def test_orm(monkeypatch):
sleep(1.5) sleep(1.5)
assert broker.queue_size() == 0 assert broker.queue_size() == 0
# delete job # delete job
task_id = broker.enqueue('test') task_id = broker.enqueue("test")
broker.delete(task_id) broker.delete(task_id)
assert broker.dequeue() is None assert broker.dequeue() is None
# fail # fail
task_id = broker.enqueue('test') task_id = broker.enqueue("test")
broker.fail(task_id) broker.fail(task_id)
# bulk test # bulk test
for i in range(5): for i in range(5):
broker.enqueue('test') broker.enqueue("test")
monkeypatch.setattr(Conf, 'BULK', 5) monkeypatch.setattr(Conf, "BULK", 5)
tasks = broker.dequeue() tasks = broker.dequeue()
assert broker.lock_size() == Conf.BULK assert broker.lock_size() == Conf.BULK
for task in tasks: for task in tasks:
@@ -285,32 +303,32 @@ def test_orm(monkeypatch):
# test duplicate acknowledge # test duplicate acknowledge
broker.acknowledge(task[0]) broker.acknowledge(task[0])
# delete queue # delete queue
broker.enqueue('test') broker.enqueue("test")
broker.enqueue('test') broker.enqueue("test")
broker.delete_queue() broker.delete_queue()
assert broker.queue_size() == 0 assert broker.queue_size() == 0
@pytest.mark.django_db @pytest.mark.django_db
def test_mongo(monkeypatch): def test_mongo(monkeypatch):
monkeypatch.setattr(Conf, 'MONGO', {'host': '127.0.0.1', 'port': 27017}) monkeypatch.setattr(Conf, "MONGO", {"host": "127.0.0.1", "port": 27017})
# check broker # check broker
broker = get_broker(list_key='mongo_test') broker = get_broker(list_key="mongo_test")
assert broker.ping() is True assert broker.ping() is True
assert broker.info() is not None assert broker.info() is not None
# clear before we start # clear before we start
broker.delete_queue() broker.delete_queue()
# async_task # async_task
broker.enqueue('test') broker.enqueue("test")
assert broker.queue_size() == 1 assert broker.queue_size() == 1
# dequeue # dequeue
task = broker.dequeue()[0] task = broker.dequeue()[0]
assert task[1] == 'test' assert task[1] == "test"
broker.acknowledge(task[0]) broker.acknowledge(task[0])
assert broker.queue_size() == 0 assert broker.queue_size() == 0
# Retry test # Retry test
monkeypatch.setattr(Conf, 'RETRY', 1) monkeypatch.setattr(Conf, "RETRY", 1)
broker.enqueue('test') broker.enqueue("test")
assert broker.queue_size() == 1 assert broker.queue_size() == 1
broker.dequeue() broker.dequeue()
assert broker.queue_size() == 0 assert broker.queue_size() == 0
@@ -322,15 +340,15 @@ def test_mongo(monkeypatch):
sleep(1.5) sleep(1.5)
assert broker.queue_size() == 0 assert broker.queue_size() == 0
# delete job # delete job
task_id = broker.enqueue('test') task_id = broker.enqueue("test")
broker.delete(task_id) broker.delete(task_id)
assert broker.dequeue() is None assert broker.dequeue() is None
# fail # fail
task_id = broker.enqueue('test') task_id = broker.enqueue("test")
broker.fail(task_id) broker.fail(task_id)
# bulk test # bulk test
for i in range(5): for i in range(5):
broker.enqueue('test') broker.enqueue("test")
tasks = [] tasks = []
for i in range(5): for i in range(5):
tasks.append(broker.dequeue()[0]) tasks.append(broker.dequeue()[0])
@@ -343,8 +361,8 @@ def test_mongo(monkeypatch):
# test duplicate acknowledge # test duplicate acknowledge
broker.acknowledge(task[0]) broker.acknowledge(task[0])
# delete queue # delete queue
broker.enqueue('test') broker.enqueue("test")
broker.enqueue('test') broker.enqueue("test")
broker.purge_queue() broker.purge_queue()
broker.delete_queue() broker.delete_queue()
assert broker.queue_size() == 0 assert broker.queue_size() == 0