Spaces:

Text-to-Document-Generation
/

Docgenie-API

Paused

Docgenie-API / api /tests /reliability /test_reliability.py

Ahadhassan-2003

deploy: update HF Space

dc4e6da 6 months ago

14.1 kB

"""
Reliability tests — repeated correctness, error handling, recovery
===================================================================
3.1  Repeated identical requests → same status code every time
3.2  Invalid-input handling → correct HTTP status codes for each bad-input class
3.3  Recovery after a bad request → next valid request still works
3.4  Health endpoint availability under sustained request fire
3.5  Sustained load test — repeated lightweight calls with spacing
3.6  Error-response contract consistency
"""
import json
import pathlib
import time
import statistics
import pytest
import requests
from tests.conftest import (
    BASE_URL, TIMEOUT, SEED_IMAGE_URL,
    NONEXISTENT_REQUEST_ID, NONEXISTENT_USER_ID,
    MINIMAL_GENERATE_PAYLOAD,
)

ARTIFACTS = pathlib.Path(__file__).parent.parent / "artifacts"
ARTIFACTS.mkdir(exist_ok=True)

N_REPEAT = 4   # repetitions for idempotency / consistency checks
SPACING  = 2   # seconds between calls in sustained load test

_reliability_results: dict = {}


# ---------------------------------------------------------------------------
# Session-end fixture — persist metrics to JSON for compile_results.py
# ---------------------------------------------------------------------------

@pytest.fixture(scope="session", autouse=True)
def _persist_reliability_metrics():
    """Yield during the test session; write metrics JSON afterwards."""
    yield
    try:
        (ARTIFACTS / "reliability_metrics.json").write_text(
            json.dumps(_reliability_results, indent=2)
        )
    except Exception as e:
        print(f"Warning: could not save reliability metrics: {e}")


# ---------------------------------------------------------------------------
# Retry helper — HuggingFace Spaces can throttle and reset connections
# ---------------------------------------------------------------------------

def _get(session: requests.Session, url: str, **kw) -> requests.Response:
    """GET with up to 3 retries on connection errors (500ms backoff each)."""
    for attempt in range(3):
        try:
            return session.get(url, **kw)
        except (requests.ConnectionError, requests.Timeout):
            if attempt == 2:
                raise
            time.sleep(0.5 * (attempt + 1))


def _post(session: requests.Session, url: str, **kw) -> requests.Response:
    """POST with up to 3 retries on connection errors (500ms backoff each)."""
    for attempt in range(3):
        try:
            return session.post(url, **kw)
        except (requests.ConnectionError, requests.Timeout):
            if attempt == 2:
                raise
            time.sleep(0.5 * (attempt + 1))


# ---------------------------------------------------------------------------
# 3.1  Repeated identical requests are consistent
# ---------------------------------------------------------------------------

class TestRepeatedRequestConsistency:
    """The same request should always return the same status code."""

    def test_health_always_returns_200(self, http):
        statuses = [
            _get(http, f"{BASE_URL}/health", timeout=TIMEOUT).status_code
            for _ in range(N_REPEAT)
        ]
        _reliability_results["repeated_health"] = {
            "iterations": N_REPEAT, "statuses": statuses,
            "consistent": len(set(statuses)) == 1,
        }
        assert all(s == 200 for s in statuses), (
            f"Health did not return 200 every time: {statuses}"
        )

    def test_root_always_returns_200(self, http):
        statuses = [
            _get(http, f"{BASE_URL}/", timeout=TIMEOUT).status_code
            for _ in range(N_REPEAT)
        ]
        assert all(s == 200 for s in statuses), (
            f"Root did not return 200 every time: {statuses}"
        )

    def test_unknown_job_always_returns_same_code(self, http):
        url = f"{BASE_URL}/jobs/{NONEXISTENT_REQUEST_ID}/status"
        statuses = [_get(http, url, timeout=TIMEOUT).status_code for _ in range(N_REPEAT)]
        _reliability_results["repeated_job_status"] = {
            "iterations": N_REPEAT, "statuses": statuses,
            "consistent": len(set(statuses)) == 1,
        }
        # Must be consistent (all same code), even if it's 404 or 500
        assert len(set(statuses)) == 1, (
            f"Inconsistent status codes for same unknown job: {statuses}"
        )

    def test_user_jobs_always_returns_200(self, http):
        url = f"{BASE_URL}/jobs/user/{NONEXISTENT_USER_ID}"
        statuses = [_get(http, url, timeout=TIMEOUT).status_code for _ in range(N_REPEAT)]
        assert all(s == 200 for s in statuses), (
            f"User-jobs did not return 200 every time: {statuses}"
        )

    def test_422_always_returned_for_missing_request_id(self, http):
        statuses = [
            _post(http, f"{BASE_URL}/generate/pdf", json={}, timeout=TIMEOUT).status_code
            for _ in range(N_REPEAT)
        ]
        assert all(s == 422 for s in statuses), (
            f"Schema validation should always give 422, got: {statuses}"
        )


