Jared Miller
50a1d356a8
Remove redundant bare-truthy and .called checks where more specific content or entity validation already exists on subsequent lines.
368 lines
11 KiB
Python
368 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 is True
|
|
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.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()
|