74 lines
2.2 KiB
Python
74 lines
2.2 KiB
Python
from urllib.parse import urlparse
|
|
|
|
import pytest
|
|
from rest_framework.test import APIClient
|
|
|
|
from jobs.services import create_job
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_jobs_use_limit_offset_pagination():
|
|
create_job(job_type="demo.success")
|
|
create_job(job_type="demo.fail")
|
|
create_job(job_type="demo.slow")
|
|
|
|
response = APIClient().get("/api/jobs/?limit=2")
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["count"] == 3
|
|
assert body["next"] is not None
|
|
assert body["previous"] is None
|
|
assert len(body["results"]) == 2
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_jobs_pagination_supports_status_and_type_filters():
|
|
create_job(job_type="demo.success")
|
|
create_job(job_type="demo.fail")
|
|
create_job(job_type="demo.slow")
|
|
|
|
response = APIClient().get("/api/jobs/?status=queued&type=success&limit=10")
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["count"] == 1
|
|
assert body["results"][0]["type"] == "demo.success"
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_global_events_use_cursor_pagination():
|
|
create_job(job_type="demo.success")
|
|
create_job(job_type="demo.fail")
|
|
create_job(job_type="demo.slow")
|
|
|
|
response = APIClient().get("/api/job-events/?limit=2")
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["next"] is not None
|
|
assert body["previous"] is None
|
|
assert len(body["results"]) == 2
|
|
assert body["results"][0]["id"] > body["results"][1]["id"]
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_global_events_next_cursor_returns_older_events():
|
|
create_job(job_type="demo.success")
|
|
create_job(job_type="demo.fail")
|
|
create_job(job_type="demo.slow")
|
|
client = APIClient()
|
|
first_response = client.get("/api/job-events/?limit=2")
|
|
first_body = first_response.json()
|
|
parsed_next = urlparse(first_body["next"])
|
|
next_path = f"{parsed_next.path}?{parsed_next.query}"
|
|
|
|
next_response = client.get(next_path)
|
|
|
|
assert next_response.status_code == 200
|
|
body = next_response.json()
|
|
assert body["next"] is None
|
|
assert body["previous"] is not None
|
|
assert len(body["results"]) == 1
|
|
assert body["results"][0]["id"] < first_body["results"][1]["id"]
|