# ---------------------------------------------------------------------------
# 3.2  Invalid-input handling — correct codes for each bad-input class
# ---------------------------------------------------------------------------

class TestInvalidInputHandling:
    """Each class of bad input should yield a predictable HTTP code."""

    CASES = [
        # (description, endpoint, method, payload/params, expected_codes)
        (
            "missing_request_id_pdf",
            f"{BASE_URL}/generate/pdf", "POST",
            {"seed_images": [SEED_IMAGE_URL]},
            [422],
        ),
        (
            "missing_request_id_async",
            f"{BASE_URL}/generate/async", "POST",
            {"seed_images": [SEED_IMAGE_URL]},
            [422],
        ),
        (
            "empty_seed_images_pdf",
            f"{BASE_URL}/generate/pdf", "POST",
            {**MINIMAL_GENERATE_PAYLOAD, "seed_images": []},
            [422],
        ),
        (
            "num_solutions_zero_pdf",
            f"{BASE_URL}/generate/pdf", "POST",
            {**MINIMAL_GENERATE_PAYLOAD,
             "prompt_params": {**MINIMAL_GENERATE_PAYLOAD["prompt_params"], "num_solutions": 0}},
            [422],
        ),
        (
            "non_int_user_id",
            f"{BASE_URL}/jobs/user/abc", "GET",
            None,
            [422],
        ),
        (
            "nonexistent_job_status",
            f"{BASE_URL}/jobs/{NONEXISTENT_REQUEST_ID}/status", "GET",
            None,
            [404, 500],   # Supabase 404 or internal 500
        ),
        (
            "nonexistent_request_id_pdf",
            f"{BASE_URL}/generate/pdf", "POST",
            MINIMAL_GENERATE_PAYLOAD,
            [404],
        ),
    ]

    @pytest.mark.parametrize("description,url,method,payload,expected", CASES)
    def test_case(self, http, description, url, method, payload, expected):
        if method == "POST":
            r = _post(http, url, json=payload, timeout=TIMEOUT)
        else:
            r = _get(http, url, timeout=TIMEOUT)

        _reliability_results.setdefault("invalid_input_cases", {})[description] = {
            "status_code": r.status_code,
            "allowed": expected,
            "ok": r.status_code in expected,
        }
        assert r.status_code in expected, (
            f"[{description}] Expected {expected}, got {r.status_code}: {r.text[:200]}"
        )


# ---------------------------------------------------------------------------
# 3.3  Recovery after a bad request
# ---------------------------------------------------------------------------

class TestRecoveryAfterBadRequest:
    """A valid request after a bad request should not be contaminated."""

    def test_health_recovers_after_bad_generate_pdf(self, http):
        # Fire a bad request first
        _post(http, f"{BASE_URL}/generate/pdf", json={}, timeout=TIMEOUT)
        # Immediately hit /health
        r = _get(http, f"{BASE_URL}/health", timeout=TIMEOUT)
        assert r.status_code == 200

    def test_user_jobs_recovers_after_bad_job_status(self, http):
        # Bad request
        _get(http, f"{BASE_URL}/jobs/{NONEXISTENT_REQUEST_ID}/status", timeout=TIMEOUT)
        # Valid request immediately after
        url = f"{BASE_URL}/jobs/user/{NONEXISTENT_USER_ID}"
        r = _get(http, url, timeout=TIMEOUT)
        assert r.status_code == 200

    def test_sequential_mixed_valid_invalid(self, http):
        """Interleave valid and invalid requests — valid ones must always succeed."""
        for i in range(4):
            # Alternate: bad, good, bad, good …
            if i % 2 == 0:
                _post(http, f"{BASE_URL}/generate/async", json={}, timeout=TIMEOUT)
            else:
                r = _get(http, f"{BASE_URL}/health", timeout=TIMEOUT)
                assert r.status_code == 200, (
                    f"Health failed after a bad request at iteration {i}"
                )

        _reliability_results["recovery"] = {"passed": True}


