mirror of
https://github.com/django-q2/django-q2.git
synced 2026-09-23 09:38:12 +08:00
Ilan Steemers
439c8a2f77
This way only qmonitor and qinfo need Curses and you can still run the cluster without it.
47 lines
963 B
Python
47 lines
963 B
Python
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
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_monitor():
|
|
assert Stat.get(0).sentinel == 0
|
|
c = Cluster()
|
|
c.start()
|
|
stats = monitor(run_once=True)
|
|
c.stop()
|
|
assert len(stats) > 0
|
|
found_c = False
|
|
for stat in stats:
|
|
if stat.cluster_id == c.pid:
|
|
found_c = True
|
|
assert stat.uptime() > 0
|
|
assert stat.empty_queues() is True
|
|
break
|
|
assert found_c is True
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_info():
|
|
info()
|
|
do_sync()
|
|
info()
|
|
for _ in range(24):
|
|
do_sync()
|
|
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)
|