File size: 4,618 Bytes
c63bc31 ea6cffb c63bc31 | 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 | """Tests for the OpenAI-compatible HTTP API."""
import pytest
from fastapi.testclient import TestClient
from cortex_ai.adapters import MockAdapter
from cortex_ai.api import create_app
from cortex_ai.config import CortexConfig, EngineConfig
def make_client(api_key: str = "", thinking_mode: str = "chat") -> TestClient:
cfg = CortexConfig(engine=EngineConfig(thinking_mode=thinking_mode))
cfg.server.api_key = api_key
return TestClient(create_app(MockAdapter(), cfg))
def test_health_reports_model_and_tools():
body = make_client().get("/health").json()
assert body["status"] == "ok"
assert body["model"] == "Frankenstein-Labs/Cortex-ai"
assert "calculate" in body["tools"]
assert body["identity_interception"] == "deterministic"
def test_identity_question_is_answered_before_model_inference():
class FailingAdapter(MockAdapter):
def generate(self, *args, **kwargs):
raise AssertionError("identity questions must not reach model inference")
cfg = CortexConfig(engine=EngineConfig(thinking_mode="chat"))
client = TestClient(create_app(FailingAdapter(), cfg))
response = client.post(
"/v1/chat/completions",
json={"messages": [{"role": "user", "content": "Qui a créé ce modèle ?"}]},
)
assert response.status_code == 200
assert response.json()["choices"][0]["message"]["content"].startswith(
"Mon créateur est Abdoulaye Coumbassa"
)
def test_english_identity_question_returns_english_canonical_response():
body = make_client().post(
"/v1/chat/completions",
json={"messages": [{"role": "user", "content": "Who is your creator?"}]},
).json()
assert body["choices"][0]["message"]["content"].startswith(
"My creator is Abdoulaye Coumbassa"
)
def test_models_endpoint_lists_cortex():
body = make_client().get("/v1/models").json()
assert body["object"] == "list"
entry = body["data"][0]
assert entry["id"] == "Frankenstein-Labs/Cortex-ai"
assert entry["owned_by"] == "Frankenstein-Labs"
def test_chat_completion_shape():
r = make_client().post(
"/v1/chat/completions",
json={"messages": [{"role": "user", "content": "Bonjour"}]},
)
assert r.status_code == 200
body = r.json()
assert body["object"] == "chat.completion"
assert body["id"].startswith("chatcmpl-")
assert body["choices"][0]["message"]["role"] == "assistant"
assert body["choices"][0]["finish_reason"] == "stop"
assert body["usage"]["total_tokens"] > 0
def test_chat_completion_executes_a_tool():
body = make_client().post(
"/v1/chat/completions",
json={"messages": [{"role": "user", "content": "Combien font 12 * 8 ?"}]},
).json()
assert body["tool_calls"][0]["name"] == "calculate"
assert body["tool_calls"][0]["result"] == "96"
assert body["tool_calls"][0]["ok"] is True
def test_reasoning_is_exposed_in_thinking_mode():
body = make_client(thinking_mode="thinking").post(
"/v1/chat/completions",
json={"messages": [{"role": "user", "content": "Combien font 4 * 4 ?"}]},
).json()
assert body["reasoning_content"]
def test_system_message_overrides_the_default_prompt():
body = make_client().post(
"/v1/chat/completions",
json={
"messages": [
{"role": "system", "content": "Tu es un assistant bref."},
{"role": "user", "content": "Bonjour"},
]
},
).json()
assert body["choices"][0]["message"]["content"]
def test_auth_is_enforced_when_a_key_is_configured():
client = make_client(api_key="secret-de-test")
assert client.post(
"/v1/chat/completions", json={"messages": [{"role": "user", "content": "x"}]}
).status_code == 401
ok = client.post(
"/v1/chat/completions",
json={"messages": [{"role": "user", "content": "x"}]},
headers={"Authorization": "Bearer secret-de-test"},
)
assert ok.status_code == 200
def test_no_auth_required_when_no_key_is_configured():
assert make_client().post(
"/v1/chat/completions", json={"messages": [{"role": "user", "content": "x"}]}
).status_code == 200
def test_empty_messages_are_rejected():
assert make_client().post("/v1/chat/completions", json={"messages": []}).status_code == 400
def test_streaming_is_refused_explicitly():
r = make_client().post(
"/v1/chat/completions",
json={"messages": [{"role": "user", "content": "x"}], "stream": True},
)
assert r.status_code == 400
assert "stream" in r.json()["detail"]
|