truetype
6b048dc0cc
test / test (push) Has been cancelled
Fase 2 del remediation plan: - set_puzzle_state / is_puzzle_solved per enigmi area - update_inventory su character_items con add/remove/equip/list - combat() legge HP da combat_state e persiste danno - dm_cycle inietta campaign_id nei tool che ne hanno bisogno - helpers db: dedup inventario, remove/equip per nome Fase 4 del remediation plan: - test integrazione dm_cycle con LLM mockato (5 test) - test WebSocket handler con mock (4 test) - CI minimo .gitea/workflows/test.yml
130 lines
3.3 KiB
Python
130 lines
3.3 KiB
Python
"""WebSocket test: test handler logica via mock WS."""
|
|
import json
|
|
import sys
|
|
sys.path.insert(0, "src")
|
|
|
|
from unittest.mock import patch, AsyncMock
|
|
|
|
import pytest
|
|
|
|
from src.engine import db
|
|
|
|
MOCK_RESP = json.dumps({
|
|
"tool": "roll_dice",
|
|
"params": {"die": "d20", "count": 1},
|
|
"narration": "Tendi l'orecchio nel buio...",
|
|
})
|
|
|
|
|
|
def _setup_campaign():
|
|
conn = db.init_db()
|
|
camp_id = db.create_campaign("Test WS", "Avventura")
|
|
char_id = db.create_character(camp_id, "Eroe", "guerriero", "soldato")
|
|
db.update_character(
|
|
char_id,
|
|
level=1, str=15, agi=11, sta=14,
|
|
per=9, int=8, lck=10,
|
|
hp_current=12, hp_max=12, ac=15,
|
|
armor_type="light",
|
|
)
|
|
db.seed_adventure_state(camp_id)
|
|
return camp_id
|
|
|
|
|
|
def _make_mock_ws():
|
|
ws = AsyncMock()
|
|
ws.send_json = AsyncMock()
|
|
ws.send_text = AsyncMock()
|
|
ws.close = AsyncMock()
|
|
ws.receive_text = AsyncMock()
|
|
return ws
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("src.engine.dm.llm_ask", return_value=MOCK_RESP)
|
|
async def test_ws_handler_game_loop(mock_ask):
|
|
from src.web.server import game_ws
|
|
|
|
camp_id = _setup_campaign()
|
|
ws = _make_mock_ws()
|
|
ws.receive_text.side_effect = [
|
|
json.dumps({"type": "action", "payload": "Provo ad ascoltare"}),
|
|
Exception("WS chiuso"),
|
|
]
|
|
|
|
try:
|
|
await game_ws(ws, camp_id)
|
|
except Exception:
|
|
pass
|
|
|
|
ws.accept.assert_awaited_once()
|
|
|
|
calls = [c[0][0] for c in ws.send_json.call_args_list]
|
|
assert calls[0]["type"] == "state"
|
|
resp = calls[1]
|
|
assert resp["type"] == "response"
|
|
assert resp["tool"] == "roll_dice"
|
|
assert resp["turn"] == 1
|
|
assert calls[2]["type"] == "state"
|
|
assert calls[2]["campaign"]["turn"] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("src.engine.dm.llm_ask", return_value=MOCK_RESP)
|
|
async def test_ws_handler_unknown_campaign(mock_ask):
|
|
from src.web.server import game_ws
|
|
|
|
ws = _make_mock_ws()
|
|
await game_ws(ws, 99999)
|
|
|
|
ws.accept.assert_awaited_once()
|
|
ws.send_json.assert_awaited_once_with(
|
|
{"type": "error", "message": "Campagna non trovata"}
|
|
)
|
|
ws.close.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("src.engine.dm.llm_ask", return_value=MOCK_RESP)
|
|
async def test_ws_handler_plain_text(mock_ask):
|
|
from src.web.server import game_ws
|
|
|
|
camp_id = _setup_campaign()
|
|
ws = _make_mock_ws()
|
|
ws.receive_text.side_effect = [
|
|
"Attacco il guardiano!",
|
|
Exception("fine"),
|
|
]
|
|
|
|
try:
|
|
await game_ws(ws, camp_id)
|
|
except Exception:
|
|
pass
|
|
|
|
calls = [c[0][0] for c in ws.send_json.call_args_list]
|
|
assert calls[0]["type"] == "state"
|
|
assert calls[1]["type"] == "response"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("src.engine.dm.llm_ask", return_value=MOCK_RESP)
|
|
async def test_ws_handler_error_boundary(mock_ask):
|
|
from src.web.server import game_ws
|
|
|
|
camp_id = _setup_campaign()
|
|
ws = _make_mock_ws()
|
|
ws.receive_text.side_effect = [
|
|
json.dumps({"type": "action", "payload": "test"}),
|
|
Exception("fine"),
|
|
]
|
|
|
|
with patch("src.web.server.dm_cycle", side_effect=[ValueError("Test errore")]):
|
|
try:
|
|
await game_ws(ws, camp_id)
|
|
except Exception:
|
|
pass
|
|
|
|
calls = [c[0][0] for c in ws.send_json.call_args_list]
|
|
error_msgs = [c for c in calls if c.get("type") == "error"]
|
|
assert len(error_msgs) >= 1
|