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.
64 lines
2.3 KiB
Python
64 lines
2.3 KiB
Python
"""Unit tests for ``neuronetz_gateway.auth.keys`` (key gen + prefix parsing).
|
|
|
|
SPEC §11/§4.3 key scheme: a full key is ``nz_<random base62>``; the **stored
|
|
prefix** is ``full_key[:PREFIX_LEN]`` (the ``nz_`` namespace plus leading random
|
|
chars) and is used as the Redis cache key / DB lookup. The whole full key is
|
|
argon2id-hashed; the full key is shown once. Part of the 100%-coverage gate on
|
|
``auth/``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from neuronetz_gateway.auth import keys
|
|
from tests._skip import call_or_skip
|
|
|
|
|
|
def test_namespace_and_prefix_len_match_spec() -> None:
|
|
assert keys.KEY_NAMESPACE == "nz_"
|
|
assert keys.PREFIX_LEN == 12
|
|
|
|
|
|
def test_generate_key_shape() -> None:
|
|
gen = call_or_skip(keys.generate_key)
|
|
assert gen.full_key.startswith(keys.KEY_NAMESPACE)
|
|
# Full key = namespace + SECRET_LEN random chars.
|
|
assert len(gen.full_key) == len(keys.KEY_NAMESPACE) + keys.SECRET_LEN
|
|
# Stored prefix is exactly the first PREFIX_LEN chars of the full key and
|
|
# therefore a literal prefix of it (SPEC §4.3 "first 12 chars").
|
|
assert gen.prefix == gen.full_key[: keys.PREFIX_LEN]
|
|
assert len(gen.prefix) == keys.PREFIX_LEN
|
|
assert gen.full_key.startswith(gen.prefix)
|
|
|
|
|
|
def test_generate_key_is_unique() -> None:
|
|
a = call_or_skip(keys.generate_key)
|
|
b = call_or_skip(keys.generate_key)
|
|
assert a.full_key != b.full_key
|
|
assert a.prefix != b.prefix # CSPRNG => prefixes differ with overwhelming prob.
|
|
|
|
|
|
def test_generated_key_is_url_safe_ascii() -> None:
|
|
gen = call_or_skip(keys.generate_key)
|
|
body = gen.full_key[len(keys.KEY_NAMESPACE) :]
|
|
assert body.isascii()
|
|
assert body.isalnum() # base62 alphabet, no separators
|
|
assert " " not in gen.full_key
|
|
|
|
|
|
def test_extract_prefix_roundtrips_generated_key() -> None:
|
|
gen = call_or_skip(keys.generate_key)
|
|
assert call_or_skip(keys.extract_prefix, gen.full_key) == gen.prefix
|
|
|
|
|
|
def test_extract_prefix_rejects_bad_format() -> None:
|
|
# Missing namespace / too short must raise rather than silently truncate.
|
|
with pytest.raises((ValueError, NotImplementedError)):
|
|
keys.extract_prefix("definitely-not-a-key")
|
|
|
|
|
|
def test_extract_prefix_rejects_too_short() -> None:
|
|
with pytest.raises((ValueError, NotImplementedError)):
|
|
keys.extract_prefix("nz_short")
|