Jared Miller
ee0dc839d8
The server never proactively offered GMCP or MSDP to clients, so telnetlib3 logged "cannot send MSDP without negotiation" every second. Now the server sends WILL GMCP and WILL MSDP on connection, and send_msdp_vitals checks negotiation state before attempting to send.
369 lines
11 KiB
Python
369 lines
11 KiB
Python
"""Tests for embedded z-machine MUD integration.
|
|
|
|
Tests the MUD UI components and EmbeddedIFSession integration with real zork1.z3.
|
|
"""
|
|
|
|
import threading
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from mudlib.zmachine.mud_ui import MudFilesystem, MudInputStream, MudScreen
|
|
|
|
ZORK_PATH = Path(__file__).parent.parent / "content" / "stories" / "zork1.z3"
|
|
requires_zork = pytest.mark.skipif(not ZORK_PATH.exists(), reason="zork1.z3 not found")
|
|
|
|
|
|
@dataclass
|
|
class MockWriter:
|
|
def __post_init__(self):
|
|
# Mock option negotiation state
|
|
from unittest.mock import MagicMock
|
|
|
|
self.local_option = MagicMock()
|
|
self.local_option.enabled = MagicMock(return_value=False)
|
|
self.remote_option = MagicMock()
|
|
self.remote_option.enabled = MagicMock(return_value=False)
|
|
|
|
def write(self, data):
|
|
pass
|
|
|
|
async def drain(self):
|
|
pass
|
|
|
|
|
|
# Unit tests for MUD UI components
|
|
|
|
|
|
def test_mud_screen_captures_output():
|
|
"""MudScreen captures written text and flush returns it."""
|
|
screen = MudScreen()
|
|
screen.write("Hello ")
|
|
screen.write("world!")
|
|
output = screen.flush()
|
|
assert output == "Hello world!"
|
|
|
|
|
|
def test_mud_screen_flush_clears_buffer():
|
|
"""MudScreen flush clears buffer, second flush returns empty."""
|
|
screen = MudScreen()
|
|
screen.write("test")
|
|
first = screen.flush()
|
|
assert first == "test"
|
|
|
|
second = screen.flush()
|
|
assert second == ""
|
|
|
|
|
|
def test_mud_screen_suppresses_upper_window_writes():
|
|
"""MudScreen discards writes to upper window (status line)."""
|
|
screen = MudScreen()
|
|
screen.write("before ")
|
|
screen.select_window(1) # upper window
|
|
screen.write("STATUS LINE")
|
|
screen.select_window(0) # back to lower window
|
|
screen.write("after")
|
|
output = screen.flush()
|
|
assert output == "before after"
|
|
|
|
|
|
def test_mud_screen_lower_window_is_default():
|
|
"""MudScreen starts with lower window active, writes are captured."""
|
|
screen = MudScreen()
|
|
screen.write("hello")
|
|
assert screen.flush() == "hello"
|
|
|
|
|
|
def test_mud_screen_upper_window_multiple_writes():
|
|
"""MudScreen discards all writes while upper window is active."""
|
|
screen = MudScreen()
|
|
screen.select_window(1)
|
|
screen.write("Room Name")
|
|
screen.write(" | Score: 0")
|
|
screen.select_window(0)
|
|
screen.write("You are in a dark room.\n")
|
|
output = screen.flush()
|
|
assert output == "You are in a dark room.\n"
|
|
|
|
|
|
def test_mud_input_stream_feed_and_read():
|
|
"""MudInputStream feed and read_line work with threading."""
|
|
stream = MudInputStream()
|
|
result = []
|
|
|
|
def reader():
|
|
result.append(stream.read_line())
|
|
|
|
t = threading.Thread(target=reader)
|
|
t.start()
|
|
# Wait for stream to signal it's waiting
|
|
stream._waiting.wait(timeout=2)
|
|
stream.feed("hello")
|
|
t.join(timeout=2)
|
|
assert result == ["hello"]
|
|
|
|
|
|
def test_mud_filesystem_save_restore(tmp_path):
|
|
"""MudFilesystem save and restore bytes correctly."""
|
|
save_path = tmp_path / "test.qzl"
|
|
filesystem = MudFilesystem(save_path)
|
|
|
|
test_data = b"\x01\x02\x03\x04\x05"
|
|
success = filesystem.save_game(test_data)
|
|
assert success
|
|
assert save_path.exists()
|
|
|
|
restored = filesystem.restore_game()
|
|
assert restored == test_data
|
|
|
|
|
|
# Integration tests with real zork1.z3
|
|
|
|
|
|
@requires_zork
|
|
@pytest.mark.asyncio
|
|
async def test_embedded_session_start():
|
|
"""EmbeddedIFSession starts and returns intro containing game info."""
|
|
from mudlib.embedded_if_session import EmbeddedIFSession
|
|
from mudlib.player import Player
|
|
|
|
mock_writer = MockWriter()
|
|
player = Player(name="tester", x=5, y=5, writer=mock_writer)
|
|
|
|
session = EmbeddedIFSession(player, str(ZORK_PATH), "zork1")
|
|
# Clean up any existing save to get a fresh start
|
|
if session.save_path.exists():
|
|
session.save_path.unlink()
|
|
|
|
intro = await session.start()
|
|
|
|
assert intro is not None
|
|
assert len(intro) > 0
|
|
# Intro should contain game title or location
|
|
assert "ZORK" in intro or "West of House" in intro
|
|
|
|
|
|
@requires_zork
|
|
@pytest.mark.asyncio
|
|
async def test_embedded_session_handle_input():
|
|
"""EmbeddedIFSession handles input and returns response."""
|
|
from mudlib.embedded_if_session import EmbeddedIFSession
|
|
from mudlib.player import Player
|
|
|
|
mock_writer = MockWriter()
|
|
player = Player(name="tester", x=5, y=5, writer=mock_writer)
|
|
|
|
session = EmbeddedIFSession(player, str(ZORK_PATH), "zork1")
|
|
# Clean up any existing save to get a fresh start
|
|
if session.save_path.exists():
|
|
session.save_path.unlink()
|
|
|
|
await session.start()
|
|
|
|
response = await session.handle_input("look")
|
|
|
|
assert response is not None
|
|
assert response.done is False
|
|
assert len(response.output) > 0
|
|
# Looking should describe the starting location
|
|
assert "West of House" in response.output or "house" in response.output
|
|
|
|
|
|
@requires_zork
|
|
@pytest.mark.asyncio
|
|
async def test_embedded_session_escape_help():
|
|
"""EmbeddedIFSession ::help returns help text."""
|
|
from mudlib.embedded_if_session import EmbeddedIFSession
|
|
from mudlib.player import Player
|
|
|
|
mock_writer = MockWriter()
|
|
player = Player(name="tester", x=5, y=5, writer=mock_writer)
|
|
|
|
session = EmbeddedIFSession(player, str(ZORK_PATH), "zork1")
|
|
await session.start()
|
|
|
|
response = await session.handle_input("::help")
|
|
|
|
assert response.done is False
|
|
assert "::quit" in response.output
|
|
assert "::save" in response.output
|
|
assert "::help" in response.output
|
|
|
|
|
|
@requires_zork
|
|
@pytest.mark.asyncio
|
|
async def test_embedded_session_escape_quit():
|
|
"""EmbeddedIFSession ::quit returns done=True."""
|
|
from mudlib.embedded_if_session import EmbeddedIFSession
|
|
from mudlib.player import Player
|
|
|
|
mock_writer = MockWriter()
|
|
player = Player(name="tester", x=5, y=5, writer=mock_writer)
|
|
|
|
session = EmbeddedIFSession(player, str(ZORK_PATH), "zork1")
|
|
await session.start()
|
|
|
|
response = await session.handle_input("::quit")
|
|
|
|
assert response.done is True
|
|
assert "saved" in response.output.lower()
|
|
|
|
|
|
@requires_zork
|
|
@pytest.mark.asyncio
|
|
async def test_embedded_session_location_name():
|
|
"""EmbeddedIFSession get_location_name returns location after input."""
|
|
from mudlib.embedded_if_session import EmbeddedIFSession
|
|
from mudlib.player import Player
|
|
|
|
mock_writer = MockWriter()
|
|
player = Player(name="tester", x=5, y=5, writer=mock_writer)
|
|
|
|
session = EmbeddedIFSession(player, str(ZORK_PATH), "zork1")
|
|
await session.start()
|
|
|
|
# Send a command to advance game state
|
|
await session.handle_input("look")
|
|
|
|
location = session.get_location_name()
|
|
|
|
# Location may be None or a string depending on game state
|
|
assert location is None or isinstance(location, str)
|
|
|
|
|
|
@requires_zork
|
|
@pytest.mark.asyncio
|
|
async def test_embedded_session_room_objects():
|
|
"""EmbeddedIFSession get_room_objects returns a list after start."""
|
|
from mudlib.embedded_if_session import EmbeddedIFSession
|
|
from mudlib.player import Player
|
|
|
|
mock_writer = MockWriter()
|
|
player = Player(name="tester", x=5, y=5, writer=mock_writer)
|
|
|
|
session = EmbeddedIFSession(player, str(ZORK_PATH), "zork1")
|
|
await session.start()
|
|
|
|
objects = session.get_room_objects()
|
|
|
|
assert isinstance(objects, list)
|
|
# Zork1 starting location usually has some objects
|
|
assert len(objects) >= 0 # May or may not have visible objects initially
|
|
|
|
|
|
@requires_zork
|
|
@pytest.mark.asyncio
|
|
async def test_embedded_session_try_restore_before_thread():
|
|
"""_try_restore() is called synchronously before interpreter thread starts."""
|
|
from unittest.mock import patch
|
|
|
|
from mudlib.embedded_if_session import EmbeddedIFSession
|
|
from mudlib.player import Player
|
|
|
|
mock_writer = MockWriter()
|
|
player = Player(name="tester", x=5, y=5, writer=mock_writer)
|
|
session = EmbeddedIFSession(player, str(ZORK_PATH), "zork1")
|
|
|
|
# Create a save file
|
|
if session.save_path.exists():
|
|
session.save_path.unlink()
|
|
session.save_path.parent.mkdir(parents=True, exist_ok=True)
|
|
# Write a minimal valid save (header only, won't actually restore correctly)
|
|
session.save_path.write_bytes(b"FORM\x00\x00\x00\x08IFZSQUTZ\x00\x00\x00\x00")
|
|
|
|
call_order = []
|
|
|
|
original_try_restore = session._try_restore
|
|
original_run_interpreter = session._run_interpreter
|
|
|
|
def track_try_restore():
|
|
call_order.append("try_restore")
|
|
return original_try_restore()
|
|
|
|
def track_run_interpreter():
|
|
call_order.append("run_interpreter")
|
|
original_run_interpreter()
|
|
|
|
with (
|
|
patch.object(session, "_try_restore", side_effect=track_try_restore),
|
|
patch.object(session, "_run_interpreter", side_effect=track_run_interpreter),
|
|
):
|
|
await session.start()
|
|
|
|
# Verify _try_restore was called before _run_interpreter
|
|
assert call_order[0] == "try_restore"
|
|
assert call_order[1] == "run_interpreter"
|
|
|
|
await session.stop()
|
|
if session.save_path.exists():
|
|
session.save_path.unlink()
|
|
|
|
|
|
@requires_zork
|
|
@pytest.mark.asyncio
|
|
async def test_embedded_session_no_restore_without_save():
|
|
"""start() does not restore when no save file exists."""
|
|
from mudlib.embedded_if_session import EmbeddedIFSession
|
|
from mudlib.player import Player
|
|
|
|
mock_writer = MockWriter()
|
|
player = Player(name="nosaveplayer", writer=mock_writer, x=0, y=0)
|
|
session = EmbeddedIFSession(player, str(ZORK_PATH), "zork1")
|
|
|
|
# Ensure no save file exists
|
|
if session.save_path.exists():
|
|
session.save_path.unlink()
|
|
|
|
intro = await session.start()
|
|
|
|
# Should NOT contain restore message
|
|
assert "restoring" not in intro.lower()
|
|
# Should contain normal game intro
|
|
assert "ZORK" in intro or "West of House" in intro
|
|
|
|
await session.stop()
|
|
|
|
|
|
@requires_zork
|
|
@pytest.mark.asyncio
|
|
async def test_embedded_session_save_and_restore():
|
|
"""Save a game, create new session, restore it via start()."""
|
|
from mudlib.embedded_if_session import EmbeddedIFSession
|
|
from mudlib.player import Player
|
|
|
|
mock_writer = MockWriter()
|
|
# Start first session
|
|
player = Player(name="testplayer", writer=mock_writer, x=0, y=0)
|
|
session = EmbeddedIFSession(player, str(ZORK_PATH), "zork1")
|
|
# Clean up any existing save to get a fresh start
|
|
if session.save_path.exists():
|
|
session.save_path.unlink()
|
|
|
|
await session.start()
|
|
|
|
# Do something to change state
|
|
await session.handle_input("open mailbox")
|
|
|
|
# Save
|
|
save_result = await session.handle_input("::save")
|
|
assert "saved" in save_result.output.lower()
|
|
await session.stop()
|
|
|
|
# Start new session - should auto-restore via start()
|
|
# start() calls _try_restore() BEFORE launching the interpreter thread
|
|
session2 = EmbeddedIFSession(player, str(ZORK_PATH), "zork1")
|
|
intro = await session2.start()
|
|
# Should contain restore message prefixed to output
|
|
assert "restoring saved game" in intro.lower()
|
|
assert "restored" in intro.lower()
|
|
|
|
# The game state should reflect the restored state
|
|
# (location may differ after restore, just verify it works)
|
|
response = await session2.handle_input("look")
|
|
assert response.output # Should get some output
|
|
await session2.stop()
|
|
|
|
# Clean up save file
|
|
if session2.save_path.exists():
|
|
session2.save_path.unlink()
|