"""Tests for the server module.""" import asyncio import contextlib import os import tempfile from unittest.mock import AsyncMock, MagicMock, patch import pytest from mudlib import server from mudlib.store import init_db from mudlib.world.terrain import World from mudlib.zone import Zone @pytest.fixture def temp_db(): """Create a temporary database for testing.""" with tempfile.NamedTemporaryFile(delete=False, suffix=".db") as f: db_path = f.name init_db(db_path) yield db_path # Cleanup os.unlink(db_path) def test_port_constant(): assert server.PORT == 6789 assert isinstance(server.PORT, int) def test_shell_exists(): assert callable(server.shell) assert asyncio.iscoroutinefunction(server.shell) def test_run_server_exists(): assert callable(server.run_server) assert asyncio.iscoroutinefunction(server.run_server) def test_find_passable_start(): world = World(seed=42, width=100, height=100) zone = Zone( name="test", width=100, height=100, terrain=world.terrain, toroidal=True ) x, y = server.find_passable_start(zone, 50, 50) assert isinstance(x, int) assert isinstance(y, int) assert zone.is_passable(x, y) @pytest.mark.asyncio async def test_shell_greets_and_accepts_commands(temp_db): world = World(seed=42, width=100, height=100) zone = Zone( name="overworld", width=100, height=100, terrain=world.terrain, toroidal=True ) server._overworld = zone reader = AsyncMock() writer = MagicMock() writer.is_closing.side_effect = [False, False, False, True] writer.drain = AsyncMock() writer.close = MagicMock() readline = "mudlib.server.readline2" with patch(readline, new_callable=AsyncMock) as mock_readline: # Simulate: name, create (y), pass, confirm pass, description, look, quit mock_readline.side_effect = [ "TestPlayer", "y", "password", "password", "A test character", "look", "quit", ] await server.shell(reader, writer) calls = [str(call) for call in writer.write.call_args_list] assert any("Welcome" in call for call in calls) assert any("TestPlayer" in call for call in calls) writer.close.assert_called() @pytest.mark.asyncio async def test_shell_handles_eof(): world = World(seed=42, width=100, height=100) zone = Zone( name="overworld", width=100, height=100, terrain=world.terrain, toroidal=True ) server._overworld = zone reader = AsyncMock() writer = MagicMock() writer.is_closing.return_value = False writer.drain = AsyncMock() writer.close = MagicMock() readline = "mudlib.server.readline2" with patch(readline, new_callable=AsyncMock) as mock_readline: mock_readline.return_value = None await server.shell(reader, writer) writer.close.assert_called_once() @pytest.mark.asyncio async def test_shell_handles_quit(temp_db): world = World(seed=42, width=100, height=100) zone = Zone( name="overworld", width=100, height=100, terrain=world.terrain, toroidal=True ) server._overworld = zone reader = AsyncMock() writer = MagicMock() writer.is_closing.side_effect = [False, False, True] writer.drain = AsyncMock() writer.close = MagicMock() readline = "mudlib.server.readline2" with patch(readline, new_callable=AsyncMock) as mock_readline: # Simulate: name, create (y), pass, confirm pass, description, quit mock_readline.side_effect = [ "TestPlayer", "y", "password", "password", "A test character", "quit", ] await server.shell(reader, writer) calls = [str(call) for call in writer.write.call_args_list] assert any("Goodbye" in call for call in calls) writer.close.assert_called() def test_load_world_config(): """Config loader returns expected values from worlds/earth/config.toml.""" config = server.load_world_config() assert config["world"]["seed"] == 42 assert config["world"]["width"] == 1000 assert config["world"]["height"] == 1000 def test_load_world_config_missing(): """Config loader raises FileNotFoundError for nonexistent world.""" with pytest.raises(FileNotFoundError): server.load_world_config("nonexistent") def test_tick_constants(): """Tick rate and interval are configured correctly.""" assert server.TICK_RATE == 10 assert pytest.approx(0.1) == server.TICK_INTERVAL def test_game_loop_exists(): """Game loop is an async callable.""" assert callable(server.game_loop) assert asyncio.iscoroutinefunction(server.game_loop) @pytest.mark.asyncio async def test_game_loop_calls_clear_expired(): """Game loop calls clear_expired each tick.""" with ( patch("mudlib.server.clear_expired") as mock_clear, patch("mudlib.server.process_combat", new_callable=AsyncMock), patch("mudlib.server.process_mobs", new_callable=AsyncMock), patch("mudlib.server.process_resting", new_callable=AsyncMock), patch("mudlib.server.process_unconscious", new_callable=AsyncMock), patch("mudlib.server.get_game_hour", return_value=12), patch("mudlib.server.process_schedules"), ): task = asyncio.create_task(server.game_loop()) await asyncio.sleep(0.25) task.cancel() with contextlib.suppress(asyncio.CancelledError): await task assert mock_clear.call_count >= 1