llm-connect/tests/test_server.py

"""
Tests for LLMServer HTTP serve mode (FR-1).
"""

import json
import urllib.error
import urllib.request

import pytest

from llm_connect.adapter import MockLLMAdapter, ErrorLLMAdapter
from llm_connect.models import RunConfig
from llm_connect.server import LLMServer


@pytest.fixture()
def server():
    """Start a server on a free port; stop after each test."""
    s = LLMServer(adapter=MockLLMAdapter(mock_response="hello world"), port=0)
    s.start()
    yield s
    s.stop()


def _get(url: str) -> tuple[int, dict]:
    try:
        with urllib.request.urlopen(url) as resp:
            return resp.status, json.loads(resp.read())
    except urllib.error.HTTPError as exc:
        return exc.code, json.loads(exc.read())


def _post(url: str, body: dict) -> tuple[int, dict]:
    payload = json.dumps(body).encode()
    req = urllib.request.Request(
        url,
        data=payload,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req) as resp:
            return resp.status, json.loads(resp.read())
    except urllib.error.HTTPError as exc:
        return exc.code, json.loads(exc.read())


class TestHealth:
    def test_health_returns_200(self, server):
        status, body = _get(f"http://127.0.0.1:{server.port}/health")
        assert status == 200
        assert body["status"] == "ok"

    def test_unknown_get_returns_404(self, server):
        status, body = _get(f"http://127.0.0.1:{server.port}/nope")
        assert status == 404


class TestExecute:
    def test_post_execute_round_trip(self, server):
        status, body = _post(
            f"http://127.0.0.1:{server.port}/execute",
            {"prompt": "say hello"},
        )
        assert status == 200
        assert body["content"] == "hello world"
        assert body["finish_reason"] == "stop"

    def test_response_includes_usage(self, server):
        status, body = _post(
            f"http://127.0.0.1:{server.port}/execute",
            {"prompt": "count tokens"},
        )
        assert status == 200
        assert "usage" in body
        assert body["usage"]["total_tokens"] > 0

    def test_missing_prompt_returns_400(self, server):
        status, body = _post(
            f"http://127.0.0.1:{server.port}/execute",
            {"config": {}},
        )
        assert status == 400
        assert "prompt" in body["error"]

    def test_invalid_json_returns_400(self, server):
        req = urllib.request.Request(
            f"http://127.0.0.1:{server.port}/execute",
            data=b"not json",
            headers={"Content-Type": "application/json"},
            method="POST",
        )
        try:
            with urllib.request.urlopen(req) as resp:
                status, body = resp.status, json.loads(resp.read())
        except urllib.error.HTTPError as exc:
            status, body = exc.code, json.loads(exc.read())
        assert status == 400

    def test_unknown_post_path_returns_404(self, server):
        status, body = _post(
            f"http://127.0.0.1:{server.port}/wrong",
            {"prompt": "hi"},
        )
        assert status == 404

    def test_adapter_error_returns_500(self):
        s = LLMServer(adapter=ErrorLLMAdapter("boom"), port=0)
        s.start()
        try:
            status, body = _post(
                f"http://127.0.0.1:{s.port}/execute",
                {"prompt": "hello"},
            )
            assert status == 500
            assert "boom" in body["error"]
        finally:
            s.stop()

    def test_config_fields_forwarded(self):
        """Config fields in request body reach the adapter via RunConfig."""
        adapter = MockLLMAdapter(mock_response="x")
        s = LLMServer(adapter=adapter, port=0)
        s.start()
        try:
            status, body = _post(
                f"http://127.0.0.1:{s.port}/execute",
                {"prompt": "hi", "config": {"model_name": "gpt-3.5-turbo", "max_tokens": 100}},
            )
            assert status == 200
            assert adapter.last_config.model_name == "gpt-3.5-turbo"
            assert adapter.last_config.max_tokens == 100
        finally:
            s.stop()