481 lines
14 KiB
Python
481 lines
14 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_quit_returns_done():
|
|
"""handle_input('::quit') returns done=True without sending to dfrotz."""
|
|
player = MagicMock()
|
|
session = IFSession(player, "/path/to/story.z5")
|
|
|
|
# Mock process
|
|
mock_process = AsyncMock()
|
|
mock_process.stdin = AsyncMock()
|
|
session.process = mock_process
|
|
|
|
response = await session.handle_input("::quit")
|
|
|
|
assert response.done is True
|
|
# Should NOT have written to dfrotz stdin
|
|
mock_process.stdin.write.assert_not_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_terminates_subprocess():
|
|
"""stop() terminates the dfrotz process."""
|
|
player = MagicMock()
|
|
session = IFSession(player, "/path/to/story.z5")
|
|
|
|
# Mock process
|
|
mock_process = MagicMock()
|
|
mock_process.returncode = None
|
|
mock_process.terminate = MagicMock()
|
|
mock_process.wait = AsyncMock()
|
|
session.process = mock_process
|
|
|
|
await session.stop()
|
|
|
|
mock_process.terminate.assert_called_once()
|
|
mock_process.wait.assert_called_once()
|
|
|
|
|
|
@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
|