truetype
69ba824bc8
- Implement LLMMemory module with SQLite + FTS5 - Add conversation tracking with context awareness - Track NPC reputation and interactions - Integrate memory with LLMBrain for contextual replies - Add comprehensive test suite (13 tests) - Update settings with memory_db_path config
310 lines
10 KiB
Python
310 lines
10 KiB
Python
"""Tests for the LLM module (client, brain, social AI).
|
|
|
|
All tests use mocked HTTP responses — no Ollama server required.
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import pytest
|
|
from unittest.mock import AsyncMock, patch, MagicMock
|
|
|
|
from modules.llm.client import (
|
|
LLMClient, LLMConfig, LLMResponse, render_template, PROMPT_TEMPLATES,
|
|
)
|
|
from modules.llm.brain import LLMBrain
|
|
from modules.llm.social_ai import SocialAI
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_mock_response(text="", tokens=10, success=True):
|
|
return LLMResponse(
|
|
text=text, model="test", tokens_used=tokens,
|
|
latency_ms=50.0, success=success,
|
|
)
|
|
|
|
|
|
def _make_json_response(data, tokens=10):
|
|
return LLMResponse(
|
|
text=json.dumps(data), model="test", tokens_used=tokens,
|
|
latency_ms=50.0, success=True,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LLMClient
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestLLMConfig:
|
|
def test_defaults(self):
|
|
cfg = LLMConfig()
|
|
assert cfg.model == "llama3.2"
|
|
assert cfg.temperature == 0.3
|
|
assert cfg.base_url == "http://localhost:11434"
|
|
|
|
|
|
class TestLLMResponse:
|
|
def test_as_json_valid(self):
|
|
r = LLMResponse(text='{"key": "value"}', success=True)
|
|
data = r.as_json()
|
|
assert data == {"key": "value"}
|
|
|
|
def test_as_json_invalid(self):
|
|
r = LLMResponse(text="not json", success=True)
|
|
assert r.as_json() is None
|
|
|
|
def test_as_json_empty(self):
|
|
r = LLMResponse(text="", success=True)
|
|
assert r.as_json() is None
|
|
|
|
|
|
class TestRenderTemplate:
|
|
def test_quest_parse_template(self):
|
|
result = render_template("quest_parse", text="Kill 5 goblins")
|
|
assert "Kill 5 goblins" in result
|
|
assert "JSON" in result
|
|
|
|
def test_social_reply_template(self):
|
|
result = render_template(
|
|
"social_reply",
|
|
message="Hello!",
|
|
character_name="Grok",
|
|
character_class="warrior",
|
|
)
|
|
assert "Grok" in result
|
|
assert "Hello!" in result
|
|
|
|
def test_decide_action_template(self):
|
|
result = render_template(
|
|
"decide_action",
|
|
room_title="Town Square",
|
|
hp=100, max_hp=100, mp=50, max_mp=50,
|
|
exits="north, south",
|
|
npcs="guard",
|
|
items="potion",
|
|
combat_state="idle",
|
|
goal="explore",
|
|
)
|
|
assert "Town Square" in result
|
|
|
|
def test_missing_variable(self):
|
|
# Should not crash, just leave placeholder
|
|
result = render_template("quest_parse")
|
|
assert isinstance(result, str)
|
|
|
|
def test_custom_template(self):
|
|
result = render_template("Hello {name}!", name="World")
|
|
assert result == "Hello World!"
|
|
|
|
|
|
class TestLLMClient:
|
|
def test_client_creation(self):
|
|
client = LLMClient(LLMConfig(model="test-model"))
|
|
assert client.config.model == "test-model"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generate_unavailable(self):
|
|
"""When server is down, should return error response."""
|
|
client = LLMClient(LLMConfig(
|
|
base_url="http://localhost:99999",
|
|
max_retries=0,
|
|
timeout_seconds=1,
|
|
))
|
|
resp = await client.generate("test")
|
|
assert resp.success is False
|
|
assert resp.error is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_is_available_offline(self):
|
|
client = LLMClient(LLMConfig(base_url="http://localhost:99999"))
|
|
available = await client.is_available()
|
|
assert available is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LLMBrain
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestLLMBrain:
|
|
def test_creation(self):
|
|
brain = LLMBrain(LLMConfig())
|
|
assert brain.is_available is False
|
|
assert brain._query_count == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_initialize_offline(self):
|
|
brain = LLMBrain(LLMConfig(base_url="http://localhost:99999"))
|
|
result = await brain.initialize()
|
|
assert result is False
|
|
assert brain.is_available is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_parse_quest_offline_returns_none(self):
|
|
brain = LLMBrain()
|
|
brain._available = False
|
|
result = await brain.parse_quest("Kill the dragon")
|
|
assert result is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_parse_quest_with_mock(self):
|
|
brain = LLMBrain()
|
|
brain._available = True
|
|
mock_resp = _make_json_response([
|
|
{"type": "kill", "target": "dragon", "count": 1, "location": "cave"}
|
|
])
|
|
brain.client.ask = AsyncMock(return_value=mock_resp)
|
|
|
|
result = await brain.parse_quest("Slay the dragon in the cave")
|
|
assert result is not None
|
|
assert len(result) == 1
|
|
assert result[0]["type"] == "kill"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_social_reply_offline_returns_none(self):
|
|
brain = LLMBrain()
|
|
brain._available = False
|
|
result = await brain.social_reply("Hello!")
|
|
assert result is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_social_reply_with_mock(self):
|
|
brain = LLMBrain()
|
|
brain._available = True
|
|
# Mock generate since social_reply now uses it with context
|
|
brain.client.generate = AsyncMock(
|
|
return_value=_make_mock_response("Greetings, fellow adventurer!")
|
|
)
|
|
result = await brain.social_reply("Hello!", sender="TestPlayer")
|
|
assert result is not None
|
|
assert "adventurer" in result.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_suggest_action_offline(self):
|
|
brain = LLMBrain()
|
|
brain._available = False
|
|
from core.game_state import GameState
|
|
gs = GameState()
|
|
result = await brain.suggest_action(gs)
|
|
assert result is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_suggest_action_with_mock(self):
|
|
brain = LLMBrain()
|
|
brain._available = True
|
|
brain.client.ask = AsyncMock(
|
|
return_value=_make_mock_response("north")
|
|
)
|
|
from core.game_state import GameState, Room
|
|
gs = GameState()
|
|
gs.current_room = Room(title="Town Square", exits={"north": 0})
|
|
result = await brain.suggest_action(gs)
|
|
assert result == "north"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_parse_ambiguous_offline(self):
|
|
brain = LLMBrain()
|
|
brain._available = False
|
|
result = await brain.parse_ambiguous("Something happens")
|
|
assert result is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_evaluate_npc_with_mock(self):
|
|
brain = LLMBrain()
|
|
brain._available = True
|
|
brain.client.ask = AsyncMock(
|
|
return_value=_make_json_response(
|
|
{"hostile": True, "confidence": 0.9, "reason": "It's a goblin"}
|
|
)
|
|
)
|
|
result = await brain.evaluate_npc("goblin", "A dark cave")
|
|
assert result is not None
|
|
assert result["hostile"] is True
|
|
|
|
def test_stats(self):
|
|
brain = LLMBrain()
|
|
stats = brain.stats()
|
|
assert stats["queries"] == 0
|
|
assert stats["available"] is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SocialAI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestSocialAI:
|
|
def _make_social_ai(self):
|
|
client = LLMClient(LLMConfig())
|
|
return SocialAI(client, character_name="Grok", character_class="warrior")
|
|
|
|
def test_detect_say(self):
|
|
ai = self._make_social_ai()
|
|
event = ai.detect_social_event("Gandalf says: Hello there!")
|
|
assert event is not None
|
|
assert event["type"] == "say"
|
|
assert event["speaker"] == "Gandalf"
|
|
|
|
def test_detect_tell(self):
|
|
ai = self._make_social_ai()
|
|
event = ai.detect_social_event("Legolas tells you: Follow me")
|
|
assert event is not None
|
|
assert event["type"] == "tell"
|
|
assert event["speaker"] == "Legolas"
|
|
|
|
def test_detect_shout(self):
|
|
ai = self._make_social_ai()
|
|
event = ai.detect_social_event("Gimli shouts: For glory!")
|
|
assert event is not None
|
|
assert event["type"] == "shout"
|
|
|
|
def test_detect_group_invite(self):
|
|
ai = self._make_social_ai()
|
|
event = ai.detect_social_event("Aragorn invites you to join a party")
|
|
assert event is not None
|
|
assert event["type"] == "group_invite"
|
|
|
|
def test_detect_no_social(self):
|
|
ai = self._make_social_ai()
|
|
event = ai.detect_social_event("The wind howls through the trees.")
|
|
assert event is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_respond_say(self):
|
|
ai = self._make_social_ai()
|
|
ai.client.ask = AsyncMock(
|
|
return_value=_make_mock_response("Well met, friend!")
|
|
)
|
|
event = {"type": "say", "speaker": "Gandalf", "message": "Hello"}
|
|
result = await ai.respond(event)
|
|
assert result is not None
|
|
assert result.startswith("say ")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_respond_tell(self):
|
|
ai = self._make_social_ai()
|
|
ai.client.ask = AsyncMock(
|
|
return_value=_make_mock_response("On my way!")
|
|
)
|
|
event = {"type": "tell", "speaker": "Legolas", "message": "Come here"}
|
|
result = await ai.respond(event)
|
|
assert result is not None
|
|
assert result.startswith("tell Legolas")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_respond_group_invite(self):
|
|
ai = self._make_social_ai()
|
|
event = {"type": "group_invite", "speaker": "Aragorn", "message": ""}
|
|
result = await ai.respond(event)
|
|
assert result == "group accept"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generate_emote(self):
|
|
ai = self._make_social_ai()
|
|
ai.client.generate = AsyncMock(
|
|
return_value=_make_mock_response("sharpens his sword quietly")
|
|
)
|
|
result = await ai.generate_emote("resting at the inn")
|
|
assert result is not None
|
|
assert result.startswith("emote ")
|