File size: 1,888 Bytes
12496fc | 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 | import json
import threading
import urllib.request
import urllib.error
import pytest
from nexora.server import create_server
class Backend:
def complete(self, messages):
return "test response"
def stream(self, messages):
yield "test "
yield "stream"
def test_local_server_auth_schema_stream():
token = "test-token-at-least-16"
server = create_server(Backend(), token, 0)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
url = f"http://127.0.0.1:{server.server_port}"
def post(body, auth=True, origin=None):
headers = {"Content-Type": "application/json"}
if auth:
headers["Authorization"] = "Bearer " + token
if origin:
headers["Origin"] = origin
return urllib.request.urlopen(urllib.request.Request(url + "/v1/chat/completions", json.dumps(body).encode(), headers), timeout=5)
try:
with urllib.request.urlopen(url + "/health") as r:
assert json.load(r)["status"] == "ready"
body = {"messages": [{"role": "user", "content": "hello"}]}
with pytest.raises(urllib.error.HTTPError) as e:
post(body, auth=False)
assert e.value.code == 401
with pytest.raises(urllib.error.HTTPError) as e:
post(body, origin="https://untrusted.example")
assert e.value.code == 403
with post(body) as r:
assert json.load(r)["choices"][0]["message"]["content"] == "test response"
with post({**body, "stream": True}) as r:
text = r.read().decode()
assert "test " in text and "[DONE]" in text
with pytest.raises(urllib.error.HTTPError) as e:
post({"messages": "invalid"})
assert e.value.code == 400
finally:
server.shutdown()
server.server_close()
thread.join(timeout=5)
|