75 lines
2.4 KiB
Python
75 lines
2.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Tests for the OpenClaw WebSocket client session helpers."""
|
|
|
|
import pytest
|
|
|
|
from shared.client.openclaw_websocket_client import OpenClawWebSocketClient
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_session_parses_gateway_key_response():
|
|
client = OpenClawWebSocketClient(gateway_token="test-token")
|
|
|
|
async def fake_send_request(method, params=None, _allow_handshake=False):
|
|
assert method == "sessions.resolve"
|
|
assert params["agentId"] == "main"
|
|
return {"ok": True, "key": "agent:main:main"}
|
|
|
|
client._send_request = fake_send_request # type: ignore[method-assign]
|
|
|
|
resolved = await client.resolve_session(agent_id="main")
|
|
|
|
assert resolved == "agent:main:main"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_message_uses_session_send_payload():
|
|
client = OpenClawWebSocketClient(gateway_token="test-token")
|
|
|
|
async def fake_send_request(method, params=None, _allow_handshake=False):
|
|
assert method == "sessions.send"
|
|
assert params == {
|
|
"key": "agent:main:main",
|
|
"message": "hello",
|
|
"thinking": "medium",
|
|
}
|
|
return {"ok": True, "runId": "run-1"}
|
|
|
|
client._send_request = fake_send_request # type: ignore[method-assign]
|
|
|
|
result = await client.send_message("agent:main:main", "hello", thinking="medium")
|
|
|
|
assert result["runId"] == "run-1"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_session_history_uses_sessions_preview():
|
|
client = OpenClawWebSocketClient(gateway_token="test-token")
|
|
|
|
async def fake_send_request(method, params=None, _allow_handshake=False):
|
|
assert method == "sessions.preview"
|
|
assert params == {"keys": ["agent:main:main"], "limit": 12}
|
|
return {"previews": []}
|
|
|
|
client._send_request = fake_send_request # type: ignore[method-assign]
|
|
|
|
result = await client.get_session_history("agent:main:main", limit=12)
|
|
|
|
assert result == {"previews": []}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unsubscribe_uses_session_messages_unsubscribe():
|
|
client = OpenClawWebSocketClient(gateway_token="test-token")
|
|
|
|
async def fake_send_request(method, params=None, _allow_handshake=False):
|
|
assert method == "sessions.messages.unsubscribe"
|
|
assert params == {"key": "agent:main:main"}
|
|
return {"subscribed": False}
|
|
|
|
client._send_request = fake_send_request # type: ignore[method-assign]
|
|
|
|
result = await client.unsubscribe("agent:main:main")
|
|
|
|
assert result == {"subscribed": False}
|