# ---------------------------------------------------------------------------
# 3.4  Health availability while other endpoints are called
# ---------------------------------------------------------------------------

class TestHealthAvailabilityUnderLoad:
    """Health endpoint must remain available while other calls are in-flight."""

    def test_health_available_during_job_status_calls(self, http):
        health_codes = []
        for _ in range(3):
            # Trigger a Supabase lookup (non-trivial)
            _get(http, f"{BASE_URL}/jobs/{NONEXISTENT_REQUEST_ID}/status", timeout=TIMEOUT)
            time.sleep(0.5)   # brief pause so Space doesn't throttle
            # Check health immediately after
            r = _get(http, f"{BASE_URL}/health", timeout=TIMEOUT)
            health_codes.append(r.status_code)

        _reliability_results["health_under_load"] = {
            "health_pings": len(health_codes),
            "health_200s": health_codes.count(200),
        }
        assert all(c == 200 for c in health_codes), (
            f"Health endpoint returned non-200 while load was happening: {health_codes}"
        )


# ---------------------------------------------------------------------------
# 3.5  Sustained load test
# ---------------------------------------------------------------------------

class TestSustainedLoad:
    """
    Fire N sequential /health calls with a small spacing.
    Measures stability: no degradation in response time over time.
    """

    def test_sustained_health_calls(self, http):
        n       = 6
        samples = []
        ok      = 0
        wall_start = time.perf_counter()

        for i in range(n):
            t0 = time.perf_counter()
            r  = _get(http, f"{BASE_URL}/health", timeout=TIMEOUT)
            elapsed = time.perf_counter() - t0
            samples.append(elapsed)
            if r.status_code == 200:
                ok += 1
            if i < n - 1:
                time.sleep(SPACING)

        wall = time.perf_counter() - wall_start
        result = {
            "iterations": n, "ok": ok, "fail": n - ok,
            "success_rate": round(ok / n, 4),
            "min_s":  round(min(samples), 3),
            "mean_s": round(statistics.mean(samples), 3),
            "max_s":  round(max(samples), 3),
            "stdev_s": round(statistics.stdev(samples), 3) if n > 1 else 0,
            "wall_s": round(wall, 3),
        }
        _reliability_results["sustained_load"] = result
        print(f"\n  Sustained load — {result}")

        assert ok == n, f"Expected all {n} requests to succeed, got {ok}"
        assert result["success_rate"] == 1.0


# ---------------------------------------------------------------------------
# 3.6  Error-response contract consistency
# ---------------------------------------------------------------------------

