151 lines
4.1 KiB
Python
151 lines
4.1 KiB
Python
import aiohttp
|
|
import pytest
|
|
|
|
from app.backend_adapters import (
|
|
HistoryDisabledBackendAdapter,
|
|
HttpBackendAdapter,
|
|
NullBackendAdapter,
|
|
build_backend_adapter,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_backend_adapter_without_url_returns_null_adapter():
|
|
adapter = build_backend_adapter(
|
|
backend_url=None,
|
|
backend_mode="auto",
|
|
history_enabled=True,
|
|
timeout_sec=3,
|
|
)
|
|
assert isinstance(adapter, NullBackendAdapter)
|
|
|
|
assert await adapter.fetch_assistant_config("assistant_1") is None
|
|
assert (
|
|
await adapter.create_call_record(
|
|
user_id=1,
|
|
assistant_id="assistant_1",
|
|
source="debug",
|
|
)
|
|
is None
|
|
)
|
|
assert (
|
|
await adapter.add_transcript(
|
|
call_id="call_1",
|
|
turn_index=0,
|
|
speaker="human",
|
|
content="hi",
|
|
start_ms=0,
|
|
end_ms=100,
|
|
confidence=0.9,
|
|
duration_ms=100,
|
|
)
|
|
is False
|
|
)
|
|
assert (
|
|
await adapter.finalize_call_record(
|
|
call_id="call_1",
|
|
status="connected",
|
|
duration_seconds=2,
|
|
)
|
|
is False
|
|
)
|
|
assert await adapter.search_knowledge_context(kb_id="kb_1", query="hello", n_results=3) == []
|
|
assert await adapter.fetch_tool_resource("tool_1") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_http_backend_adapter_create_call_record_posts_expected_payload(monkeypatch):
|
|
captured = {}
|
|
|
|
class _FakeResponse:
|
|
def __init__(self, status=200, payload=None):
|
|
self.status = status
|
|
self._payload = payload if payload is not None else {}
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, tb):
|
|
return None
|
|
|
|
async def json(self):
|
|
return self._payload
|
|
|
|
def raise_for_status(self):
|
|
if self.status >= 400:
|
|
raise RuntimeError("http_error")
|
|
|
|
class _FakeClientSession:
|
|
def __init__(self, timeout=None):
|
|
captured["timeout"] = timeout
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, tb):
|
|
return None
|
|
|
|
def post(self, url, json=None):
|
|
captured["url"] = url
|
|
captured["json"] = json
|
|
return _FakeResponse(status=200, payload={"id": "call_123"})
|
|
|
|
monkeypatch.setattr("app.backend_adapters.aiohttp.ClientSession", _FakeClientSession)
|
|
|
|
adapter = build_backend_adapter(
|
|
backend_url="http://localhost:8100",
|
|
backend_mode="auto",
|
|
history_enabled=True,
|
|
timeout_sec=7,
|
|
)
|
|
assert isinstance(adapter, HttpBackendAdapter)
|
|
|
|
call_id = await adapter.create_call_record(
|
|
user_id=99,
|
|
assistant_id="assistant_9",
|
|
source="debug",
|
|
)
|
|
|
|
assert call_id == "call_123"
|
|
assert captured["url"] == "http://localhost:8100/api/history"
|
|
assert captured["json"] == {
|
|
"user_id": 99,
|
|
"assistant_id": "assistant_9",
|
|
"source": "debug",
|
|
"status": "connected",
|
|
}
|
|
assert isinstance(captured["timeout"], aiohttp.ClientTimeout)
|
|
assert captured["timeout"].total == 7
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backend_mode_disabled_forces_null_even_with_url():
|
|
adapter = build_backend_adapter(
|
|
backend_url="http://localhost:8100",
|
|
backend_mode="disabled",
|
|
history_enabled=True,
|
|
timeout_sec=7,
|
|
)
|
|
assert isinstance(adapter, NullBackendAdapter)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_history_disabled_wraps_backend_adapter():
|
|
adapter = build_backend_adapter(
|
|
backend_url="http://localhost:8100",
|
|
backend_mode="auto",
|
|
history_enabled=False,
|
|
timeout_sec=7,
|
|
)
|
|
assert isinstance(adapter, HistoryDisabledBackendAdapter)
|
|
assert await adapter.create_call_record(user_id=1, assistant_id="a1", source="debug") is None
|
|
assert await adapter.add_transcript(
|
|
call_id="c1",
|
|
turn_index=0,
|
|
speaker="human",
|
|
content="hi",
|
|
start_ms=0,
|
|
end_ms=10,
|
|
duration_ms=10,
|
|
) is False
|