#!/usr/bin/env python3 """ Bannerlord Harness Test Suite Comprehensive tests for the Bannerlord MCP Harness implementing the GamePortal Protocol. Test Categories: - Unit Tests: Test individual components in isolation - Mock Tests: Test without requiring Bannerlord or MCP servers running - Integration Tests: Test with actual MCP servers (skip if game not running) - ODA Loop Tests: Test the full Observe-Decide-Act cycle Usage: pytest tests/test_bannerlord_harness.py -v pytest tests/test_bannerlord_harness.py -v -k mock # Only mock tests pytest tests/test_bannerlord_harness.py -v --run-integration # Include integration tests """ import asyncio import json import os import sys from pathlib import Path from unittest.mock import AsyncMock, MagicMock, Mock, patch import pytest # Ensure nexus module is importable sys.path.insert(0, str(Path(__file__).parent.parent)) from nexus.bannerlord_harness import ( BANNERLORD_APP_ID, BANNERLORD_WINDOW_TITLE, ActionResult, BannerlordHarness, GameContext, GameState, MCPClient, VisualState, simple_test_decision, ) # Mark all tests in this file as asyncio pytestmark = pytest.mark.asyncio # ═══════════════════════════════════════════════════════════════════════════ # FIXTURES # ═══════════════════════════════════════════════════════════════════════════ @pytest.fixture def mock_mcp_client(): """Create a mock MCP client for testing.""" client = MagicMock(spec=MCPClient) client.call_tool = AsyncMock(return_value="success") client.list_tools = AsyncMock(return_value=["click", "press_key", "take_screenshot"]) client.start = AsyncMock(return_value=True) client.stop = Mock() return client @pytest.fixture def mock_harness(): """Create a BannerlordHarness in mock mode.""" harness = BannerlordHarness(enable_mock=True) harness.session_id = "test-session-001" return harness @pytest.fixture def mock_harness_with_ws(): """Create a mock harness with mocked WebSocket.""" harness = BannerlordHarness(enable_mock=True) harness.session_id = "test-session-002" harness.ws_connected = True harness.ws = AsyncMock() return harness @pytest.fixture def sample_game_state(): """Create a sample GameState for testing.""" return GameState( portal_id="bannerlord", session_id="test-session", visual=VisualState( screenshot_path="/tmp/test_capture.png", screen_size=(1920, 1080), mouse_position=(960, 540), window_found=True, window_title=BANNERLORD_WINDOW_TITLE, ), game_context=GameContext( app_id=BANNERLORD_APP_ID, playtime_hours=142.5, achievements_unlocked=23, achievements_total=96, current_players_online=8421, game_name="Mount & Blade II: Bannerlord", is_running=True, ), ) # ═══════════════════════════════════════════════════════════════════════════ # GAME STATE DATA CLASS TESTS # ═══════════════════════════════════════════════════════════════════════════ class TestGameState: """Test GameState data class and serialization.""" def test_game_state_default_creation(self): """Test creating a GameState with defaults.""" state = GameState() assert state.portal_id == "bannerlord" assert state.session_id is not None assert len(state.session_id) == 8 assert state.timestamp is not None def test_game_state_to_dict(self): """Test GameState serialization to dict.""" state = GameState( portal_id="bannerlord", session_id="test1234", visual=VisualState( screenshot_path="/tmp/test.png", screen_size=(1920, 1080), mouse_position=(100, 200), window_found=True, window_title="Test Window", ), game_context=GameContext( app_id=261550, playtime_hours=10.5, achievements_unlocked=5, achievements_total=50, current_players_online=1000, game_name="Test Game", is_running=True, ), ) d = state.to_dict() assert d["portal_id"] == "bannerlord" assert d["session_id"] == "test1234" assert d["visual"]["screenshot_path"] == "/tmp/test.png" assert d["visual"]["screen_size"] == [1920, 1080] assert d["visual"]["mouse_position"] == [100, 200] assert d["visual"]["window_found"] is True assert d["game_context"]["app_id"] == 261550 assert d["game_context"]["playtime_hours"] == 10.5 assert d["game_context"]["is_running"] is True def test_visual_state_defaults(self): """Test VisualState default values.""" visual = VisualState() assert visual.screenshot_path is None assert visual.screen_size == (1920, 1080) assert visual.mouse_position == (0, 0) assert visual.window_found is False assert visual.window_title == "" def test_game_context_defaults(self): """Test GameContext default values.""" context = GameContext() assert context.app_id == BANNERLORD_APP_ID assert context.playtime_hours == 0.0 assert context.achievements_unlocked == 0 assert context.achievements_total == 0 assert context.current_players_online == 0 assert context.game_name == "Mount & Blade II: Bannerlord" assert context.is_running is False class TestActionResult: """Test ActionResult data class.""" def test_action_result_default_creation(self): """Test creating ActionResult with defaults.""" result = ActionResult() assert result.success is False assert result.action == "" assert result.params == {} assert result.error is None def test_action_result_to_dict(self): """Test ActionResult serialization.""" result = ActionResult( success=True, action="press_key", params={"key": "space"}, error=None, ) d = result.to_dict() assert d["success"] is True assert d["action"] == "press_key" assert d["params"] == {"key": "space"} assert "error" not in d def test_action_result_with_error(self): """Test ActionResult includes error when present.""" result = ActionResult( success=False, action="click", params={"x": 100, "y": 200}, error="MCP server not running", ) d = result.to_dict() assert d["success"] is False assert d["error"] == "MCP server not running" # ═══════════════════════════════════════════════════════════════════════════ # BANNERLORD HARNESS UNIT TESTS # ═══════════════════════════════════════════════════════════════════════════ class TestBannerlordHarnessUnit: """Unit tests for BannerlordHarness.""" def test_harness_initialization(self): """Test harness initializes with correct defaults.""" harness = BannerlordHarness() assert harness.hermes_ws_url == "ws://localhost:8000/ws" assert harness.enable_mock is False assert harness.session_id is not None assert len(harness.session_id) == 8 assert harness.desktop_mcp is None assert harness.steam_mcp is None assert harness.ws_connected is False def test_harness_mock_mode_initialization(self): """Test harness initializes correctly in mock mode.""" harness = BannerlordHarness(enable_mock=True) assert harness.enable_mock is True assert harness.desktop_mcp is None assert harness.steam_mcp is None async def test_capture_state_returns_gamestate(self, mock_harness): """Test capture_state() returns a valid GameState object.""" state = await mock_harness.capture_state() assert isinstance(state, GameState) assert state.portal_id == "bannerlord" assert state.session_id == "test-session-001" assert "timestamp" in state.to_dict() async def test_capture_state_includes_visual(self, mock_harness): """Test capture_state() includes visual information.""" state = await mock_harness.capture_state() assert isinstance(state.visual, VisualState) assert state.visual.window_found is True assert state.visual.window_title == BANNERLORD_WINDOW_TITLE assert state.visual.screen_size == (1920, 1080) assert state.visual.screenshot_path is not None async def test_capture_state_includes_game_context(self, mock_harness): """Test capture_state() includes game context.""" state = await mock_harness.capture_state() assert isinstance(state.game_context, GameContext) assert state.game_context.app_id == BANNERLORD_APP_ID assert state.game_context.game_name == "Mount & Blade II: Bannerlord" assert state.game_context.is_running is True assert state.game_context.playtime_hours == 142.5 assert state.game_context.current_players_online == 8421 async def test_capture_state_sends_telemetry(self, mock_harness_with_ws): """Test capture_state() sends telemetry when connected.""" harness = mock_harness_with_ws await harness.capture_state() # Verify telemetry was sent assert harness.ws.send.called call_args = harness.ws.send.call_args[0][0] telemetry = json.loads(call_args) assert telemetry["type"] == "game_state_captured" assert telemetry["portal_id"] == "bannerlord" assert telemetry["session_id"] == "test-session-002" # ═══════════════════════════════════════════════════════════════════════════ # MOCK MODE TESTS (No external dependencies) # ═══════════════════════════════════════════════════════════════════════════ class TestMockModeActions: """Test harness actions in mock mode (no game/MCP required).""" async def test_execute_action_click(self, mock_harness): """Test click action in mock mode.""" result = await mock_harness.execute_action({ "type": "click", "x": 100, "y": 200, }) assert isinstance(result, ActionResult) assert result.success is True assert result.action == "click" assert result.params["x"] == 100 assert result.params["y"] == 200 async def test_execute_action_press_key(self, mock_harness): """Test press_key action in mock mode.""" result = await mock_harness.execute_action({ "type": "press_key", "key": "space", }) assert result.success is True assert result.action == "press_key" assert result.params["key"] == "space" async def test_execute_action_hotkey(self, mock_harness): """Test hotkey action in mock mode.""" result = await mock_harness.execute_action({ "type": "hotkey", "keys": "ctrl s", }) assert result.success is True assert result.action == "hotkey" assert result.params["keys"] == "ctrl s" async def test_execute_action_move_to(self, mock_harness): """Test move_to action in mock mode.""" result = await mock_harness.execute_action({ "type": "move_to", "x": 500, "y": 600, }) assert result.success is True assert result.action == "move_to" async def test_execute_action_type_text(self, mock_harness): """Test type_text action in mock mode.""" result = await mock_harness.execute_action({ "type": "type_text", "text": "Hello Bannerlord", }) assert result.success is True assert result.action == "type_text" assert result.params["text"] == "Hello Bannerlord" async def test_execute_action_unknown_type(self, mock_harness): """Test handling of unknown action type.""" result = await mock_harness.execute_action({ "type": "unknown_action", "param": "value", }) # In mock mode, unknown actions still succeed but don't execute assert isinstance(result, ActionResult) assert result.action == "unknown_action" async def test_execute_action_sends_telemetry(self, mock_harness_with_ws): """Test action execution sends telemetry.""" harness = mock_harness_with_ws await harness.execute_action({"type": "press_key", "key": "i"}) # Verify telemetry was sent assert harness.ws.send.called call_args = harness.ws.send.call_args[0][0] telemetry = json.loads(call_args) assert telemetry["type"] == "action_executed" assert telemetry["action"] == "press_key" assert telemetry["success"] is True class TestBannerlordSpecificActions: """Test Bannerlord-specific convenience actions.""" async def test_open_inventory(self, mock_harness): """Test open_inventory() sends 'i' key.""" result = await mock_harness.open_inventory() assert result.success is True assert result.action == "press_key" assert result.params["key"] == "i" async def test_open_character(self, mock_harness): """Test open_character() sends 'c' key.""" result = await mock_harness.open_character() assert result.success is True assert result.action == "press_key" assert result.params["key"] == "c" async def test_open_party(self, mock_harness): """Test open_party() sends 'p' key.""" result = await mock_harness.open_party() assert result.success is True assert result.action == "press_key" assert result.params["key"] == "p" async def test_save_game(self, mock_harness): """Test save_game() sends Ctrl+S.""" result = await mock_harness.save_game() assert result.success is True assert result.action == "hotkey" assert result.params["keys"] == "ctrl s" async def test_load_game(self, mock_harness): """Test load_game() sends Ctrl+L.""" result = await mock_harness.load_game() assert result.success is True assert result.action == "hotkey" assert result.params["keys"] == "ctrl l" # ═══════════════════════════════════════════════════════════════════════════ # ODA LOOP TESTS # ═══════════════════════════════════════════════════════════════════════════ class TestODALoop: """Test the Observe-Decide-Act loop.""" async def test_oda_loop_single_iteration(self, mock_harness): """Test ODA loop completes one iteration.""" actions_executed = [] def decision_fn(state: GameState) -> list[dict]: """Simple decision function for testing.""" return [ {"type": "move_to", "x": 100, "y": 100}, {"type": "press_key", "key": "space"}, ] # Run for 1 iteration await mock_harness.run_observe_decide_act_loop( decision_fn=decision_fn, max_iterations=1, iteration_delay=0.1, ) assert mock_harness.cycle_count == 0 assert mock_harness.running is True async def test_oda_loop_multiple_iterations(self, mock_harness): """Test ODA loop completes multiple iterations.""" iteration_count = [0] def decision_fn(state: GameState) -> list[dict]: iteration_count[0] += 1 return [{"type": "press_key", "key": "space"}] await mock_harness.run_observe_decide_act_loop( decision_fn=decision_fn, max_iterations=3, iteration_delay=0.01, ) assert iteration_count[0] == 3 assert mock_harness.cycle_count == 2 async def test_oda_loop_empty_decisions(self, mock_harness): """Test ODA loop handles empty decision list.""" def decision_fn(state: GameState) -> list[dict]: return [] await mock_harness.run_observe_decide_act_loop( decision_fn=decision_fn, max_iterations=1, iteration_delay=0.01, ) # Should complete without errors assert mock_harness.cycle_count == 0 def test_simple_test_decision_function(self, sample_game_state): """Test the built-in simple_test_decision function.""" actions = simple_test_decision(sample_game_state) assert len(actions) == 2 assert actions[0]["type"] == "move_to" assert actions[0]["x"] == 960 # Center of 1920 assert actions[0]["y"] == 540 # Center of 1080 assert actions[1]["type"] == "press_key" assert actions[1]["key"] == "space" # ═══════════════════════════════════════════════════════════════════════════ # INTEGRATION TESTS (Require MCP servers or game running) # ═══════════════════════════════════════════════════════════════════════════ def integration_test_enabled(): """Check if integration tests should run.""" return os.environ.get("RUN_INTEGRATION_TESTS") == "1" @pytest.mark.skipif( not integration_test_enabled(), reason="Integration tests require RUN_INTEGRATION_TESTS=1 and MCP servers running" ) class TestIntegration: """Integration tests requiring actual MCP servers.""" @pytest.fixture async def real_harness(self): """Create a real harness with MCP servers.""" harness = BannerlordHarness(enable_mock=False) await harness.start() yield harness await harness.stop() async def test_real_capture_state(self, real_harness): """Test capture_state with real MCP servers.""" state = await real_harness.capture_state() assert isinstance(state, GameState) assert state.portal_id == "bannerlord" assert state.visual.screen_size[0] > 0 assert state.visual.screen_size[1] > 0 async def test_real_execute_action(self, real_harness): """Test execute_action with real MCP server.""" # Move mouse to safe position result = await real_harness.execute_action({ "type": "move_to", "x": 100, "y": 100, }) assert result.success is True # ═══════════════════════════════════════════════════════════════════════════ # MCP CLIENT TESTS # ═══════════════════════════════════════════════════════════════════════════ class TestMCPClient: """Test the MCPClient class.""" def test_mcp_client_initialization(self): """Test MCPClient initializes correctly.""" client = MCPClient("test-server", ["npx", "test-mcp"]) assert client.name == "test-server" assert client.command == ["npx", "test-mcp"] assert client.process is None assert client.request_id == 0 async def test_mcp_client_call_tool_not_running(self): """Test calling tool when server not started.""" client = MCPClient("test-server", ["npx", "test-mcp"]) result = await client.call_tool("click", {"x": 100, "y": 200}) assert "error" in result assert "not running" in str(result).lower() # ═══════════════════════════════════════════════════════════════════════════ # TELEMETRY TESTS # ═══════════════════════════════════════════════════════════════════════════ class TestTelemetry: """Test telemetry sending functionality.""" async def test_telemetry_sent_on_state_capture(self, mock_harness_with_ws): """Test telemetry is sent when state is captured.""" harness = mock_harness_with_ws await harness.capture_state() # Should send game_state_captured telemetry calls = harness.ws.send.call_args_list telemetry_types = [json.loads(c[0][0])["type"] for c in calls] assert "game_state_captured" in telemetry_types async def test_telemetry_sent_on_action(self, mock_harness_with_ws): """Test telemetry is sent when action is executed.""" harness = mock_harness_with_ws await harness.execute_action({"type": "press_key", "key": "space"}) # Should send action_executed telemetry calls = harness.ws.send.call_args_list telemetry_types = [json.loads(c[0][0])["type"] for c in calls] assert "action_executed" in telemetry_types async def test_telemetry_not_sent_when_disconnected(self, mock_harness): """Test telemetry is not sent when WebSocket disconnected.""" harness = mock_harness harness.ws_connected = False harness.ws = AsyncMock() await harness.capture_state() # Should not send telemetry when disconnected assert not harness.ws.send.called # ═══════════════════════════════════════════════════════════════════════════ # GAMEPORTAL PROTOCOL COMPLIANCE TESTS # ═══════════════════════════════════════════════════════════════════════════ class TestGamePortalProtocolCompliance: """Test compliance with the GamePortal Protocol specification.""" async def test_capture_state_returns_valid_schema(self, mock_harness): """Test capture_state returns valid GamePortal Protocol schema.""" state = await mock_harness.capture_state() data = state.to_dict() # Required fields per GAMEPORTAL_PROTOCOL.md assert "portal_id" in data assert "timestamp" in data assert "session_id" in data assert "visual" in data assert "game_context" in data # Visual sub-fields visual = data["visual"] assert "screenshot_path" in visual assert "screen_size" in visual assert "mouse_position" in visual assert "window_found" in visual assert "window_title" in visual # Game context sub-fields context = data["game_context"] assert "app_id" in context assert "playtime_hours" in context assert "achievements_unlocked" in context assert "achievements_total" in context assert "current_players_online" in context assert "game_name" in context assert "is_running" in context async def test_execute_action_returns_valid_schema(self, mock_harness): """Test execute_action returns valid ActionResult schema.""" result = await mock_harness.execute_action({ "type": "press_key", "key": "space", }) data = result.to_dict() # Required fields per GAMEPORTAL_PROTOCOL.md assert "success" in data assert "action" in data assert "params" in data assert "timestamp" in data async def test_all_action_types_supported(self, mock_harness): """Test all GamePortal Protocol action types are supported.""" action_types = [ "click", "right_click", "double_click", "move_to", "drag_to", "press_key", "hotkey", "type_text", "scroll", ] for action_type in action_types: action = {"type": action_type} # Add required params based on action type if action_type in ["click", "right_click", "double_click", "move_to", "drag_to"]: action["x"] = 100 action["y"] = 200 elif action_type == "press_key": action["key"] = "space" elif action_type == "hotkey": action["keys"] = "ctrl s" elif action_type == "type_text": action["text"] = "test" elif action_type == "scroll": action["amount"] = 3 result = await mock_harness.execute_action(action) assert isinstance(result, ActionResult), f"Action {action_type} failed" # ═══════════════════════════════════════════════════════════════════════════ # MAIN ENTRYPOINT # ═══════════════════════════════════════════════════════════════════════════ if __name__ == "__main__": pytest.main([__file__, "-v"])