70 lines
2.2 KiB
Python
70 lines
2.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Deterministic Rekor error-mode shim used by negative path test lane."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RekorCase:
|
|
case_id: str
|
|
status_code: int
|
|
entry_type: str
|
|
expected_code: str
|
|
message: str
|
|
response_body: dict[str, object]
|
|
|
|
|
|
def default_cases() -> list[RekorCase]:
|
|
return [
|
|
RekorCase(
|
|
case_id="oversized-payload-413",
|
|
status_code=413,
|
|
entry_type="intoto",
|
|
expected_code="payload_too_large",
|
|
message="payload size exceeds configured limit",
|
|
response_body={"error": "payload too large", "maxBytes": 10_000_000},
|
|
),
|
|
RekorCase(
|
|
case_id="unsupported-entry-type-400",
|
|
status_code=400,
|
|
entry_type="unknown",
|
|
expected_code="unsupported_entry_type",
|
|
message="unsupported entry type",
|
|
response_body={"error": "unsupported entry type", "entryType": "unknown"},
|
|
),
|
|
RekorCase(
|
|
case_id="failed-dependency-424",
|
|
status_code=424,
|
|
entry_type="intoto",
|
|
expected_code="failed_dependency",
|
|
message="rekor backend dependency failure",
|
|
response_body={"error": "ledger gap", "reprocessToken": "rekor-gap-001"},
|
|
),
|
|
RekorCase(
|
|
case_id="gateway-timeout-504",
|
|
status_code=504,
|
|
entry_type="intoto",
|
|
expected_code="upstream_timeout",
|
|
message="rekor upstream timeout",
|
|
response_body={"error": "timeout", "retryAfterSeconds": 30},
|
|
),
|
|
RekorCase(
|
|
case_id="accepted-for-reprocess-202",
|
|
status_code=202,
|
|
entry_type="intoto",
|
|
expected_code="reprocess_pending",
|
|
message="accepted for asynchronous replay",
|
|
response_body={"status": "accepted", "reprocessToken": "rekor-async-001"},
|
|
),
|
|
]
|
|
|
|
|
|
def simulate_submit(case: RekorCase) -> tuple[int, dict[str, object], dict[str, str]]:
|
|
headers = {
|
|
"x-correlation-id": f"corr-{case.case_id}",
|
|
"content-type": "application/json",
|
|
}
|
|
return case.status_code, dict(case.response_body), headers
|