wehub-resource-sync
34 lines
1.1 KiB
Python
34 lines
1.1 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
from typing import cast
|
|
|
|
import pytest
|
|
|
|
from livekit.agents import llm
|
|
from livekit.agents.llm.remote_chat_context import RemoteChatContext
|
|
from livekit.plugins.openai.realtime.realtime_model import RealtimeSession
|
|
|
|
pytestmark = pytest.mark.unit
|
|
|
|
|
|
def test_update_chat_ctx_deletes_empty_remote_items() -> None:
|
|
remote_ctx = RemoteChatContext()
|
|
audio_item = llm.ChatMessage(id="audio_item", role="user", content=[])
|
|
kept_item = llm.ChatMessage(id="assistant_item", role="assistant", content=["kept"])
|
|
remote_ctx.insert(None, audio_item)
|
|
remote_ctx.insert(audio_item.id, kept_item)
|
|
|
|
session = cast(RealtimeSession, SimpleNamespace(_remote_chat_ctx=remote_ctx))
|
|
events = RealtimeSession._create_update_chat_ctx_events(
|
|
session,
|
|
llm.ChatContext(items=[kept_item]),
|
|
)
|
|
|
|
delete_ids = [
|
|
getattr(event, "item_id", None)
|
|
for event in events
|
|
if getattr(event, "type", None) == "conversation.item.delete"
|
|
]
|
|
assert delete_ids == ["audio_item"]
|