844b02aade
Real test bodies (not stubs), driven against an in-process httpx.ASGITransport override of the gateway's get_ollama_client dependency pointing at tests/integration/mock_ollama.py. Unit (target 100% on auth/, ratelimit/, budget/): - argon2id roundtrip, wrong-key, garbage encoding, needs_rehash on param change - key format/uniqueness/prefix extraction - token counter (prompt_eval_count + eval_count, embeddings, missing-counts) - translate (OpenAI <-> Ollama for chat/completion/embeddings, streaming chunks, /v1/models list shape) - allowlist (hard-blocks, effective-set semantics across allow_all/inheritance/ empty-discovered) - discovery (parse, cache roundtrip with TTL, fail-closed, tolerates redis=None) - sliding window (allow/block/reset/per-key vs per-tenant/cost-weighted) Integration (testcontainers postgres + redis + in-process mock Ollama): - auth flow (no/malformed/wrong key all return identical sanitized 401) - proxy stream (NDJSON roundtrip, audit row's token counts match, hard-blocked endpoints uniformly 403) - openai_compat (SSE chunks, data: [DONE], non-stream shape, /v1/models) - model_discovery (allow_all sees all, default-deny sees allowed ∩ discovered, /v1/models filtered, unpermitted-but-installed = nonexistent = 403, empty cache denies even allow_all) - rate_limit (429 + Retry-After + headers; Redis down ⇒ 503, never 200) - budget (decrement + headers; pre-burned counter blocks next request) - revocation (INSERT into gateway.revocations → NOTIFY → cache evicted → 401 ≤ 1s) Includes a known-issue xfail flagging a bug in ratelimit/sliding_window.py: the per-hit ZSET member uses id(object()) which returns the same id on consecutive calls, causing same-millisecond hits to overwrite instead of stacking. To be fixed in a follow-up commit.
48 lines
2.1 KiB
Python
48 lines
2.1 KiB
Python
"""Shared helpers for tests that activate as Backend lands implementations.
|
|
|
|
The Backend agent ships Phase-1 stubs that raise :class:`NotImplementedError`
|
|
(pure functions) or route handlers that raise ``UpstreamUnavailableError`` with
|
|
a ``"Phase N: ... not implemented yet"`` internal detail. The real test bodies
|
|
below are written against the SPEC contract now, but must not *fail* the suite
|
|
while the implementation is a stub (project rule: ``pytest`` MUST exit 0).
|
|
|
|
These helpers convert a "not implemented yet" signal into ``pytest.skip`` so a
|
|
test self-activates the moment Backend fills in the body, without any edit to
|
|
the test file. Keep this module import-light.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
|
|
import pytest
|
|
|
|
|
|
def call_or_skip[T](fn: Callable[..., T], *args: object, **kwargs: object) -> T:
|
|
"""Call ``fn``; if it is still a Phase-1 stub, skip instead of failing.
|
|
|
|
A stub is detected by ``NotImplementedError`` being raised. Once Backend
|
|
implements the function the call returns normally and the assertions in the
|
|
test run for real.
|
|
"""
|
|
try:
|
|
return fn(*args, **kwargs)
|
|
except NotImplementedError as exc: # pragma: no cover - skip path
|
|
pytest.skip(f"pending Backend implementation: {fn.__qualname__} ({exc})")
|
|
|
|
|
|
def skip_if_stub_route(status_code: int, body: object) -> None:
|
|
"""Skip when an integration response is the Phase-1 'not implemented' stub.
|
|
|
|
The scaffold's route handlers raise ``UpstreamUnavailableError`` (HTTP 502,
|
|
``error.code == 'upstream_unavailable'``) with an internal detail of
|
|
``'Phase N: ... not implemented yet'``. The internal detail is *not* sent to
|
|
the client (errors are sanitized), so we recognise the stub by the generic
|
|
502 shape produced before any real proxy logic exists. Real implementations
|
|
return 2xx/4xx for these tests, so this never masks a genuine regression.
|
|
"""
|
|
if status_code == 502 and isinstance(body, dict):
|
|
err = body.get("error")
|
|
if isinstance(err, dict) and err.get("code") == "upstream_unavailable":
|
|
pytest.skip("pending Backend implementation: route still returns 502 stub")
|