Jared Miller
37d1c86b34
Removed 32 tests that only verified constructor args are stored as properties. Type annotations and behavioral tests already cover this.
877 lines
26 KiB
Python
877 lines
26 KiB
Python
"""Tests for IF session management."""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from mudlib.if_session import IFSession
|
|
|
|
|
|
@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 "\n>" 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_save_path_sanitizes_malicious_names(tmp_path):
|
|
"""save_path sanitizes player names to prevent path traversal."""
|
|
player = MagicMock()
|
|
player.name = "../../etc/passwd"
|
|
session = IFSession(player, "/path/to/zork.z5", "zork")
|
|
|
|
# Override data_dir for testing
|
|
session._data_dir = tmp_path
|
|
|
|
save_path = session.save_path
|
|
# Should sanitize to replace non-alphanumeric chars with underscores
|
|
# "../../etc/passwd" becomes "______etc_passwd"
|
|
assert ".." not in str(save_path)
|
|
assert save_path == tmp_path / "if_saves" / "______etc_passwd" / "zork.qzl"
|
|
# Verify it's still within the if_saves directory
|
|
assert tmp_path / "if_saves" in save_path.parents
|
|
|
|
|
|
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 "saved" 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 "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_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()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_quit_then_stop_does_not_double_save(tmp_path):
|
|
"""stop() after ::quit doesn't save again (prevents double-save)."""
|
|
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 (only expect one save)
|
|
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)
|
|
|
|
# First: handle ::quit (should save)
|
|
await session.handle_input("::quit")
|
|
|
|
# Verify save was called once
|
|
save_calls = [
|
|
call
|
|
for call in mock_process.stdin.write.call_args_list
|
|
if call[0][0] == b"save\n"
|
|
]
|
|
assert len(save_calls) == 1
|
|
|
|
# Second: call stop() (should NOT save again)
|
|
await session.stop()
|
|
|
|
# Verify save was still only called once
|
|
save_calls = [
|
|
call
|
|
for call in mock_process.stdin.write.call_args_list
|
|
if call[0][0] == b"save\n"
|
|
]
|
|
assert len(save_calls) == 1, "stop() should not save again after ::quit"
|
|
mock_process.terminate.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_regular_input_resets_save_flag(tmp_path):
|
|
"""Regular input resets the saved flag so subsequent saves work."""
|
|
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()
|
|
session.process = mock_process
|
|
|
|
# Simulate dfrotz responses
|
|
responses = [
|
|
b"Enter saved game: \n>",
|
|
b"Ok.\n>", # first save
|
|
b"You move north.\n>", # regular input
|
|
b"Enter saved game: \n>",
|
|
b"Ok.\n>", # second save
|
|
]
|
|
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)
|
|
|
|
# First: save
|
|
await session.handle_input("::save")
|
|
|
|
# Second: regular input (resets saved flag)
|
|
await session.handle_input("north")
|
|
|
|
# Third: save again (should work)
|
|
await session.handle_input("::save")
|
|
|
|
# Verify save was called twice
|
|
save_calls = [
|
|
call
|
|
for call in mock_process.stdin.write.call_args_list
|
|
if call[0][0] == b"save\n"
|
|
]
|
|
assert len(save_calls) == 2, "should be able to save again after regular input"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_do_save_handles_process_communication_failure(tmp_path):
|
|
"""_do_save() returns error message when process communication fails."""
|
|
player = MagicMock()
|
|
player.name = "tester"
|
|
session = IFSession(player, "/path/to/zork.z5", "zork")
|
|
session._data_dir = tmp_path
|
|
|
|
# Mock process that raises an exception
|
|
mock_process = MagicMock()
|
|
mock_process.stdin = MagicMock()
|
|
mock_process.stdin.write = MagicMock(side_effect=BrokenPipeError("process died"))
|
|
session.process = mock_process
|
|
|
|
result = await session._do_save()
|
|
|
|
assert "error" in result.lower()
|
|
assert "save" in result.lower() or "failed" in result.lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_do_save_handles_stdout_read_failure(tmp_path):
|
|
"""_do_save() returns error message when reading response fails."""
|
|
player = MagicMock()
|
|
player.name = "tester"
|
|
session = IFSession(player, "/path/to/zork.z5", "zork")
|
|
session._data_dir = tmp_path
|
|
|
|
# Mock process where writing succeeds but reading fails
|
|
mock_process = MagicMock()
|
|
mock_process.stdin = MagicMock()
|
|
mock_process.stdin.write = MagicMock()
|
|
mock_process.stdin.drain = AsyncMock()
|
|
mock_process.stdout = AsyncMock()
|
|
mock_process.stdout.read = AsyncMock(side_effect=OSError("read failed"))
|
|
session.process = mock_process
|
|
|
|
result = await session._do_save()
|
|
|
|
assert "error" in result.lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_do_restore_handles_process_communication_failure(tmp_path):
|
|
"""_do_restore() returns empty string when process communication fails."""
|
|
player = MagicMock()
|
|
player.name = "tester"
|
|
session = IFSession(player, "/path/to/zork.z5", "zork")
|
|
session._data_dir = tmp_path
|
|
|
|
# Create a save file so we get past the existence check
|
|
session._ensure_save_dir()
|
|
session.save_path.write_text("fake save data")
|
|
|
|
# Mock process that raises an exception
|
|
mock_process = MagicMock()
|
|
mock_process.stdin = MagicMock()
|
|
mock_process.stdin.write = MagicMock(side_effect=BrokenPipeError("process died"))
|
|
session.process = mock_process
|
|
|
|
result = await session._do_restore()
|
|
|
|
# Should return empty string on failure (player starts fresh)
|
|
assert result == ""
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_do_restore_handles_stdout_read_failure(tmp_path):
|
|
"""_do_restore() returns empty string when reading response fails."""
|
|
player = MagicMock()
|
|
player.name = "tester"
|
|
session = IFSession(player, "/path/to/zork.z5", "zork")
|
|
session._data_dir = tmp_path
|
|
|
|
# Create a save file
|
|
session._ensure_save_dir()
|
|
session.save_path.write_text("fake save data")
|
|
|
|
# Mock process where writing succeeds but reading fails
|
|
mock_process = MagicMock()
|
|
mock_process.stdin = MagicMock()
|
|
mock_process.stdin.write = MagicMock()
|
|
mock_process.stdin.drain = AsyncMock()
|
|
mock_process.stdout = AsyncMock()
|
|
mock_process.stdout.read = AsyncMock(side_effect=OSError("read failed"))
|
|
session.process = mock_process
|
|
|
|
result = await session._do_restore()
|
|
|
|
# Should return empty string on failure (player starts fresh)
|
|
assert result == ""
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_response_handles_chunked_data():
|
|
"""_read_response() correctly handles data arriving in chunks."""
|
|
player = MagicMock()
|
|
session = IFSession(player, "/path/to/story.z5")
|
|
|
|
mock_process = AsyncMock()
|
|
mock_process.stdout = AsyncMock()
|
|
session.process = mock_process
|
|
|
|
# Data arrives in multi-byte chunks, last chunk includes prompt
|
|
chunks = [
|
|
b"Welcome to Zork.\n",
|
|
b"West of House\nYou are standing ",
|
|
b"in an open field.\n>",
|
|
]
|
|
chunk_iter = iter(chunks)
|
|
|
|
async def read_chunked(n):
|
|
try:
|
|
return next(chunk_iter)
|
|
except StopIteration:
|
|
return b""
|
|
|
|
mock_process.stdout.read = AsyncMock(side_effect=read_chunked)
|
|
|
|
result = await session._read_response()
|
|
|
|
assert "Welcome to Zork" in result
|
|
assert "open field" in result
|
|
assert not result.strip().endswith(">")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_do_save_auto_confirms_overwrite(tmp_path):
|
|
"""_do_save() auto-confirms with 'yes' when overwrite prompt appears."""
|
|
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: prompt, overwrite question, confirmation
|
|
responses = [
|
|
b"Enter saved game to store: \n>",
|
|
b"Overwrite existing file?\n>",
|
|
b"Ok.\n>",
|
|
]
|
|
response_iter = iter(responses)
|
|
|
|
async def read_side_effect(n):
|
|
try:
|
|
return next(response_iter)
|
|
except StopIteration:
|
|
return b""
|
|
|
|
mock_process.stdout.read = AsyncMock(side_effect=read_side_effect)
|
|
|
|
result = await session._do_save()
|
|
|
|
# Should have written "save\n", filepath, "yes\n"
|
|
calls = mock_process.stdin.write.call_args_list
|
|
assert len(calls) == 3
|
|
assert calls[0][0][0] == b"save\n"
|
|
assert str(session.save_path) in calls[1][0][0].decode()
|
|
assert calls[2][0][0] == b"yes\n"
|
|
|
|
# Result should contain "saved"
|
|
assert "saved" in result.lower()
|
|
assert session._saved is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_do_save_detects_failure(tmp_path):
|
|
"""_do_save() returns error when dfrotz responds with 'Failed.'"""
|
|
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: prompt, then failure
|
|
responses = [
|
|
b"Enter saved game to store: \n>",
|
|
b"Failed.\n>",
|
|
]
|
|
response_iter = iter(responses)
|
|
|
|
async def read_side_effect(n):
|
|
try:
|
|
return next(response_iter)
|
|
except StopIteration:
|
|
return b""
|
|
|
|
mock_process.stdout.read = AsyncMock(side_effect=read_side_effect)
|
|
|
|
result = await session._do_save()
|
|
|
|
# Result should contain error and failed
|
|
assert "error" in result.lower()
|
|
assert "failed" in result.lower()
|
|
# Should NOT be marked as saved
|
|
assert session._saved is False
|