118 lines
3.7 KiB
Python
118 lines
3.7 KiB
Python
import pytest
|
|
|
|
from core.protocol import SettingsAction
|
|
from sdk.interface import MessageChunk, MessageResponse, UserSettings
|
|
from sdk.prototype_state import PrototypeStateStore
|
|
from sdk.real import RealPlatformClient
|
|
|
|
|
|
class FakeAgentSessionClient:
|
|
def __init__(self) -> None:
|
|
self.send_calls: list[tuple[str, str]] = []
|
|
self.stream_calls: list[tuple[str, str]] = []
|
|
|
|
async def send_message(self, *, thread_key: str, text: str) -> MessageResponse:
|
|
self.send_calls.append((thread_key, text))
|
|
return MessageResponse(
|
|
message_id=thread_key,
|
|
response=f"echo:{text}",
|
|
tokens_used=3,
|
|
finished=True,
|
|
)
|
|
|
|
async def stream_message(self, *, thread_key: str, text: str):
|
|
self.stream_calls.append((thread_key, text))
|
|
yield MessageChunk(message_id=thread_key, delta=text[:2], finished=False)
|
|
yield MessageChunk(message_id=thread_key, delta=text[2:], finished=True, tokens_used=3)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_real_platform_client_get_or_create_user_uses_local_state():
|
|
client = RealPlatformClient(
|
|
agent_sessions=FakeAgentSessionClient(),
|
|
prototype_state=PrototypeStateStore(),
|
|
platform="telegram",
|
|
)
|
|
|
|
first = await client.get_or_create_user("u1", "telegram", "Alice")
|
|
second = await client.get_or_create_user("u1", "telegram")
|
|
|
|
assert first.user_id == "usr-telegram-u1"
|
|
assert first.is_new is True
|
|
assert second.user_id == first.user_id
|
|
assert second.is_new is False
|
|
assert second.display_name == "Alice"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_real_platform_client_send_message_uses_configured_platform():
|
|
agent_sessions = FakeAgentSessionClient()
|
|
client = RealPlatformClient(
|
|
agent_sessions=agent_sessions,
|
|
prototype_state=PrototypeStateStore(),
|
|
platform="telegram",
|
|
)
|
|
|
|
result = await client.send_message("usr-telegram-u1", "C1", "hello")
|
|
|
|
assert result == MessageResponse(
|
|
message_id="8:telegram15:usr-telegram-u12:C1",
|
|
response="echo:hello",
|
|
tokens_used=3,
|
|
finished=True,
|
|
)
|
|
assert agent_sessions.send_calls == [
|
|
("8:telegram15:usr-telegram-u12:C1", "hello")
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_real_platform_client_stream_message_uses_configured_platform():
|
|
agent_sessions = FakeAgentSessionClient()
|
|
client = RealPlatformClient(
|
|
agent_sessions=agent_sessions,
|
|
prototype_state=PrototypeStateStore(),
|
|
platform="telegram",
|
|
)
|
|
|
|
chunks = []
|
|
async for chunk in client.stream_message("usr-telegram-u1", "C1", "hello"):
|
|
chunks.append(chunk)
|
|
|
|
assert chunks == [
|
|
MessageChunk(
|
|
message_id="8:telegram15:usr-telegram-u12:C1",
|
|
delta="he",
|
|
finished=False,
|
|
tokens_used=0,
|
|
),
|
|
MessageChunk(
|
|
message_id="8:telegram15:usr-telegram-u12:C1",
|
|
delta="llo",
|
|
finished=True,
|
|
tokens_used=3,
|
|
),
|
|
]
|
|
assert agent_sessions.stream_calls == [
|
|
("8:telegram15:usr-telegram-u12:C1", "hello")
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_real_platform_client_settings_are_local():
|
|
client = RealPlatformClient(
|
|
agent_sessions=FakeAgentSessionClient(),
|
|
prototype_state=PrototypeStateStore(),
|
|
platform="matrix",
|
|
)
|
|
|
|
await client.update_settings(
|
|
"usr-matrix-u1",
|
|
SettingsAction(action="toggle_skill", payload={"skill": "browser", "enabled": True}),
|
|
)
|
|
|
|
settings = await client.get_settings("usr-matrix-u1")
|
|
|
|
assert isinstance(settings, UserSettings)
|
|
assert settings.skills["browser"] is True
|
|
assert settings.skills["web-search"] is True
|