304 lines
11 KiB
Python
304 lines
11 KiB
Python
"""Tests for nexus.evennia_mempalace commands and NPC helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from nexus.evennia_mempalace.commands.recall import CmdRecall, CmdEnterRoom, CmdAsk, _closest_room
|
|
from nexus.evennia_mempalace.commands.write import CmdRecord, CmdNote, CmdEvent
|
|
from nexus.evennia_mempalace.typeclasses.npcs import StewardNPC, _extract_topic
|
|
from nexus.mempalace.searcher import MemPalaceResult, MemPalaceUnavailable
|
|
|
|
|
|
# ── Helpers ──────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _make_caller(wing: str = "bezalel"):
|
|
"""Build a minimal mock Evennia caller."""
|
|
caller = MagicMock()
|
|
caller.db = MagicMock()
|
|
caller.db.wing = wing
|
|
caller.account = MagicMock()
|
|
return caller
|
|
|
|
|
|
def _make_cmd(cls, args: str = "", switches: list | None = None, wing: str = "bezalel"):
|
|
"""Instantiate an Evennia command mock and wire it up."""
|
|
cmd = cls()
|
|
cmd.caller = _make_caller(wing)
|
|
cmd.args = args
|
|
cmd.switches = switches or []
|
|
return cmd
|
|
|
|
|
|
# ── CmdRecall ────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_recall_no_args_shows_usage():
|
|
cmd = _make_cmd(CmdRecall, args="")
|
|
cmd.func()
|
|
cmd.caller.msg.assert_called_once()
|
|
assert "Usage" in cmd.caller.msg.call_args[0][0]
|
|
|
|
|
|
def test_recall_calls_search_memories():
|
|
results = [
|
|
MemPalaceResult(text="CI pipeline failed", room="forge", wing="bezalel", score=0.9)
|
|
]
|
|
with patch("nexus.evennia_mempalace.commands.recall.search_memories", return_value=results):
|
|
cmd = _make_cmd(CmdRecall, args="CI failures")
|
|
cmd.func()
|
|
|
|
calls = [c[0][0] for c in cmd.caller.msg.call_args_list]
|
|
assert any("CI pipeline failed" in c for c in calls)
|
|
|
|
|
|
def test_recall_fleet_flag_calls_search_fleet():
|
|
results = [
|
|
MemPalaceResult(text="Fleet doc", room="nexus", wing="timmy", score=0.8)
|
|
]
|
|
with patch("nexus.evennia_mempalace.commands.recall.search_fleet", return_value=results) as mock_fleet:
|
|
cmd = _make_cmd(CmdRecall, args="architecture --fleet", switches=["--fleet"])
|
|
cmd.func()
|
|
|
|
mock_fleet.assert_called_once()
|
|
query_arg = mock_fleet.call_args[0][0]
|
|
assert "--fleet" not in query_arg
|
|
assert "architecture" in query_arg
|
|
|
|
|
|
def test_recall_unavailable_shows_error():
|
|
with patch(
|
|
"nexus.evennia_mempalace.commands.recall.search_memories",
|
|
side_effect=MemPalaceUnavailable("ChromaDB not installed"),
|
|
):
|
|
cmd = _make_cmd(CmdRecall, args="anything")
|
|
cmd.func()
|
|
|
|
msg = cmd.caller.msg.call_args[0][0]
|
|
assert "unavailable" in msg.lower()
|
|
|
|
|
|
def test_recall_no_results_shows_no_memories():
|
|
with patch("nexus.evennia_mempalace.commands.recall.search_memories", return_value=[]):
|
|
cmd = _make_cmd(CmdRecall, args="obscure query")
|
|
cmd.func()
|
|
|
|
calls = " ".join(c[0][0] for c in cmd.caller.msg.call_args_list)
|
|
assert "No memories" in calls
|
|
|
|
|
|
# ── _closest_room ─────────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.parametrize("topic,expected", [
|
|
("forge", "forge"),
|
|
("CI pipeline", "forge"),
|
|
("hermes agent", "hermes"),
|
|
("nexus report", "nexus"),
|
|
("issue triage", "issues"),
|
|
("spike experiment", "experiments"),
|
|
("totally unknown topic xyz", "general"),
|
|
])
|
|
def test_closest_room(topic, expected):
|
|
assert _closest_room(topic) == expected
|
|
|
|
|
|
# ── CmdEnterRoom ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_enter_room_no_args_shows_usage():
|
|
cmd = _make_cmd(CmdEnterRoom, args="")
|
|
cmd.func()
|
|
output = " ".join(c[0][0] for c in cmd.caller.msg.call_args_list)
|
|
assert "Usage" in output
|
|
assert "forge" in output # shows core rooms
|
|
|
|
|
|
def test_enter_room_exact_match_no_room_found():
|
|
"""When an exact room name is given but no room exists, show a help message."""
|
|
# evennia.utils.search raises Django config errors outside a live server;
|
|
# CmdEnterRoom catches all exceptions and falls back to a help message.
|
|
cmd = _make_cmd(CmdEnterRoom, args="forge")
|
|
cmd.func()
|
|
assert cmd.caller.msg.called
|
|
output = " ".join(c[0][0] for c in cmd.caller.msg.call_args_list)
|
|
# Should mention the room name or MemPalaceRoom typeclass
|
|
assert "forge" in output or "MemPalaceRoom" in output or "No palace room" in output
|
|
|
|
|
|
# ── Write commands ───────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_record_no_args_shows_usage():
|
|
cmd = _make_cmd(CmdRecord, args="")
|
|
cmd.func()
|
|
assert "Usage" in cmd.caller.msg.call_args[0][0]
|
|
|
|
|
|
def test_record_calls_add_memory():
|
|
with patch("nexus.evennia_mempalace.commands.write.add_memory", return_value="fake-uuid-1234-5678-abcd") as mock_add:
|
|
cmd = _make_cmd(CmdRecord, args="Use ChromaDB for storage.")
|
|
cmd.func()
|
|
|
|
mock_add.assert_called_once()
|
|
kwargs = mock_add.call_args[1]
|
|
assert kwargs["room"] == "hall_facts"
|
|
assert "ChromaDB" in mock_add.call_args[0][0]
|
|
|
|
|
|
def test_note_files_to_hall_discoveries():
|
|
with patch("nexus.evennia_mempalace.commands.write.add_memory", return_value="uuid") as mock_add:
|
|
cmd = _make_cmd(CmdNote, args="AAAK reduces cost by 40%.")
|
|
cmd.func()
|
|
|
|
assert mock_add.call_args[1]["room"] == "hall_discoveries"
|
|
|
|
|
|
def test_event_files_to_hall_events():
|
|
with patch("nexus.evennia_mempalace.commands.write.add_memory", return_value="uuid") as mock_add:
|
|
cmd = _make_cmd(CmdEvent, args="Deployed Evennia bridge to Alpha.")
|
|
cmd.func()
|
|
|
|
assert mock_add.call_args[1]["room"] == "hall_events"
|
|
|
|
|
|
def test_write_command_unavailable_shows_error():
|
|
with patch(
|
|
"nexus.evennia_mempalace.commands.write.add_memory",
|
|
side_effect=MemPalaceUnavailable("no palace"),
|
|
):
|
|
cmd = _make_cmd(CmdRecord, args="some text")
|
|
cmd.func()
|
|
|
|
msg = cmd.caller.msg.call_args[0][0]
|
|
assert "unavailable" in msg.lower()
|
|
|
|
|
|
# ── _extract_topic ────────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.parametrize("question,expected_substring", [
|
|
("about nightly watch failures", "nightly watch failures"),
|
|
("what do you know about CI pipeline?", "CI pipeline"),
|
|
("tell me about hermes", "hermes"),
|
|
("regarding the forge build", "forge build"),
|
|
("nightly watch failures", "nightly watch failures"),
|
|
])
|
|
def test_extract_topic(question, expected_substring):
|
|
result = _extract_topic(question)
|
|
assert expected_substring.lower() in result.lower()
|
|
|
|
|
|
# ── StewardNPC.respond_to_question ───────────────────────────────────────────
|
|
|
|
|
|
def test_steward_responds_with_results():
|
|
npc = StewardNPC()
|
|
npc.db = MagicMock()
|
|
npc.db.steward_wing = "bezalel"
|
|
npc.db.steward_name = "Bezalel-Steward"
|
|
npc.db.steward_n_results = 3
|
|
npc.key = "steward"
|
|
|
|
results = [
|
|
MemPalaceResult(text="Three failures last week.", room="forge", wing="bezalel", score=0.95)
|
|
]
|
|
with patch("nexus.evennia_mempalace.typeclasses.npcs.search_memories", return_value=results):
|
|
response = npc.respond_to_question("about nightly watch failures")
|
|
|
|
assert "Bezalel-Steward" in response
|
|
assert "Three failures" in response
|
|
|
|
|
|
def test_steward_responds_not_found():
|
|
npc = StewardNPC()
|
|
npc.db = MagicMock()
|
|
npc.db.steward_wing = "bezalel"
|
|
npc.db.steward_name = "Steward"
|
|
npc.db.steward_n_results = 3
|
|
npc.key = "steward"
|
|
|
|
with patch("nexus.evennia_mempalace.typeclasses.npcs.search_memories", return_value=[]):
|
|
response = npc.respond_to_question("about unknown_topic_xyz")
|
|
|
|
assert "nothing" in response.lower() or "found" in response.lower()
|
|
|
|
|
|
def test_steward_responds_unavailable():
|
|
npc = StewardNPC()
|
|
npc.db = MagicMock()
|
|
npc.db.steward_wing = "bezalel"
|
|
npc.db.steward_name = "Steward"
|
|
npc.db.steward_n_results = 3
|
|
npc.key = "steward"
|
|
|
|
with patch(
|
|
"nexus.evennia_mempalace.typeclasses.npcs.search_memories",
|
|
side_effect=MemPalaceUnavailable("no palace"),
|
|
):
|
|
response = npc.respond_to_question("about anything")
|
|
|
|
assert "unreachable" in response.lower()
|
|
|
|
|
|
# ── CmdAsk ────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_ask_no_about_shows_usage():
|
|
cmd = _make_cmd(CmdAsk, args="steward")
|
|
cmd.func()
|
|
assert "Usage" in cmd.caller.msg.call_args[0][0]
|
|
|
|
|
|
def test_ask_routes_to_steward_npc():
|
|
npc = StewardNPC()
|
|
npc.db = MagicMock()
|
|
npc.db.steward_wing = "bezalel"
|
|
npc.db.steward_name = "Steward"
|
|
npc.db.steward_n_results = 3
|
|
npc.key = "steward"
|
|
|
|
results = [MemPalaceResult(text="CI failed twice.", room="forge", wing="bezalel", score=0.9)]
|
|
with patch(
|
|
"nexus.evennia_mempalace.typeclasses.npcs.search_memories", return_value=results
|
|
):
|
|
with patch(
|
|
"nexus.evennia_mempalace.commands.recall.search_object",
|
|
return_value=[npc],
|
|
create=True,
|
|
):
|
|
cmd = _make_cmd(CmdAsk, args="steward about runner outages")
|
|
# Patch search_object inside the command module
|
|
import nexus.evennia_mempalace.commands.recall as recall_mod
|
|
orig = getattr(recall_mod, "_search_object_for_ask", None)
|
|
cmd.func()
|
|
|
|
assert cmd.caller.msg.called
|
|
output = " ".join(c[0][0] for c in cmd.caller.msg.call_args_list)
|
|
# Either found a steward result or showed no-steward-found message
|
|
assert "CI failed" in output or "steward" in output.lower()
|
|
|
|
|
|
def test_ask_no_npc_found_shows_help():
|
|
cmd = _make_cmd(CmdAsk, args="nobody about anything")
|
|
cmd.caller.location = MagicMock()
|
|
cmd.caller.location.contents = []
|
|
cmd.func()
|
|
output = cmd.caller.msg.call_args[0][0]
|
|
assert "nobody" in output.lower() or "steward" in output.lower() or "No steward" in output
|
|
|
|
|
|
# ── added_by metadata ─────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_record_includes_added_by_evennia():
|
|
with patch("nexus.evennia_mempalace.commands.write.add_memory", return_value="uuid") as mock_add:
|
|
cmd = _make_cmd(CmdRecord, args="Deploy decision recorded.")
|
|
cmd.func()
|
|
|
|
extra = mock_add.call_args[1].get("extra_metadata", {})
|
|
assert extra.get("added_by") == "evennia"
|