Spaces:
Runtime error
Runtime error
| #!/usr/bin/env python3 | |
| """ | |
| Unit Tests for Service Layer Components | |
| Tests individual service components in isolation. | |
| """ | |
| import pytest | |
| import sys | |
| import os | |
| from unittest.mock import Mock, patch | |
| # Add project root to path | |
| sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(__file__)))) | |
| from src.services.player_service import PlayerService | |
| from src.services.chat_service import ChatService | |
| from src.services.npc_service import NPCService | |
| from src.services.mcp_service import MCPService | |
| from src.services.plugin_service import PluginService | |
| class TestPlayerService: | |
| """Test player service functionality.""" | |
| def test_player_service_initialization(self): | |
| """Test player service initializes correctly.""" | |
| service = PlayerService() | |
| assert service is not None | |
| assert hasattr(service, 'players') | |
| def test_create_player(self): | |
| """Test player creation.""" | |
| service = PlayerService() | |
| player_id = service.create_player("TestPlayer") | |
| assert player_id is not None | |
| assert player_id in service.players | |
| assert service.players[player_id]['name'] == "TestPlayer" | |
| def test_get_player(self): | |
| """Test player retrieval.""" | |
| service = PlayerService() | |
| player_id = service.create_player("TestPlayer") | |
| player = service.get_player(player_id) | |
| assert player is not None | |
| assert player['name'] == "TestPlayer" | |
| def test_get_all_players(self): | |
| """Test getting all players.""" | |
| service = PlayerService() | |
| player_id1 = service.create_player("Player1") | |
| player_id2 = service.create_player("Player2") | |
| all_players = service.get_all_players() | |
| assert len(all_players) == 2 | |
| assert player_id1 in all_players | |
| assert player_id2 in all_players | |
| def test_remove_player(self): | |
| """Test player removal.""" | |
| service = PlayerService() | |
| player_id = service.create_player("TestPlayer") | |
| assert player_id in service.players | |
| service.remove_player(player_id) | |
| assert player_id not in service.players | |
| def test_move_player(self): | |
| """Test player movement.""" | |
| service = PlayerService() | |
| player_id = service.create_player("TestPlayer") | |
| # Get initial position | |
| player = service.get_player(player_id) | |
| initial_x, initial_y = player['x'], player['y'] | |
| # Move player | |
| success = service.move_player(player_id, "right") | |
| assert success | |
| # Check new position | |
| player = service.get_player(player_id) | |
| assert player['x'] == initial_x + 1 | |
| assert player['y'] == initial_y | |
| class TestChatService: | |
| """Test chat service functionality.""" | |
| def test_chat_service_initialization(self): | |
| """Test chat service initializes correctly.""" | |
| service = ChatService() | |
| assert service is not None | |
| assert hasattr(service, 'messages') | |
| def test_send_public_message(self): | |
| """Test sending public messages.""" | |
| service = ChatService() | |
| message_id = service.send_public_message("player1", "Hello everyone!") | |
| assert message_id is not None | |
| messages = service.get_recent_messages() | |
| assert len(messages) > 0 | |
| assert any(msg['content'] == "Hello everyone!" for msg in messages) | |
| def test_send_private_message(self): | |
| """Test sending private messages.""" | |
| service = ChatService() | |
| message_id = service.send_private_message("player1", "player2", "Hello privately!") | |
| assert message_id is not None | |
| private_messages = service.get_private_messages("player1", "player2") | |
| assert len(private_messages) > 0 | |
| assert any(msg['content'] == "Hello privately!" for msg in private_messages) | |
| def test_get_recent_messages(self): | |
| """Test getting recent messages.""" | |
| service = ChatService() | |
| service.send_public_message("player1", "Message 1") | |
| service.send_public_message("player2", "Message 2") | |
| messages = service.get_recent_messages(limit=10) | |
| assert len(messages) >= 2 | |
| assert any(msg['content'] == "Message 1" for msg in messages) | |
| assert any(msg['content'] == "Message 2" for msg in messages) | |
| class TestNPCService: | |
| """Test NPC service functionality.""" | |
| def test_npc_service_initialization(self): | |
| """Test NPC service initializes correctly.""" | |
| service = NPCService() | |
| assert service is not None | |
| assert hasattr(service, 'npcs') | |
| def test_get_all_npcs(self): | |
| """Test getting all NPCs.""" | |
| service = NPCService() | |
| npcs = service.get_all_npcs() | |
| assert isinstance(npcs, dict) | |
| # Should have at least Donald NPC | |
| assert len(npcs) > 0 | |
| def test_donald_npc_exists(self): | |
| """Test that Donald NPC is properly initialized.""" | |
| service = NPCService() | |
| npcs = service.get_all_npcs() | |
| # Check if Donald exists | |
| donald_exists = any(npc.get('id') == 'donald' for npc in npcs.values()) | |
| assert donald_exists, "Donald NPC should exist" | |
| # Find Donald and check properties | |
| donald_npc = next((npc for npc in npcs.values() if npc.get('id') == 'donald'), None) | |
| assert donald_npc is not None | |
| assert donald_npc['name'] == 'Donald the Trader' | |
| assert donald_npc['char'] == '💼' | |
| def test_get_npc_by_id(self): | |
| """Test getting NPC by ID.""" | |
| service = NPCService() | |
| donald = service.get_npc_by_id('donald') | |
| assert donald is not None | |
| assert donald['name'] == 'Donald the Trader' | |
| class TestMCPService: | |
| """Test MCP service functionality.""" | |
| def test_mcp_service_initialization(self): | |
| """Test MCP service initializes correctly.""" | |
| service = MCPService() | |
| assert service is not None | |
| def test_register_ai_agent(self): | |
| """Test AI agent registration.""" | |
| service = MCPService() | |
| agent_id = "test_agent" | |
| capabilities = ["chat", "move"] | |
| success = service.register_ai_agent(agent_id, capabilities) | |
| assert success | |
| def test_get_game_state(self): | |
| """Test getting game state.""" | |
| service = MCPService() | |
| game_state = service.get_game_state() | |
| assert isinstance(game_state, dict) | |
| assert 'players' in game_state | |
| assert 'npcs' in game_state | |
| assert 'world' in game_state | |
| class TestPluginService: | |
| """Test plugin service functionality.""" | |
| def test_plugin_service_initialization(self): | |
| """Test plugin service initializes correctly.""" | |
| service = PluginService() | |
| assert service is not None | |
| assert hasattr(service, '_loaded_plugins') | |
| def test_load_plugins(self): | |
| """Test plugin loading.""" | |
| service = PluginService() | |
| # Mock plugin directory | |
| with patch('os.path.exists') as mock_exists, \ | |
| patch('os.listdir') as mock_listdir, \ | |
| patch('importlib.util.spec_from_file_location') as mock_spec: | |
| mock_exists.return_value = True | |
| mock_listdir.return_value = ['test_plugin.py'] | |
| # Mock successful plugin loading | |
| mock_module = Mock() | |
| mock_spec.return_value.loader.load_module.return_value = mock_module | |
| service.load_plugins() | |
| assert len(service._loaded_plugins) >= 0 | |
| def test_get_loaded_plugins(self): | |
| """Test getting loaded plugins.""" | |
| service = PluginService() | |
| plugins = service.get_loaded_plugins() | |
| assert isinstance(plugins, dict) | |
| def test_plugin_directory_exists(self): | |
| """Test that plugin directory exists.""" | |
| plugins_dir = os.path.join( | |
| os.path.dirname(os.path.dirname(os.path.dirname(__file__))), | |
| 'plugins' | |
| ) | |
| assert os.path.exists(plugins_dir), "Plugins directory should exist" | |
| # Check for known plugin files | |
| expected_plugins = [ | |
| 'sample_plugin.py', | |
| 'trading_system_plugin.py', | |
| 'weather_plugin.py', | |
| 'enhanced_chat_plugin.py' | |
| ] | |
| for plugin_file in expected_plugins: | |
| plugin_path = os.path.join(plugins_dir, plugin_file) | |
| assert os.path.exists(plugin_path), f"Plugin {plugin_file} should exist" | |