mirror of
https://github.com/django-q2/django-q2.git
synced 2026-09-15 21:47:53 +08:00
15155c7a99
* Black linting * Adding Black to dev dependencies and upping minimal python to 3.6.2 for compatibility * Updating packages * Removing pip-tools input and exporting requirements with poetry * Deleting old setup files and test runner * Trying 1.3.7 * Looser extras requirements to prevent conflicts * Sorted imports with isort * Added iSort to dev dependencies * Fixes localtime for naive setups
53 lines
1.2 KiB
Python
53 lines
1.2 KiB
Python
import uuid
|
|
|
|
import pytest
|
|
|
|
from django_q.brokers import get_broker
|
|
from django_q.cluster import Cluster
|
|
from django_q.conf import Conf
|
|
from django_q.monitor import get_ids, info, monitor
|
|
from django_q.status import Stat
|
|
from django_q.tasks import async_task
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_monitor(monkeypatch):
|
|
cluster_id = uuid.uuid4()
|
|
assert Stat.get(pid=0, cluster_id=cluster_id).sentinel == 0
|
|
c = Cluster()
|
|
c.start()
|
|
stats = monitor(run_once=True)
|
|
assert get_ids() is True
|
|
c.stop()
|
|
assert len(stats) > 0
|
|
found_c = False
|
|
for stat in stats:
|
|
if stat.cluster_id == c.cluster_id:
|
|
found_c = True
|
|
assert stat.uptime() > 0
|
|
assert stat.empty_queues() is True
|
|
break
|
|
assert found_c
|
|
# test lock size
|
|
monkeypatch.setattr(Conf, "ORM", "default")
|
|
b = get_broker("monitor_test")
|
|
b.enqueue("test")
|
|
b.dequeue()
|
|
assert b.lock_size() == 1
|
|
monitor(run_once=True, broker=b)
|
|
b.delete_queue()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_info():
|
|
info()
|
|
do_sync()
|
|
info()
|
|
for _ in range(24):
|
|
do_sync()
|
|
info()
|
|
|
|
|
|
def do_sync():
|
|
async_task("django_q.tests.tasks.countdown", 1, sync=True, save=True)
|