Add IFSession class for interactive fiction subprocess management
TDD implementation of IFSession that manages a dfrotz subprocess. IFResponse dataclass follows the editor pattern with output/done fields. IFSession handles spawning dfrotz, routing input, and detecting the prompt. Escape commands (::quit, ::help) are handled without sending to dfrotz.
This commit is contained in:
parent
bc5e829f6b
commit
d210033f33
GPG key ID:
22B5C6D66A38B06C
2 changed files with 397 additions and 0 deletions
105
src/mudlib/if_session.py
Normal file
105
src/mudlib/if_session.py
Normal file
|
|
@ -0,0 +1,105 @@
|
|||
"""Interactive fiction session management via dfrotz subprocess."""
|
||||
|
||||
import asyncio
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
@dataclass
|
||||
class IFResponse:
|
||||
"""Response from IF session input handling."""
|
||||
|
||||
output: str
|
||||
done: bool
|
||||
|
||||
|
||||
class IFSession:
|
||||
"""Manages an interactive fiction session via dfrotz subprocess."""
|
||||
|
||||
def __init__(self, player, story_path: str, game_name: str = ""):
|
||||
self.player = player
|
||||
self.story_path = story_path
|
||||
self.game_name = game_name or Path(story_path).stem
|
||||
self.process: asyncio.subprocess.Process | None = None
|
||||
|
||||
async def start(self) -> str:
|
||||
"""Spawn dfrotz and return intro text."""
|
||||
self.process = await asyncio.create_subprocess_exec(
|
||||
"dfrotz",
|
||||
"-p", # plain ASCII, no formatting
|
||||
"-w",
|
||||
"80", # screen width
|
||||
"-m", # no MORE prompts
|
||||
self.story_path,
|
||||
stdin=asyncio.subprocess.PIPE,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
)
|
||||
# Read intro text until we see the prompt
|
||||
intro = await self._read_response()
|
||||
return intro
|
||||
|
||||
async def handle_input(self, text: str) -> IFResponse:
|
||||
"""Handle player input. Route to dfrotz or handle escape commands."""
|
||||
# Check for escape commands (:: prefix)
|
||||
if text == "::quit":
|
||||
return IFResponse(output="", done=True)
|
||||
|
||||
if text == "::help":
|
||||
help_text = """escape commands:
|
||||
::quit - exit the game
|
||||
::help - show this help"""
|
||||
return IFResponse(output=help_text, done=False)
|
||||
|
||||
# Regular game input - send to dfrotz
|
||||
if self.process and self.process.stdin:
|
||||
stripped = text.strip()
|
||||
self.process.stdin.write(f"{stripped}\n".encode())
|
||||
await self.process.stdin.drain()
|
||||
|
||||
# Read response
|
||||
output = await self._read_response()
|
||||
return IFResponse(output=output, done=False)
|
||||
|
||||
# Process not running
|
||||
return IFResponse(output="error: game not running", done=True)
|
||||
|
||||
async def stop(self):
|
||||
"""Terminate dfrotz process."""
|
||||
if self.process and self.process.returncode is None:
|
||||
self.process.terminate()
|
||||
await self.process.wait()
|
||||
|
||||
async def _read_response(self) -> str:
|
||||
"""Read dfrotz output until the '>' prompt appears."""
|
||||
if not self.process or not self.process.stdout:
|
||||
return ""
|
||||
|
||||
output = []
|
||||
try:
|
||||
# Read byte by byte until we see "\n>"
|
||||
while True:
|
||||
byte = await asyncio.wait_for(self.process.stdout.read(1), timeout=5.0)
|
||||
if not byte:
|
||||
break
|
||||
|
||||
char = byte.decode("latin-1")
|
||||
output.append(char)
|
||||
|
||||
# Check if we've hit the prompt
|
||||
# Prompt is "\n>" or just ">" at start
|
||||
if len(output) >= 2:
|
||||
if output[-2] == "\n" and output[-1] == ">":
|
||||
# Strip the trailing "\n>"
|
||||
output = output[:-2]
|
||||
break
|
||||
elif len(output) == 1 and output[0] == ">":
|
||||
# Prompt at very start
|
||||
output = []
|
||||
break
|
||||
except TimeoutError:
|
||||
# If we timeout, return what we got
|
||||
pass
|
||||
|
||||
result = "".join(output)
|
||||
return result.rstrip()
|
||||
292
tests/test_if_session.py
Normal file
292
tests/test_if_session.py
Normal file
|
|
@ -0,0 +1,292 @@
|
|||
"""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()
|
||||
Loading…
Reference in a new issue