wehub-resource-sync
141 lines
5.1 KiB
Python
141 lines
5.1 KiB
Python
import warnings
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
from livekit.agents.utils import http_context
|
|
from livekit.plugins.tavus.api import DEFAULT_PAL_ID, TavusAPI
|
|
from livekit.plugins.tavus.avatar import AvatarSession
|
|
|
|
pytestmark = pytest.mark.unit
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _env(monkeypatch):
|
|
for v in ("TAVUS_FACE_ID", "TAVUS_PAL_ID", "TAVUS_REPLICA_ID", "TAVUS_PERSONA_ID"):
|
|
monkeypatch.delenv(v, raising=False)
|
|
monkeypatch.setenv("TAVUS_API_KEY", "test-key")
|
|
|
|
|
|
def _api() -> TavusAPI:
|
|
# session is unused because _post is always mocked in these tests
|
|
return TavusAPI(session=object()) # type: ignore[arg-type]
|
|
|
|
|
|
def _mock_post() -> AsyncMock:
|
|
return AsyncMock(return_value={"conversation_id": "conv1", "persona_id": "pal_auto"})
|
|
|
|
|
|
def _no_deprecation(rec: list[warnings.WarningMessage]) -> bool:
|
|
return not [w for w in rec if issubclass(w.category, DeprecationWarning)]
|
|
|
|
|
|
async def test_new_args_map_to_unchanged_wire_keys():
|
|
api = _api()
|
|
with patch.object(api, "_post", new=_mock_post()) as m:
|
|
with warnings.catch_warnings(record=True) as rec:
|
|
warnings.simplefilter("always")
|
|
cid = await api.create_conversation(face_id="f1", pal_id="p1")
|
|
assert cid == "conv1"
|
|
payload = m.call_args.args[1]
|
|
assert payload["face_id"] == "f1"
|
|
assert payload["pal_id"] == "p1"
|
|
assert _no_deprecation(rec)
|
|
|
|
|
|
async def test_deprecated_args_still_work_and_warn():
|
|
api = _api()
|
|
with patch.object(api, "_post", new=_mock_post()) as m:
|
|
with pytest.warns(DeprecationWarning) as rec:
|
|
await api.create_conversation(replica_id="r1", persona_id="x1")
|
|
payload = m.call_args.args[1]
|
|
assert payload["face_id"] == "r1"
|
|
assert payload["pal_id"] == "x1"
|
|
msgs = [str(w.message) for w in rec]
|
|
assert any("replica_id" in s and "face_id" in s for s in msgs)
|
|
assert any("persona_id" in s and "pal_id" in s for s in msgs)
|
|
|
|
|
|
async def test_no_warning_when_new_and_deprecated_both_given():
|
|
api = _api()
|
|
with patch.object(api, "_post", new=_mock_post()) as m:
|
|
with warnings.catch_warnings(record=True) as rec:
|
|
warnings.simplefilter("always")
|
|
await api.create_conversation(
|
|
face_id="f1", replica_id="r1", pal_id="p1", persona_id="x1"
|
|
)
|
|
# the new values win, so the deprecated aliases are unused -> no warning
|
|
assert _no_deprecation(rec)
|
|
payload = m.call_args.args[1]
|
|
assert payload["face_id"] == "f1"
|
|
assert payload["pal_id"] == "p1"
|
|
|
|
|
|
async def test_new_env_vars_fallback(monkeypatch):
|
|
monkeypatch.setenv("TAVUS_FACE_ID", "envf")
|
|
monkeypatch.setenv("TAVUS_PAL_ID", "envp")
|
|
api = _api()
|
|
with patch.object(api, "_post", new=_mock_post()) as m:
|
|
with warnings.catch_warnings(record=True) as rec:
|
|
warnings.simplefilter("always")
|
|
await api.create_conversation()
|
|
payload = m.call_args.args[1]
|
|
assert payload["face_id"] == "envf"
|
|
assert payload["pal_id"] == "envp"
|
|
assert _no_deprecation(rec)
|
|
|
|
|
|
async def test_deprecated_env_vars_still_work_and_warn(monkeypatch):
|
|
monkeypatch.setenv("TAVUS_REPLICA_ID", "oldf")
|
|
monkeypatch.setenv("TAVUS_PERSONA_ID", "oldp")
|
|
api = _api()
|
|
with patch.object(api, "_post", new=_mock_post()) as m:
|
|
with pytest.warns(DeprecationWarning):
|
|
await api.create_conversation()
|
|
payload = m.call_args.args[1]
|
|
assert payload["face_id"] == "oldf"
|
|
assert payload["pal_id"] == "oldp"
|
|
|
|
|
|
async def test_no_pal_uses_default_pal_with_face_override():
|
|
api = _api()
|
|
with patch.object(api, "_post", new=_mock_post()) as m:
|
|
await api.create_conversation(face_id="f1")
|
|
assert "pals" not in [c.args[0] for c in m.call_args_list] # no pal is created
|
|
payload = m.call_args.args[1]
|
|
assert payload["pal_id"] == DEFAULT_PAL_ID
|
|
assert payload["face_id"] == "f1"
|
|
|
|
|
|
async def test_pal_id_only_skips_pal_creation_and_omits_face():
|
|
api = _api()
|
|
with patch.object(api, "_post", new=_mock_post()) as m:
|
|
await api.create_conversation(pal_id="p1")
|
|
endpoints = [c.args[0] for c in m.call_args_list]
|
|
assert "pals" not in endpoints # an existing pal carries its own default face
|
|
payload = m.call_args.args[1]
|
|
assert payload["pal_id"] == "p1"
|
|
assert "face_id" not in payload
|
|
|
|
|
|
async def test_defaults_to_stock_pal_when_neither_given():
|
|
api = _api()
|
|
with patch.object(api, "_post", new=_mock_post()) as m:
|
|
await api.create_conversation()
|
|
assert "pals" not in [c.args[0] for c in m.call_args_list] # no pal is created
|
|
payload = m.call_args.args[1]
|
|
assert payload["pal_id"] == DEFAULT_PAL_ID
|
|
assert "face_id" not in payload
|
|
|
|
|
|
async def test_avatar_session_resolves_new_and_deprecated_args():
|
|
async with http_context.open():
|
|
with pytest.warns(DeprecationWarning):
|
|
deprecated = AvatarSession(replica_id="r9", persona_id="x9")
|
|
assert deprecated._face_id == "r9"
|
|
assert deprecated._pal_id == "x9"
|
|
|
|
renamed = AvatarSession(face_id="f9", pal_id="p9")
|
|
assert renamed._face_id == "f9"
|
|
assert renamed._pal_id == "p9"
|