133 lines
5.5 KiB
Python
133 lines
5.5 KiB
Python
import hashlib
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
sys.path.insert(0, str(Path(__file__).parents[1]))
|
|
|
|
from app.main import create_app
|
|
|
|
|
|
TOKEN = "unit-5-test-token"
|
|
PDF = b"%PDF-1.4\nunit five\n%%EOF"
|
|
|
|
|
|
def request_for(pdf: bytes = PDF, pages: list[int] | None = None) -> dict:
|
|
return {
|
|
"documentSha256": hashlib.sha256(pdf).hexdigest(),
|
|
"pages": pages or [1, 2],
|
|
"languages": ["es", "en"],
|
|
"dpi": 200,
|
|
"engine": "paddleocr",
|
|
"engineVersion": "3.4.0",
|
|
"runtimeVersion": "3.2.2",
|
|
"configVersion": "ocr-v1",
|
|
"returnLayout": True,
|
|
}
|
|
|
|
|
|
def key_for(request: dict) -> str:
|
|
pages = json.dumps(request["pages"], separators=(",", ":")).encode()
|
|
return f'{request["documentSha256"]}:ocr-v1:{hashlib.sha256(pages).hexdigest()}'
|
|
|
|
|
|
def submit(client: TestClient, request: dict, pdf: bytes = PDF, key: str | None = None):
|
|
return client.post(
|
|
"/v1/jobs",
|
|
headers={"Authorization": f"Bearer {TOKEN}", "Idempotency-Key": key or key_for(request)},
|
|
files={
|
|
"file": ("input.pdf", pdf, "application/pdf"),
|
|
"request": (None, json.dumps(request), "application/json"),
|
|
},
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def client(tmp_path: Path) -> TestClient:
|
|
return TestClient(create_app(TOKEN, tmp_path / "jobs.db", engine_ready=True))
|
|
|
|
|
|
def test_auth_rejects_missing_and_wrong_bearer_and_health_exposes_no_secret(client: TestClient):
|
|
live = client.get("/health/live")
|
|
ready = client.get("/health/ready")
|
|
assert live.json() == {"status": "ok"}
|
|
assert ready.json() == {"ready": True, "queueDepth": 0, "queueCapacity": 3, "concurrency": 1}
|
|
for authorization in (None, "Bearer wrong-token"):
|
|
headers = {"Idempotency-Key": key_for(request_for())}
|
|
if authorization:
|
|
headers["Authorization"] = authorization
|
|
response = client.post(
|
|
"/v1/jobs",
|
|
headers=headers,
|
|
files={"file": ("input.pdf", PDF, "application/pdf"), "request": (None, json.dumps(request_for()))},
|
|
)
|
|
assert response.status_code == 401
|
|
assert response.json()["detail"]["retryable"] is False
|
|
assert TOKEN not in response.text
|
|
|
|
unavailable = TestClient(create_app(TOKEN, ":memory:", engine_ready=False)).get("/health/ready")
|
|
assert unavailable.status_code == 503
|
|
assert unavailable.json()["ready"] is False
|
|
|
|
|
|
def test_auth_submission_is_idempotent_and_conflicting_payload_is_terminal(client: TestClient):
|
|
request = request_for()
|
|
first = submit(client, request)
|
|
repeated = submit(client, request)
|
|
conflict = request_for(pages=[1])
|
|
conflicting = submit(client, conflict, key=key_for(request))
|
|
assert first.status_code == repeated.status_code == 202
|
|
assert first.json() == repeated.json()
|
|
assert first.json()["requestedPages"] == [1, 2]
|
|
assert conflicting.status_code == 409
|
|
assert conflicting.json()["detail"] == {
|
|
"code": "IDEMPOTENCY_CONFLICT",
|
|
"message": "The idempotency key is already bound to another request",
|
|
"retryable": False,
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "value"),
|
|
[("languages", ["en"]), ("dpi", 300), ("engine", "tesseract"), ("returnLayout", False)],
|
|
)
|
|
def test_auth_allowlist_rejects_client_selected_configuration(client: TestClient, field: str, value: object):
|
|
request = request_for()
|
|
request[field] = value
|
|
response = submit(client, request, key="different-key")
|
|
assert response.status_code == 400
|
|
assert response.json()["detail"]["retryable"] is False
|
|
|
|
|
|
def test_auth_limits_corrupt_pdf_and_integrity_mismatch_are_terminal(tmp_path: Path):
|
|
client = TestClient(create_app(TOKEN, tmp_path / "limited.db", max_upload_bytes=8, engine_ready=True))
|
|
regular = TestClient(create_app(TOKEN, tmp_path / "regular.db", engine_ready=True))
|
|
oversized = submit(client, request_for(PDF), PDF)
|
|
too_many = submit(regular, request_for(pages=list(range(1, 102))))
|
|
corrupt = submit(regular, request_for(b"not-pdf"), b"not-pdf")
|
|
wrong_hash = request_for()
|
|
wrong_hash["documentSha256"] = "0" * 64
|
|
mismatch = submit(regular, wrong_hash, key=key_for(wrong_hash))
|
|
assert oversized.status_code == too_many.status_code == 413
|
|
assert corrupt.status_code == mismatch.status_code == 422
|
|
assert mismatch.json()["detail"]["code"] == "INTEGRITY_MISMATCH"
|
|
for response in (oversized, too_many, corrupt, mismatch):
|
|
assert response.json()["detail"]["retryable"] is False
|
|
assert "Retry-After" not in response.headers
|
|
|
|
|
|
def test_auth_queue_pressure_is_retryable_and_status_and_delete_are_authenticated(client: TestClient):
|
|
accepted = [submit(client, request_for(pdf), pdf) for pdf in (PDF, PDF + b"1", PDF + b"2")]
|
|
pressure = submit(client, request_for(PDF + b"3"), PDF + b"3")
|
|
assert [response.status_code for response in accepted] == [202, 202, 202]
|
|
assert pressure.status_code == 429
|
|
assert pressure.json()["detail"]["retryable"] is True
|
|
assert pressure.headers["Retry-After"] == "2"
|
|
job_id = accepted[0].json()["jobId"]
|
|
status = client.get(f"/v1/jobs/{job_id}", headers={"Authorization": f"Bearer {TOKEN}"})
|
|
assert status.json() == {"jobId": job_id, "status": "queued", "completedPages": 0, "totalPages": 2, "error": None}
|
|
assert client.delete(f"/v1/jobs/{job_id}", headers={"Authorization": f"Bearer {TOKEN}"}).status_code == 204
|
|
assert client.delete(f"/v1/jobs/{job_id}", headers={"Authorization": f"Bearer {TOKEN}"}).status_code == 204
|