| """ |
| Functional tests — GET /jobs/user/{user_id} |
| ============================================ |
| Tests for the per-user job listing endpoint. |
| |
| Key behaviours: |
| • Returns valid JSON for any integer user_id |
| • Required response fields: user_id, jobs, count, limit, offset |
| • Pagination params (limit / offset) are respected |
| • limit is capped at 100 by the server |
| • Non-integer user_id → 422 |
| """ |
| import pytest |
| import requests |
| from tests.conftest import BASE_URL, TIMEOUT, NONEXISTENT_USER_ID |
|
|
| ENDPOINT_TPL = f"{BASE_URL}/jobs/user/{{user_id}}" |
|
|
|
|
| class TestUserJobsEndpoint: |
| """GET /jobs/user/{user_id}""" |
|
|
| |
|
|
| def test_returns_200_for_any_user(self, http): |
| url = ENDPOINT_TPL.format(user_id=NONEXISTENT_USER_ID) |
| r = http.get(url, timeout=TIMEOUT) |
| assert r.status_code == 200, ( |
| f"Expected 200 for a user with no jobs, got {r.status_code}: {r.text}" |
| ) |
|
|
| def test_response_is_json(self, http): |
| url = ENDPOINT_TPL.format(user_id=NONEXISTENT_USER_ID) |
| r = http.get(url, timeout=TIMEOUT) |
| assert "application/json" in r.headers.get("Content-Type", "") |
|
|
| |
|
|
| def test_response_has_required_fields(self, http): |
| url = ENDPOINT_TPL.format(user_id=NONEXISTENT_USER_ID) |
| body = http.get(url, timeout=TIMEOUT).json() |
| for field in ("user_id", "jobs", "count", "limit", "offset"): |
| assert field in body, f"Missing required field '{field}'. Got: {list(body.keys())}" |
|
|
| def test_jobs_is_a_list(self, http): |
| url = ENDPOINT_TPL.format(user_id=NONEXISTENT_USER_ID) |
| body = http.get(url, timeout=TIMEOUT).json() |
| assert isinstance(body["jobs"], list) |
|
|
| def test_count_matches_jobs_length(self, http): |
| url = ENDPOINT_TPL.format(user_id=NONEXISTENT_USER_ID) |
| body = http.get(url, timeout=TIMEOUT).json() |
| assert body["count"] == len(body["jobs"]), ( |
| f"count={body['count']} does not match len(jobs)={len(body['jobs'])}" |
| ) |
|
|
| def test_user_id_echoed_in_response(self, http): |
| url = ENDPOINT_TPL.format(user_id=NONEXISTENT_USER_ID) |
| body = http.get(url, timeout=TIMEOUT).json() |
| assert body["user_id"] == NONEXISTENT_USER_ID, ( |
| f"Response user_id {body['user_id']} != requested {NONEXISTENT_USER_ID}" |
| ) |
|
|
| |
|
|
| def test_default_limit_is_50(self, http): |
| url = ENDPOINT_TPL.format(user_id=NONEXISTENT_USER_ID) |
| body = http.get(url, timeout=TIMEOUT).json() |
| assert body["limit"] == 50 |
|
|
| def test_default_offset_is_0(self, http): |
| url = ENDPOINT_TPL.format(user_id=NONEXISTENT_USER_ID) |
| body = http.get(url, timeout=TIMEOUT).json() |
| assert body["offset"] == 0 |
|
|
| def test_custom_limit_is_respected(self, http): |
| url = ENDPOINT_TPL.format(user_id=NONEXISTENT_USER_ID) |
| body = http.get(url, params={"limit": 10}, timeout=TIMEOUT).json() |
| assert body["limit"] == 10 |
|
|
| def test_custom_offset_is_respected(self, http): |
| url = ENDPOINT_TPL.format(user_id=NONEXISTENT_USER_ID) |
| body = http.get(url, params={"offset": 5}, timeout=TIMEOUT).json() |
| assert body["offset"] == 5 |
|
|
| def test_limit_above_100_is_capped(self, http): |
| """Server silently caps limit at 100.""" |
| url = ENDPOINT_TPL.format(user_id=NONEXISTENT_USER_ID) |
| body = http.get(url, params={"limit": 500}, timeout=TIMEOUT).json() |
| assert body["limit"] <= 100, ( |
| f"limit should be capped at 100, got {body['limit']}" |
| ) |
|
|
| |
|
|
| def test_non_integer_user_id_returns_422(self, http): |
| url = f"{BASE_URL}/jobs/user/not-an-int" |
| r = http.get(url, timeout=TIMEOUT) |
| assert r.status_code == 422, ( |
| f"Non-integer user_id should give 422, got {r.status_code}" |
| ) |
|
|
| def test_endpoint_is_get_only(self, http): |
| url = ENDPOINT_TPL.format(user_id=NONEXISTENT_USER_ID) |
| r = http.post(url, json={}, timeout=TIMEOUT) |
| assert r.status_code == 405 |
|
|
