46ef9d3aba
Automated salvage commit — agent session ended (exit 124). Work in progress, may need continuation.
121 lines
3.8 KiB
Python
121 lines
3.8 KiB
Python
"""Unit tests for bannerlord.gabs_client."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from bannerlord.gabs_client import GabsClient, GabsError
|
|
|
|
|
|
class TestGabsClientCall:
|
|
"""Tests for GabsClient.call() using mock StreamReader/Writer."""
|
|
|
|
def _make_client(self, response: dict) -> GabsClient:
|
|
"""Return a pre-connected GabsClient with mocked I/O."""
|
|
client = GabsClient(host="localhost", port=4825, timeout=5.0)
|
|
client._connected = True
|
|
|
|
writer = MagicMock()
|
|
writer.write = MagicMock()
|
|
writer.drain = AsyncMock()
|
|
|
|
raw_response = json.dumps(response).encode() + b"\n"
|
|
reader = MagicMock()
|
|
reader.readline = AsyncMock(return_value=raw_response)
|
|
|
|
client._reader = reader
|
|
client._writer = writer
|
|
return client
|
|
|
|
async def test_successful_call_returns_result(self):
|
|
client = self._make_client({"jsonrpc": "2.0", "id": 1, "result": {"status": "ok"}})
|
|
result = await client.call("game/ping")
|
|
assert result == {"status": "ok"}
|
|
|
|
async def test_error_response_raises_gabs_error(self):
|
|
client = self._make_client({
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"error": {"code": -32601, "message": "Method not found"},
|
|
})
|
|
with pytest.raises(GabsError) as exc_info:
|
|
await client.call("unknown/method")
|
|
assert exc_info.value.code == -32601
|
|
|
|
async def test_not_connected_raises_runtime_error(self):
|
|
client = GabsClient()
|
|
with pytest.raises(RuntimeError, match="not connected"):
|
|
await client.call("game/ping")
|
|
|
|
async def test_request_id_increments(self):
|
|
client = self._make_client({"jsonrpc": "2.0", "id": 1, "result": {}})
|
|
await client.call("game/ping")
|
|
# Reset reader for second call
|
|
client._reader.readline = AsyncMock(
|
|
return_value=json.dumps({"jsonrpc": "2.0", "id": 2, "result": {}}).encode() + b"\n"
|
|
)
|
|
await client.call("game/ping")
|
|
assert client._req_id == 2
|
|
|
|
async def test_get_game_state_returns_empty_on_error(self):
|
|
client = GabsClient()
|
|
client._connected = True
|
|
|
|
writer = MagicMock()
|
|
writer.write = MagicMock()
|
|
writer.drain = AsyncMock()
|
|
reader = MagicMock()
|
|
reader.readline = AsyncMock(side_effect=OSError("connection reset"))
|
|
|
|
client._reader = reader
|
|
client._writer = writer
|
|
|
|
result = await client.get_game_state()
|
|
assert result == {}
|
|
|
|
async def test_ping_returns_true_on_success(self):
|
|
client = self._make_client({"jsonrpc": "2.0", "id": 1, "result": "pong"})
|
|
result = await client.ping()
|
|
assert result is True
|
|
|
|
async def test_ping_returns_false_on_failure(self):
|
|
client = GabsClient()
|
|
result = await client.ping()
|
|
assert result is False
|
|
|
|
|
|
class TestGabsClientLifecycle:
|
|
async def test_connect_failure_sets_not_connected(self):
|
|
client = GabsClient(host="localhost", port=9999, timeout=0.1)
|
|
with pytest.raises(Exception):
|
|
await client.connect()
|
|
assert not client.is_connected
|
|
|
|
async def test_context_manager_calls_connect_and_disconnect(self):
|
|
client = GabsClient()
|
|
connect_called = False
|
|
disconnect_called = False
|
|
|
|
async def _fake_connect():
|
|
nonlocal connect_called
|
|
connect_called = True
|
|
client._connected = True
|
|
|
|
async def _fake_disconnect():
|
|
nonlocal disconnect_called
|
|
disconnect_called = True
|
|
client._connected = False
|
|
|
|
client.connect = _fake_connect
|
|
client.disconnect = _fake_disconnect
|
|
|
|
async with client as c:
|
|
assert c is client
|
|
assert connect_called
|
|
|
|
assert disconnect_called
|