80 lines
2.6 KiB
Python
80 lines
2.6 KiB
Python
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from unittest.mock import patch, AsyncMock
|
|
from backend.main import app
|
|
|
|
client = TestClient(app)
|
|
|
|
|
|
class TestHealthInfoEndpoints:
|
|
"""
|
|
Check the availability of health/info endpoints
|
|
and the validity of the model configuration.
|
|
"""
|
|
|
|
@pytest.mark.parametrize("route", ["/", "/info"])
|
|
def test_health_info_returns_200(self, route) -> None:
|
|
resp = client.get(route)
|
|
assert resp.status_code == 200
|
|
|
|
def test_info_models_config_valid(self) -> None:
|
|
resp = client.get("/info")
|
|
config = resp.json()["config"]
|
|
assert "default_model" in config
|
|
assert isinstance(config["available_models"], list)
|
|
assert len(config["available_models"]) > 0
|
|
assert config["default_model"] in config["available_models"]
|
|
|
|
|
|
class TestRunAgentEndpoint:
|
|
"""
|
|
Checks /api/run: SSE response, default model,
|
|
validation and message generation.
|
|
"""
|
|
|
|
def test_run_sse_contract(self, mock_agent_task) -> None:
|
|
resp = client.post("/api/run", json={
|
|
"query": "test",
|
|
"model": "gpt-oss-120b"
|
|
})
|
|
assert resp.status_code == 200
|
|
assert "text/event-stream" in resp.headers["content-type"]
|
|
assert "data:" in resp.text
|
|
assert "data: [DONE]" in resp.text
|
|
|
|
|
|
def test_run_default_model_used_when_not_specified(self, mock_agent_task) -> None:
|
|
from configs import DEFAULT_MODEL
|
|
resp = client.post("/api/run", json={"query": "test"})
|
|
assert resp.status_code == 200
|
|
|
|
|
|
def test_run_missing_query_returns_422(self) -> None:
|
|
resp = client.post("/api/run", json={"model": "gemini-2.5-flash"})
|
|
assert resp.status_code == 422
|
|
|
|
|
|
def test_run_single_message_no_history(self) -> None:
|
|
from backend.routers.run_agent import build_messages, Query
|
|
q = Query(query="test", model="gpt-oss-120b")
|
|
messages = build_messages(q)
|
|
assert len(messages) == 1
|
|
assert messages[0] == {"role": "user", "content": "test"}
|
|
|
|
|
|
def test_run_history_prepended_before_query(self) -> None:
|
|
from backend.routers.run_agent import build_messages, Query, Message
|
|
q = Query(
|
|
query="a ďalej?",
|
|
model="gpt-oss-120b",
|
|
history=[
|
|
Message(role="user", content="prvá otázka"),
|
|
Message(role="assistant", content="prvá odpoveď"),
|
|
]
|
|
)
|
|
messages = build_messages(q)
|
|
assert len(messages) == 3
|
|
assert messages[0]["role"] == "user"
|
|
assert messages[1]["role"] == "assistant"
|
|
assert messages[2]["content"] == "a ďalej?"
|