Robert Helewka
63f1a270bb
Adds read-only access to persisted call records for the dashboard and implements a client for the Rhema text-to-speech service. - api/call_history.py: New router providing paged call lists and detailed call records with transcript metadata. - services/tts.py: Async client for OpenAI-compatible TTS endpoints (Rhema/Kokoro) used for call-flow steps.
75 lines
2.6 KiB
Python
75 lines
2.6 KiB
Python
"""Tests for the AI Receptionist decision logic (services/receptionist.py)."""
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from config import ReceptionistSettings, Settings
|
|
from models.routing import (
|
|
RoutingAction,
|
|
RoutingActionType,
|
|
RoutingDecision,
|
|
)
|
|
from services.receptionist import ReceptionistService
|
|
|
|
|
|
def _make_gateway():
|
|
settings = Settings()
|
|
settings.receptionist = ReceptionistSettings()
|
|
return SimpleNamespace(settings=settings, devices={})
|
|
|
|
|
|
class TestReceptionistDecide:
|
|
def test_rule_wins_over_llm_when_rule_is_actionable(self):
|
|
gw = _make_gateway()
|
|
svc = ReceptionistService(gw)
|
|
rule_action = RoutingAction(type=RoutingActionType.REJECT, message="nope")
|
|
decision = RoutingDecision(action=rule_action, reason="rule said so")
|
|
chosen = svc._decide(decision, {"recommended_action": "ring"})
|
|
assert chosen.type == RoutingActionType.REJECT
|
|
|
|
def test_falls_back_to_llm_when_rule_is_default_take_message(self):
|
|
gw = _make_gateway()
|
|
svc = ReceptionistService(gw)
|
|
decision = RoutingDecision(
|
|
action=RoutingAction(type=RoutingActionType.TAKE_MESSAGE),
|
|
reason="default",
|
|
)
|
|
chosen = svc._decide(decision, {"recommended_action": "ring"})
|
|
assert chosen.type == RoutingActionType.RING_CHAIN
|
|
|
|
def test_llm_reject_recommendation_is_honored(self):
|
|
gw = _make_gateway()
|
|
svc = ReceptionistService(gw)
|
|
chosen = svc._decide(None, {"recommended_action": "reject"})
|
|
assert chosen.type == RoutingActionType.REJECT
|
|
assert chosen.message # should carry a polite decline message
|
|
|
|
|
|
class TestReceptionistDeviceList:
|
|
def test_ring_device_returns_explicit_device(self):
|
|
gw = _make_gateway()
|
|
svc = ReceptionistService(gw)
|
|
action = RoutingAction(type=RoutingActionType.RING_DEVICE, device_id="dev_a")
|
|
assert svc._resolve_device_list(action, {}) == ["dev_a"]
|
|
|
|
def test_ring_chain_uses_action_list_when_present(self):
|
|
gw = _make_gateway()
|
|
svc = ReceptionistService(gw)
|
|
action = RoutingAction(
|
|
type=RoutingActionType.RING_CHAIN,
|
|
device_ids=["dev_a", "dev_b"],
|
|
)
|
|
assert svc._resolve_device_list(action, {}) == ["dev_a", "dev_b"]
|
|
|
|
|
|
class TestFrameSilenceHeuristic:
|
|
def test_zeroes_are_silent(self):
|
|
# 16-bit PCM zeros → silent
|
|
assert ReceptionistService._frame_is_silent(b"\x00\x00" * 80) is True
|
|
|
|
def test_loud_pattern_not_silent(self):
|
|
# Loud values → not silent
|
|
loud = b"\xff\x7f" * 80 # max int16 every sample
|
|
assert ReceptionistService._frame_is_silent(loud) is False
|