File size: 4,367 Bytes
7302343 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | """Tests for the engine FastAPI skeleton (F-0.5).
Covers: /health envelope, error handlers, HMAC permissive/strict modes,
correlation-id roundtrip.
"""
import hmac
import time
from collections.abc import Iterator
from hashlib import sha256
import pytest
from fastapi.testclient import TestClient
from api.config import Settings, get_settings
from api.errors import EngineError
from api.main import app
from api.middleware import HEADER_CORRELATION, HEADER_SIGNATURE, HEADER_TIMESTAMP
@pytest.fixture
def client() -> TestClient:
return TestClient(app)
@pytest.fixture
def prod_client(monkeypatch: pytest.MonkeyPatch) -> Iterator[TestClient]:
"""Client with HMAC strictly enforced — production-like config."""
get_settings.cache_clear()
monkeypatch.setenv("ENV", "production")
monkeypatch.setenv("ENGINE_SHARED_SECRET", "test-secret")
try:
yield TestClient(app)
finally:
get_settings.cache_clear()
def test_health_returns_ok_envelope(client: TestClient) -> None:
r = client.get("/health")
assert r.status_code == 200
body = r.json()
assert body["ok"] is True
assert body["data"]["engine"] == "0.0.1"
assert body["data"]["model_reasoner"] == "gemini-2.5-pro"
assert body["data"]["model_summarizer"] == "gemini-2.5-flash"
def test_correlation_id_roundtrip_when_provided(client: TestClient) -> None:
cid = "test-correlation-12345"
r = client.get("/health", headers={HEADER_CORRELATION: cid})
assert r.headers[HEADER_CORRELATION] == cid
def test_correlation_id_generated_when_absent(client: TestClient) -> None:
r = client.get("/health")
assert HEADER_CORRELATION in r.headers
assert len(r.headers[HEADER_CORRELATION]) > 0
def test_hmac_permissive_in_dev(client: TestClient) -> None:
# Dev mode: no signature header required; request passes.
# /health is in the public path set anyway, but proves the middleware doesn't crash.
assert client.get("/health").status_code == 200
def test_hmac_strict_rejects_missing_signature(prod_client: TestClient) -> None:
# /health is still public — use a non-existent path to exercise middleware.
r = prod_client.post("/feedback", json={"foo": "bar"})
# 401 from HMAC middleware (UNAUTHORIZED envelope) takes precedence over 404.
assert r.status_code == 401
body = r.json()
assert body["ok"] is False
assert body["error"]["code"] == "UNAUTHORIZED"
assert body["error"]["retryable"] is False
def test_hmac_strict_accepts_valid_signature(prod_client: TestClient) -> None:
secret = "test-secret"
body = b'{"hello":"world"}'
timestamp = str(int(time.time()))
signature = hmac.new(
secret.encode("utf-8"),
f"{timestamp}.".encode() + body,
sha256,
).hexdigest()
r = prod_client.post(
"/feedback",
content=body,
headers={
HEADER_SIGNATURE: signature,
HEADER_TIMESTAMP: timestamp,
"content-type": "application/json",
},
)
# Signature passes; downstream 404 because the route doesn't exist yet.
assert r.status_code == 404
def test_hmac_strict_rejects_skewed_timestamp(prod_client: TestClient) -> None:
secret = "test-secret"
body = b"{}"
# 10 minutes in the future — past the 5-minute skew window.
timestamp = str(int(time.time()) + 600)
signature = hmac.new(
secret.encode("utf-8"),
f"{timestamp}.".encode() + body,
sha256,
).hexdigest()
r = prod_client.post(
"/feedback",
content=body,
headers={HEADER_SIGNATURE: signature, HEADER_TIMESTAMP: timestamp},
)
assert r.status_code == 401
assert "skew" in r.json()["error"]["message"]
def test_engine_error_renders_envelope(client: TestClient) -> None:
@app.get("/_test_rate_limited")
async def _boom() -> None:
raise EngineError("RATE_LIMITED", "test-only")
r = client.get("/_test_rate_limited")
assert r.status_code == 429
body = r.json()
assert body == {
"ok": False,
"error": {"code": "RATE_LIMITED", "message": "test-only", "retryable": True},
}
def test_settings_dev_mode_default() -> None:
get_settings.cache_clear()
s = Settings(_env_file=None)
assert s.is_dev is True
assert s.hmac_enforced is False
get_settings.cache_clear()
|