685 lines
20 KiB
Python
685 lines
20 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_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 "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()
|
|
|
|
|
|
@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"
|