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
87 lines
2.9 KiB
Python
87 lines
2.9 KiB
Python
import pytest
|
|
from django.urls import reverse
|
|
from django.utils import timezone
|
|
|
|
from django_q.conf import Conf
|
|
from django_q.humanhash import uuid
|
|
from django_q.models import Failure, OrmQ, Task
|
|
from django_q.signing import SignedPackage
|
|
from django_q.tasks import schedule
|
|
|
|
|
|
@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
|