tests/test_server.py

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)