Jared Miller
57afe9a3ce
When starting an IF game, check for existing save file and restore if present. Shows 'restoring saved game...' message and broadcasts restored game state to spectators. Also cleaned up redundant tests that didn't properly mock the auto-save functionality now present in ::quit and stop().
561 lines
16 KiB
Python
561 lines
16 KiB
Python
"""Tests for IF session management."""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from mudlib.if_session import IFResponse, IFSession
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_if_response_dataclass():
|
|
"""IFResponse dataclass can be created."""
|
|
response = IFResponse(output="test output", done=False)
|
|
assert response.output == "test output"
|
|
assert response.done is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_if_response_done():
|
|
"""IFResponse can signal completion."""
|
|
response = IFResponse(output="", done=True)
|
|
assert response.done is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_if_session_init():
|
|
"""IFSession can be initialized."""
|
|
player = MagicMock()
|
|
session = IFSession(player, "/path/to/story.z5", "story")
|
|
assert session.player == player
|
|
assert session.story_path == "/path/to/story.z5"
|
|
assert session.game_name == "story"
|
|
assert session.process is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_if_session_init_infers_game_name():
|
|
"""IFSession infers game_name from story_path if not provided."""
|
|
player = MagicMock()
|
|
session = IFSession(player, "/path/to/zork.z5")
|
|
assert session.game_name == "zork"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_spawns_subprocess_and_returns_intro():
|
|
"""start() spawns dfrotz subprocess and returns intro text."""
|
|
player = MagicMock()
|
|
session = IFSession(player, "/path/to/story.z5")
|
|
|
|
# Mock the subprocess
|
|
mock_process = AsyncMock()
|
|
mock_process.stdout = AsyncMock()
|
|
mock_process.stdin = AsyncMock()
|
|
|
|
# Simulate dfrotz output: intro text followed by ">" prompt
|
|
intro_bytes = b"Welcome to the story!\nYou are in a room.\n>"
|
|
|
|
async def read_side_effect(n):
|
|
nonlocal intro_bytes
|
|
if intro_bytes:
|
|
byte = intro_bytes[:1]
|
|
intro_bytes = intro_bytes[1:]
|
|
return byte
|
|
return b""
|
|
|
|
mock_process.stdout.read = AsyncMock(side_effect=read_side_effect)
|
|
|
|
with patch("asyncio.create_subprocess_exec", return_value=mock_process):
|
|
intro = await session.start()
|
|
|
|
assert session.process == mock_process
|
|
assert "Welcome to the story!" in intro
|
|
assert "You are in a room." in intro
|
|
# The prompt should be stripped from the output
|
|
assert intro.strip().endswith("room.")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_input_sends_to_dfrotz():
|
|
"""handle_input() sends regular input to dfrotz and returns response."""
|
|
player = MagicMock()
|
|
session = IFSession(player, "/path/to/story.z5")
|
|
|
|
# Mock process
|
|
mock_process = MagicMock()
|
|
mock_process.stdin = MagicMock()
|
|
mock_process.stdin.write = MagicMock()
|
|
mock_process.stdin.drain = AsyncMock()
|
|
mock_process.stdout = AsyncMock()
|
|
session.process = mock_process
|
|
|
|
# Simulate dfrotz response
|
|
response_bytes = b"You move north.\n>"
|
|
|
|
async def read_side_effect(n):
|
|
nonlocal response_bytes
|
|
if response_bytes:
|
|
byte = response_bytes[:1]
|
|
response_bytes = response_bytes[1:]
|
|
return byte
|
|
return b""
|
|
|
|
mock_process.stdout.read = AsyncMock(side_effect=read_side_effect)
|
|
|
|
response = await session.handle_input("north")
|
|
|
|
assert response.output == "You move north."
|
|
assert response.done is False
|
|
# Verify stdin.write was called
|
|
mock_process.stdin.write.assert_called()
|
|
mock_process.stdin.drain.assert_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_input_help_returns_help_text():
|
|
"""handle_input('::help') returns help text listing escape commands."""
|
|
player = MagicMock()
|
|
session = IFSession(player, "/path/to/story.z5")
|
|
session.process = AsyncMock()
|
|
|
|
response = await session.handle_input("::help")
|
|
|
|
assert response.done is False
|
|
assert "::quit" in response.output
|
|
assert "::help" in response.output
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_when_no_process():
|
|
"""stop() does nothing if process is None."""
|
|
player = MagicMock()
|
|
session = IFSession(player, "/path/to/story.z5")
|
|
session.process = None
|
|
|
|
# Should not raise
|
|
await session.stop()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_when_already_terminated():
|
|
"""stop() handles already-terminated process gracefully."""
|
|
player = MagicMock()
|
|
session = IFSession(player, "/path/to/story.z5")
|
|
|
|
# Mock process that's already done
|
|
mock_process = AsyncMock()
|
|
mock_process.returncode = 0
|
|
session.process = mock_process
|
|
|
|
await session.stop()
|
|
|
|
# Should not call terminate on already-finished process
|
|
mock_process.terminate.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_response_detects_prompt():
|
|
"""_read_response() reads until '>' prompt appears."""
|
|
player = MagicMock()
|
|
session = IFSession(player, "/path/to/story.z5")
|
|
|
|
# Mock process
|
|
mock_process = AsyncMock()
|
|
mock_process.stdout = AsyncMock()
|
|
session.process = mock_process
|
|
|
|
# Simulate multi-line output with prompt
|
|
output_bytes = b"Line 1\nLine 2\nLine 3\n>"
|
|
|
|
async def read_side_effect(n):
|
|
nonlocal output_bytes
|
|
if output_bytes:
|
|
byte = output_bytes[:1]
|
|
output_bytes = output_bytes[1:]
|
|
return byte
|
|
return b""
|
|
|
|
mock_process.stdout.read = AsyncMock(side_effect=read_side_effect)
|
|
|
|
result = await session._read_response()
|
|
|
|
assert "Line 1" in result
|
|
assert "Line 2" in result
|
|
assert "Line 3" in result
|
|
# Prompt should be stripped
|
|
assert not result.strip().endswith(">")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_input_strips_whitespace():
|
|
"""handle_input() strips input before sending to dfrotz."""
|
|
player = MagicMock()
|
|
session = IFSession(player, "/path/to/story.z5")
|
|
|
|
# Mock process
|
|
mock_process = MagicMock()
|
|
mock_process.stdin = MagicMock()
|
|
mock_process.stdin.write = MagicMock()
|
|
mock_process.stdin.drain = AsyncMock()
|
|
mock_process.stdout = AsyncMock()
|
|
session.process = mock_process
|
|
|
|
# Simulate response
|
|
response_bytes = b"ok\n>"
|
|
|
|
async def read_side_effect(n):
|
|
nonlocal response_bytes
|
|
if response_bytes:
|
|
byte = response_bytes[:1]
|
|
response_bytes = response_bytes[1:]
|
|
return byte
|
|
return b""
|
|
|
|
mock_process.stdout.read = AsyncMock(side_effect=read_side_effect)
|
|
|
|
await session.handle_input(" look ")
|
|
|
|
# Check that write was called with stripped input + newline
|
|
calls = mock_process.stdin.write.call_args_list
|
|
assert len(calls) == 1
|
|
written = calls[0][0][0]
|
|
assert written == b"look\n"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_input_empty_string():
|
|
"""handle_input() with empty string sends newline to dfrotz."""
|
|
player = MagicMock()
|
|
session = IFSession(player, "/path/to/story.z5")
|
|
|
|
# Mock process
|
|
mock_process = MagicMock()
|
|
mock_process.stdin = MagicMock()
|
|
mock_process.stdin.write = MagicMock()
|
|
mock_process.stdin.drain = AsyncMock()
|
|
mock_process.stdout = AsyncMock()
|
|
session.process = mock_process
|
|
|
|
# Simulate response
|
|
response_bytes = b"ok\n>"
|
|
|
|
async def read_side_effect(n):
|
|
nonlocal response_bytes
|
|
if response_bytes:
|
|
byte = response_bytes[:1]
|
|
response_bytes = response_bytes[1:]
|
|
return byte
|
|
return b""
|
|
|
|
mock_process.stdout.read = AsyncMock(side_effect=read_side_effect)
|
|
|
|
await session.handle_input("")
|
|
|
|
# Should still write a newline
|
|
mock_process.stdin.write.assert_called()
|
|
|
|
|
|
def test_save_path_property(tmp_path):
|
|
"""save_path returns correct path for player/game combo."""
|
|
player = MagicMock()
|
|
player.name = "tester"
|
|
session = IFSession(player, "/path/to/zork.z5", "zork")
|
|
|
|
# Override data_dir for testing
|
|
session._data_dir = tmp_path
|
|
|
|
save_path = session.save_path
|
|
assert save_path == tmp_path / "if_saves" / "tester" / "zork.qzl"
|
|
|
|
|
|
def test_ensure_save_dir_creates_directories(tmp_path):
|
|
"""_ensure_save_dir() creates parent directories."""
|
|
player = MagicMock()
|
|
player.name = "alice"
|
|
session = IFSession(player, "/path/to/story.z5", "story")
|
|
|
|
# Override data_dir for testing
|
|
session._data_dir = tmp_path
|
|
|
|
# Directory shouldn't exist yet
|
|
expected_dir = tmp_path / "if_saves" / "alice"
|
|
assert not expected_dir.exists()
|
|
|
|
# Call _ensure_save_dir
|
|
session._ensure_save_dir()
|
|
|
|
# Now it should exist
|
|
assert expected_dir.exists()
|
|
assert expected_dir.is_dir()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_do_save_sends_save_command(tmp_path):
|
|
"""_do_save() sends save command and filepath to dfrotz."""
|
|
player = MagicMock()
|
|
player.name = "tester"
|
|
session = IFSession(player, "/path/to/zork.z5", "zork")
|
|
session._data_dir = tmp_path
|
|
|
|
# Mock process
|
|
mock_process = MagicMock()
|
|
mock_process.stdin = MagicMock()
|
|
mock_process.stdin.write = MagicMock()
|
|
mock_process.stdin.drain = AsyncMock()
|
|
mock_process.stdout = AsyncMock()
|
|
session.process = mock_process
|
|
|
|
# Simulate dfrotz responses: first the filename prompt, then confirmation
|
|
responses = [
|
|
b"Enter saved game to store: \n>",
|
|
b"Ok.\n>",
|
|
]
|
|
response_data = b"".join(responses)
|
|
|
|
async def read_side_effect(n):
|
|
nonlocal response_data
|
|
if response_data:
|
|
byte = response_data[:1]
|
|
response_data = response_data[1:]
|
|
return byte
|
|
return b""
|
|
|
|
mock_process.stdout.read = AsyncMock(side_effect=read_side_effect)
|
|
|
|
result = await session._do_save()
|
|
|
|
# Should have written "save\n" then the filepath
|
|
calls = mock_process.stdin.write.call_args_list
|
|
assert len(calls) == 2
|
|
assert calls[0][0][0] == b"save\n"
|
|
assert str(session.save_path) in calls[1][0][0].decode()
|
|
assert b"\n" in calls[1][0][0]
|
|
|
|
# Result should contain confirmation
|
|
assert "Ok" in result or "ok" in result.lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_do_save_creates_save_directory(tmp_path):
|
|
"""_do_save() ensures save directory exists."""
|
|
player = MagicMock()
|
|
player.name = "alice"
|
|
session = IFSession(player, "/path/to/story.z5", "story")
|
|
session._data_dir = tmp_path
|
|
|
|
# Mock process
|
|
mock_process = MagicMock()
|
|
mock_process.stdin = MagicMock()
|
|
mock_process.stdin.write = MagicMock()
|
|
mock_process.stdin.drain = AsyncMock()
|
|
mock_process.stdout = AsyncMock()
|
|
session.process = mock_process
|
|
|
|
# Simulate dfrotz responses
|
|
responses = [b"Enter saved game: \n>", b"Ok.\n>"]
|
|
response_data = b"".join(responses)
|
|
|
|
async def read_side_effect(n):
|
|
nonlocal response_data
|
|
if response_data:
|
|
byte = response_data[:1]
|
|
response_data = response_data[1:]
|
|
return byte
|
|
return b""
|
|
|
|
mock_process.stdout.read = AsyncMock(side_effect=read_side_effect)
|
|
|
|
# Directory shouldn't exist yet
|
|
expected_dir = tmp_path / "if_saves" / "alice"
|
|
assert not expected_dir.exists()
|
|
|
|
await session._do_save()
|
|
|
|
# Now it should exist
|
|
assert expected_dir.exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_do_restore_returns_empty_if_no_save(tmp_path):
|
|
"""_do_restore() returns empty string if no save file exists."""
|
|
player = MagicMock()
|
|
player.name = "tester"
|
|
session = IFSession(player, "/path/to/zork.z5", "zork")
|
|
session._data_dir = tmp_path
|
|
|
|
# Mock process (even though we won't use it)
|
|
session.process = MagicMock()
|
|
|
|
# No save file exists
|
|
assert not session.save_path.exists()
|
|
|
|
result = await session._do_restore()
|
|
|
|
assert result == ""
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_do_restore_sends_restore_command(tmp_path):
|
|
"""_do_restore() sends restore command and filepath to dfrotz."""
|
|
player = MagicMock()
|
|
player.name = "tester"
|
|
session = IFSession(player, "/path/to/zork.z5", "zork")
|
|
session._data_dir = tmp_path
|
|
|
|
# Create a fake save file
|
|
session._ensure_save_dir()
|
|
session.save_path.write_text("fake save data")
|
|
|
|
# Mock process
|
|
mock_process = MagicMock()
|
|
mock_process.stdin = MagicMock()
|
|
mock_process.stdin.write = MagicMock()
|
|
mock_process.stdin.drain = AsyncMock()
|
|
mock_process.stdout = AsyncMock()
|
|
session.process = mock_process
|
|
|
|
# Simulate dfrotz responses: filename prompt, then game text
|
|
responses = [
|
|
b"Enter saved game to load: \n>",
|
|
b"West of House\nYou are standing in an open field.\n>",
|
|
]
|
|
response_data = b"".join(responses)
|
|
|
|
async def read_side_effect(n):
|
|
nonlocal response_data
|
|
if response_data:
|
|
byte = response_data[:1]
|
|
response_data = response_data[1:]
|
|
return byte
|
|
return b""
|
|
|
|
mock_process.stdout.read = AsyncMock(side_effect=read_side_effect)
|
|
|
|
result = await session._do_restore()
|
|
|
|
# Should have written "restore\n" then the filepath
|
|
calls = mock_process.stdin.write.call_args_list
|
|
assert len(calls) == 2
|
|
assert calls[0][0][0] == b"restore\n"
|
|
assert str(session.save_path) in calls[1][0][0].decode()
|
|
|
|
# Result should contain game text
|
|
assert "West of House" in result
|
|
assert "open field" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_input_save_triggers_save(tmp_path):
|
|
"""handle_input('::save') triggers save and returns confirmation."""
|
|
player = MagicMock()
|
|
player.name = "tester"
|
|
session = IFSession(player, "/path/to/zork.z5", "zork")
|
|
session._data_dir = tmp_path
|
|
|
|
# Mock process
|
|
mock_process = MagicMock()
|
|
mock_process.stdin = MagicMock()
|
|
mock_process.stdin.write = MagicMock()
|
|
mock_process.stdin.drain = AsyncMock()
|
|
mock_process.stdout = AsyncMock()
|
|
session.process = mock_process
|
|
|
|
# Simulate dfrotz save responses
|
|
responses = [b"Enter saved game: \n>", b"Ok.\n>"]
|
|
response_data = b"".join(responses)
|
|
|
|
async def read_side_effect(n):
|
|
nonlocal response_data
|
|
if response_data:
|
|
byte = response_data[:1]
|
|
response_data = response_data[1:]
|
|
return byte
|
|
return b""
|
|
|
|
mock_process.stdout.read = AsyncMock(side_effect=read_side_effect)
|
|
|
|
response = await session.handle_input("::save")
|
|
|
|
assert response.done is False
|
|
assert "Ok" in response.output or "ok" in response.output.lower()
|
|
# Verify save command was sent
|
|
assert mock_process.stdin.write.call_args_list[0][0][0] == b"save\n"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handle_input_quit_saves_before_exit(tmp_path):
|
|
"""handle_input('::quit') saves game before returning done=True."""
|
|
player = MagicMock()
|
|
player.name = "tester"
|
|
session = IFSession(player, "/path/to/zork.z5", "zork")
|
|
session._data_dir = tmp_path
|
|
|
|
# Mock process
|
|
mock_process = MagicMock()
|
|
mock_process.stdin = MagicMock()
|
|
mock_process.stdin.write = MagicMock()
|
|
mock_process.stdin.drain = AsyncMock()
|
|
mock_process.stdout = AsyncMock()
|
|
session.process = mock_process
|
|
|
|
# Simulate dfrotz save responses
|
|
responses = [b"Enter saved game: \n>", b"Ok.\n>"]
|
|
response_data = b"".join(responses)
|
|
|
|
async def read_side_effect(n):
|
|
nonlocal response_data
|
|
if response_data:
|
|
byte = response_data[:1]
|
|
response_data = response_data[1:]
|
|
return byte
|
|
return b""
|
|
|
|
mock_process.stdout.read = AsyncMock(side_effect=read_side_effect)
|
|
|
|
response = await session.handle_input("::quit")
|
|
|
|
assert response.done is True
|
|
assert "saved" in response.output.lower()
|
|
# Verify save command was sent
|
|
assert mock_process.stdin.write.call_args_list[0][0][0] == b"save\n"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_saves_before_terminating(tmp_path):
|
|
"""stop() saves game before terminating process."""
|
|
player = MagicMock()
|
|
player.name = "tester"
|
|
session = IFSession(player, "/path/to/zork.z5", "zork")
|
|
session._data_dir = tmp_path
|
|
|
|
# Mock process
|
|
mock_process = MagicMock()
|
|
mock_process.returncode = None
|
|
mock_process.stdin = MagicMock()
|
|
mock_process.stdin.write = MagicMock()
|
|
mock_process.stdin.drain = AsyncMock()
|
|
mock_process.stdout = AsyncMock()
|
|
mock_process.terminate = MagicMock()
|
|
mock_process.wait = AsyncMock()
|
|
session.process = mock_process
|
|
|
|
# Simulate dfrotz save responses
|
|
responses = [b"Enter saved game: \n>", b"Ok.\n>"]
|
|
response_data = b"".join(responses)
|
|
|
|
async def read_side_effect(n):
|
|
nonlocal response_data
|
|
if response_data:
|
|
byte = response_data[:1]
|
|
response_data = response_data[1:]
|
|
return byte
|
|
return b""
|
|
|
|
mock_process.stdout.read = AsyncMock(side_effect=read_side_effect)
|
|
|
|
await session.stop()
|
|
|
|
# Verify save was called before terminate
|
|
assert mock_process.stdin.write.call_args_list[0][0][0] == b"save\n"
|
|
mock_process.terminate.assert_called_once()
|
|
mock_process.wait.assert_called_once()
|