class TestErrorResponseContract:
    """Error responses must always be JSON with a 'detail' field.
    NOTE: A 3-second cooldown is applied at the start of each test so the
    HuggingFace Space can recover after the sustained-load tests above.
    """

    @pytest.fixture(autouse=True)
    def _cooldown(self):
        """Give the Space 3s to recover after sustained-load tests."""
        time.sleep(3)

    def test_422_has_detail_list(self, http):
        r = _post(http, f"{BASE_URL}/generate/pdf", json={}, timeout=TIMEOUT)
        assert r.status_code == 422
        body = r.json()
        assert "detail" in body
        # FastAPI validation errors use a list of error objects
        assert isinstance(body["detail"], list)
        for err in body["detail"]:
            assert "loc"  in err, f"Validation error missing 'loc': {err}"
            assert "msg"  in err, f"Validation error missing 'msg': {err}"
            assert "type" in err, f"Validation error missing 'type': {err}"

    def test_404_has_detail_string(self, http):
        r = _post(
            http, f"{BASE_URL}/generate/pdf",
            json=MINIMAL_GENERATE_PAYLOAD,
            timeout=TIMEOUT,
        )
        if r.status_code == 404:
            body = r.json()
            assert "detail" in body
            assert isinstance(body["detail"], str)

    def test_503_has_detail_if_redis_unavailable(self, http):
        r = _post(
            http, f"{BASE_URL}/generate/async",
            json=MINIMAL_GENERATE_PAYLOAD,
            timeout=TIMEOUT,
        )
        if r.status_code == 503:
            body = r.json()
            assert "detail" in body

    def test_repeated_422_response_is_stable(self, http):
        """The structure of 422 responses must be identical across calls."""
        responses = [
            _post(http, f"{BASE_URL}/generate/pdf", json={}, timeout=TIMEOUT).json()
            for _ in range(3)
        ]
        # All must have "detail" and it must be a list each time
        for body in responses:
            assert "detail" in body
            assert isinstance(body["detail"], list)

	"""
	Reliability tests — repeated correctness, error handling, recovery
	===================================================================

	3.1 Repeated identical requests → same status code every time
	3.2 Invalid-input handling → correct HTTP status codes for each bad-input class
	3.3 Recovery after a bad request → next valid request still works
	3.4 Health endpoint availability under sustained request fire
	3.5 Sustained load test — repeated lightweight calls with spacing
	3.6 Error-response contract consistency
	"""
	import json
	import pathlib
	import time
	import statistics
	import pytest
	import requests
	from tests.conftest import (
	BASE_URL, TIMEOUT, SEED_IMAGE_URL,
	NONEXISTENT_REQUEST_ID, NONEXISTENT_USER_ID,
	MINIMAL_GENERATE_PAYLOAD,
	)

	ARTIFACTS = pathlib.Path(__file__).parent.parent / "artifacts"
	ARTIFACTS.mkdir(exist_ok=True)

	N_REPEAT = 4 # repetitions for idempotency / consistency checks
	SPACING = 2 # seconds between calls in sustained load test

	_reliability_results: dict = {}


	# ---------------------------------------------------------------------------
	# Session-end fixture — persist metrics to JSON for compile_results.py
	# ---------------------------------------------------------------------------

	@pytest.fixture(scope="session", autouse=True)
	def _persist_reliability_metrics():
	"""Yield during the test session; write metrics JSON afterwards."""
	yield
	try:
	(ARTIFACTS / "reliability_metrics.json").write_text(
	json.dumps(_reliability_results, indent=2)
	)
	except Exception as e:
	print(f"Warning: could not save reliability metrics: {e}")


	# ---------------------------------------------------------------------------
	# Retry helper — HuggingFace Spaces can throttle and reset connections
	# ---------------------------------------------------------------------------

	def _get(session: requests.Session, url: str, **kw) -> requests.Response:
	"""GET with up to 3 retries on connection errors (500ms backoff each)."""
	for attempt in range(3):
	try:
	return session.get(url, **kw)
	except (requests.ConnectionError, requests.Timeout):
	if attempt == 2:
	raise
	time.sleep(0.5 * (attempt + 1))


	def _post(session: requests.Session, url: str, **kw) -> requests.Response:
	"""POST with up to 3 retries on connection errors (500ms backoff each)."""
	for attempt in range(3):
	try:
	return session.post(url, **kw)
	except (requests.ConnectionError, requests.Timeout):
	if attempt == 2:
	raise
	time.sleep(0.5 * (attempt + 1))


	# ---------------------------------------------------------------------------
	# 3.1 Repeated identical requests are consistent
	# ---------------------------------------------------------------------------

	class TestRepeatedRequestConsistency:
	"""The same request should always return the same status code."""

	def test_health_always_returns_200(self, http):
	statuses = [
	_get(http, f"{BASE_URL}/health", timeout=TIMEOUT).status_code
	for _ in range(N_REPEAT)
	]
	_reliability_results["repeated_health"] = {
	"iterations": N_REPEAT, "statuses": statuses,
	"consistent": len(set(statuses)) == 1,
	}
	assert all(s == 200 for s in statuses), (
	f"Health did not return 200 every time: {statuses}"
	)

	def test_root_always_returns_200(self, http):
	statuses = [
	_get(http, f"{BASE_URL}/", timeout=TIMEOUT).status_code
	for _ in range(N_REPEAT)
	]
	assert all(s == 200 for s in statuses), (
	f"Root did not return 200 every time: {statuses}"
	)

	def test_unknown_job_always_returns_same_code(self, http):
	url = f"{BASE_URL}/jobs/{NONEXISTENT_REQUEST_ID}/status"
	statuses = [_get(http, url, timeout=TIMEOUT).status_code for _ in range(N_REPEAT)]
	_reliability_results["repeated_job_status"] = {
	"iterations": N_REPEAT, "statuses": statuses,
	"consistent": len(set(statuses)) == 1,
	}
	# Must be consistent (all same code), even if it's 404 or 500
	assert len(set(statuses)) == 1, (
	f"Inconsistent status codes for same unknown job: {statuses}"
	)

	def test_user_jobs_always_returns_200(self, http):
	url = f"{BASE_URL}/jobs/user/{NONEXISTENT_USER_ID}"
	statuses = [_get(http, url, timeout=TIMEOUT).status_code for _ in range(N_REPEAT)]
	assert all(s == 200 for s in statuses), (
	f"User-jobs did not return 200 every time: {statuses}"
	)

	def test_422_always_returned_for_missing_request_id(self, http):
	statuses = [
	_post(http, f"{BASE_URL}/generate/pdf", json={}, timeout=TIMEOUT).status_code
	for _ in range(N_REPEAT)
	]
	assert all(s == 422 for s in statuses), (
	f"Schema validation should always give 422, got: {statuses}"
	)


	# ---------------------------------------------------------------------------
	# 3.2 Invalid-input handling — correct codes for each bad-input class
	# ---------------------------------------------------------------------------

	class TestInvalidInputHandling:
	"""Each class of bad input should yield a predictable HTTP code."""

	CASES = [
	# (description, endpoint, method, payload/params, expected_codes)
	(
	"missing_request_id_pdf",
	f"{BASE_URL}/generate/pdf", "POST",
	{"seed_images": [SEED_IMAGE_URL]},
	[422],
	),
	(
	"missing_request_id_async",
	f"{BASE_URL}/generate/async", "POST",
	{"seed_images": [SEED_IMAGE_URL]},
	[422],
	),
	(
	"empty_seed_images_pdf",
	f"{BASE_URL}/generate/pdf", "POST",
	{**MINIMAL_GENERATE_PAYLOAD, "seed_images": []},
	[422],
	),
	(
	"num_solutions_zero_pdf",
	f"{BASE_URL}/generate/pdf", "POST",
	{**MINIMAL_GENERATE_PAYLOAD,
	"prompt_params": {**MINIMAL_GENERATE_PAYLOAD["prompt_params"], "num_solutions": 0}},
	[422],
	),
	(
	"non_int_user_id",
	f"{BASE_URL}/jobs/user/abc", "GET",
	None,
	[422],
	),
	(
	"nonexistent_job_status",
	f"{BASE_URL}/jobs/{NONEXISTENT_REQUEST_ID}/status", "GET",
	None,
	[404, 500], # Supabase 404 or internal 500
	),
	(
	"nonexistent_request_id_pdf",
	f"{BASE_URL}/generate/pdf", "POST",
	MINIMAL_GENERATE_PAYLOAD,
	[404],
	),
	]

	@pytest.mark.parametrize("description,url,method,payload,expected", CASES)
	def test_case(self, http, description, url, method, payload, expected):
	if method == "POST":
	r = _post(http, url, json=payload, timeout=TIMEOUT)
	else:
	r = _get(http, url, timeout=TIMEOUT)

	_reliability_results.setdefault("invalid_input_cases", {})[description] = {
	"status_code": r.status_code,
	"allowed": expected,
	"ok": r.status_code in expected,
	}
	assert r.status_code in expected, (
	f"[{description}] Expected {expected}, got {r.status_code}: {r.text[:200]}"
	)


	# ---------------------------------------------------------------------------
	# 3.3 Recovery after a bad request
	# ---------------------------------------------------------------------------

	class TestRecoveryAfterBadRequest:
	"""A valid request after a bad request should not be contaminated."""

	def test_health_recovers_after_bad_generate_pdf(self, http):
	# Fire a bad request first
	_post(http, f"{BASE_URL}/generate/pdf", json={}, timeout=TIMEOUT)
	# Immediately hit /health
	r = _get(http, f"{BASE_URL}/health", timeout=TIMEOUT)
	assert r.status_code == 200

	def test_user_jobs_recovers_after_bad_job_status(self, http):
	# Bad request
	_get(http, f"{BASE_URL}/jobs/{NONEXISTENT_REQUEST_ID}/status", timeout=TIMEOUT)
	# Valid request immediately after
	url = f"{BASE_URL}/jobs/user/{NONEXISTENT_USER_ID}"
	r = _get(http, url, timeout=TIMEOUT)
	assert r.status_code == 200

	def test_sequential_mixed_valid_invalid(self, http):
	"""Interleave valid and invalid requests — valid ones must always succeed."""
	for i in range(4):
	# Alternate: bad, good, bad, good …
	if i % 2 == 0:
	_post(http, f"{BASE_URL}/generate/async", json={}, timeout=TIMEOUT)
	else:
	r = _get(http, f"{BASE_URL}/health", timeout=TIMEOUT)
	assert r.status_code == 200, (
	f"Health failed after a bad request at iteration {i}"
	)

	_reliability_results["recovery"] = {"passed": True}


	# ---------------------------------------------------------------------------
	# 3.4 Health availability while other endpoints are called
	# ---------------------------------------------------------------------------

	class TestHealthAvailabilityUnderLoad:
	"""Health endpoint must remain available while other calls are in-flight."""

	def test_health_available_during_job_status_calls(self, http):
	health_codes = []
	for _ in range(3):
	# Trigger a Supabase lookup (non-trivial)
	_get(http, f"{BASE_URL}/jobs/{NONEXISTENT_REQUEST_ID}/status", timeout=TIMEOUT)
	time.sleep(0.5) # brief pause so Space doesn't throttle
	# Check health immediately after
	r = _get(http, f"{BASE_URL}/health", timeout=TIMEOUT)
	health_codes.append(r.status_code)

	_reliability_results["health_under_load"] = {
	"health_pings": len(health_codes),
	"health_200s": health_codes.count(200),
	}
	assert all(c == 200 for c in health_codes), (
	f"Health endpoint returned non-200 while load was happening: {health_codes}"
	)


	# ---------------------------------------------------------------------------
	# 3.5 Sustained load test
	# ---------------------------------------------------------------------------

	class TestSustainedLoad:
	"""
	Fire N sequential /health calls with a small spacing.
	Measures stability: no degradation in response time over time.
	"""

	def test_sustained_health_calls(self, http):
	n = 6
	samples = []
	ok = 0
	wall_start = time.perf_counter()

	for i in range(n):
	t0 = time.perf_counter()
	r = _get(http, f"{BASE_URL}/health", timeout=TIMEOUT)
	elapsed = time.perf_counter() - t0
	samples.append(elapsed)
	if r.status_code == 200:
	ok += 1
	if i < n - 1:
	time.sleep(SPACING)

	wall = time.perf_counter() - wall_start
	result = {
	"iterations": n, "ok": ok, "fail": n - ok,
	"success_rate": round(ok / n, 4),
	"min_s": round(min(samples), 3),
	"mean_s": round(statistics.mean(samples), 3),
	"max_s": round(max(samples), 3),
	"stdev_s": round(statistics.stdev(samples), 3) if n > 1 else 0,
	"wall_s": round(wall, 3),
	}
	_reliability_results["sustained_load"] = result
	print(f"\n Sustained load — {result}")

	assert ok == n, f"Expected all {n} requests to succeed, got {ok}"
	assert result["success_rate"] == 1.0


	# ---------------------------------------------------------------------------
	# 3.6 Error-response contract consistency
	# ---------------------------------------------------------------------------

	class TestErrorResponseContract:
	"""Error responses must always be JSON with a 'detail' field.

	NOTE: A 3-second cooldown is applied at the start of each test so the
	HuggingFace Space can recover after the sustained-load tests above.
	"""

	@pytest.fixture(autouse=True)
	def _cooldown(self):
	"""Give the Space 3s to recover after sustained-load tests."""
	time.sleep(3)

	def test_422_has_detail_list(self, http):
	r = _post(http, f"{BASE_URL}/generate/pdf", json={}, timeout=TIMEOUT)
	assert r.status_code == 422
	body = r.json()
	assert "detail" in body
	# FastAPI validation errors use a list of error objects
	assert isinstance(body["detail"], list)
	for err in body["detail"]:
	assert "loc" in err, f"Validation error missing 'loc': {err}"
	assert "msg" in err, f"Validation error missing 'msg': {err}"
	assert "type" in err, f"Validation error missing 'type': {err}"

	def test_404_has_detail_string(self, http):
	r = _post(
	http, f"{BASE_URL}/generate/pdf",
	json=MINIMAL_GENERATE_PAYLOAD,
	timeout=TIMEOUT,
	)
	if r.status_code == 404:
	body = r.json()
	assert "detail" in body
	assert isinstance(body["detail"], str)

	def test_503_has_detail_if_redis_unavailable(self, http):
	r = _post(
	http, f"{BASE_URL}/generate/async",
	json=MINIMAL_GENERATE_PAYLOAD,
	timeout=TIMEOUT,
	)
	if r.status_code == 503:
	body = r.json()
	assert "detail" in body

	def test_repeated_422_response_is_stable(self, http):
	"""The structure of 422 responses must be identical across calls."""
	responses = [
	_post(http, f"{BASE_URL}/generate/pdf", json={}, timeout=TIMEOUT).json()
	for _ in range(3)
	]
	# All must have "detail" and it must be a list each time
	for body in responses:
	assert "detail" in body
	assert isinstance(body["detail"], list)