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"]