mirror of
https://github.com/django-q2/django-q2.git
synced 2026-09-15 21:47:53 +08:00
81 lines
2.9 KiB
Python
81 lines
2.9 KiB
Python
from django.urls import reverse
|
|
from django.utils import timezone
|
|
|
|
import pytest
|
|
|
|
from django_q.tasks import schedule
|
|
from django_q.models import Task, Failure, OrmQ
|
|
from django_q.humanhash import uuid
|
|
from django_q.conf import Conf
|
|
from django_q.signing import SignedPackage
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_admin_views(admin_client, monkeypatch):
|
|
monkeypatch.setattr(Conf, 'ORM', 'default')
|
|
s = schedule('schedule.test')
|
|
tag = uuid()
|
|
f = Task.objects.create(
|
|
id=tag[1],
|
|
name=tag[0],
|
|
func='test.fail',
|
|
started=timezone.now(),
|
|
stopped=timezone.now(),
|
|
success=False)
|
|
tag = uuid()
|
|
t = Task.objects.create(
|
|
id=tag[1],
|
|
name=tag[0],
|
|
func='test.success',
|
|
started=timezone.now(),
|
|
stopped=timezone.now(),
|
|
success=True)
|
|
q = OrmQ.objects.create(
|
|
key='test',
|
|
payload=SignedPackage.dumps({'id': 1, 'func': 'test', 'name': 'test'}))
|
|
admin_urls = (
|
|
# schedule
|
|
reverse('admin:django_q_schedule_changelist'),
|
|
reverse('admin:django_q_schedule_add'),
|
|
reverse('admin:django_q_schedule_change', args=(s.id,)),
|
|
reverse('admin:django_q_schedule_history', args=(s.id,)),
|
|
reverse('admin:django_q_schedule_delete', args=(s.id,)),
|
|
# success
|
|
reverse('admin:django_q_success_changelist'),
|
|
reverse('admin:django_q_success_change', args=(t.id,)),
|
|
reverse('admin:django_q_success_history', args=(t.id,)),
|
|
reverse('admin:django_q_success_delete', args=(t.id,)),
|
|
# failure
|
|
reverse('admin:django_q_failure_changelist'),
|
|
reverse('admin:django_q_failure_change', args=(f.id,)),
|
|
reverse('admin:django_q_failure_history', args=(f.id,)),
|
|
reverse('admin:django_q_failure_delete', args=(f.id,)),
|
|
# orm queue
|
|
reverse('admin:django_q_ormq_changelist'),
|
|
reverse('admin:django_q_ormq_change', args=(q.id,)),
|
|
reverse('admin:django_q_ormq_history', args=(q.id,)),
|
|
reverse('admin:django_q_ormq_delete', args=(q.id,)),
|
|
|
|
)
|
|
for url in admin_urls:
|
|
response = admin_client.get(url)
|
|
assert response.status_code == 200
|
|
|
|
# resubmit the failure
|
|
url = reverse('admin:django_q_failure_changelist')
|
|
data = {'action': 'retry_failed',
|
|
'_selected_action': [f.pk]}
|
|
response = admin_client.post(url, data)
|
|
assert response.status_code == 302
|
|
assert Failure.objects.filter(name=f.id).exists() is False
|
|
# change q
|
|
url = reverse('admin:django_q_ormq_change', args=(q.id,))
|
|
data = {'key': 'default', 'payload': 'test', 'lock_0': '2015-09-17', 'lock_1': '14:31:51', '_save': 'Save'}
|
|
response = admin_client.post(url, data)
|
|
assert response.status_code == 302
|
|
# delete q
|
|
url = reverse('admin:django_q_ormq_delete', args=(q.id,))
|
|
data = {'post': 'yes'}
|
|
response = admin_client.post(url, data)
|
|
assert response.status_code == 302
|