diff --git a/docs/tests/protocols/index.rst b/docs/tests/protocols/index.rst index 7b83a16..d8dc0ef 100644 --- a/docs/tests/protocols/index.rst +++ b/docs/tests/protocols/index.rst @@ -23,14 +23,15 @@ Protocols Tests test_ut3/index test_unreal2/index test_quake3/index + test_warcraft3/index test_nadeo/index test_battlefield/index test_fivem/index test_palworld/index test_quake2/index test_gamespy2/index + test_flatout2/index test_doom3/index test_vcmp/index test_satisfactory/index test_gamespy3/index - test_renegadex/index \ No newline at end of file diff --git a/docs/tests/protocols/test_flatout2/index.rst b/docs/tests/protocols/test_flatout2/index.rst new file mode 100644 index 0000000..bb8cf82 --- /dev/null +++ b/docs/tests/protocols/test_flatout2/index.rst @@ -0,0 +1,8 @@ +.. _test_flatout2: + +test_flatout2 +============= + +.. toctree:: + test_flatout2_get_status + test_get_status diff --git a/docs/tests/protocols/test_flatout2/test_flatout2_get_status.rst b/docs/tests/protocols/test_flatout2/test_flatout2_get_status.rst new file mode 100644 index 0000000..152b018 --- /dev/null +++ b/docs/tests/protocols/test_flatout2/test_flatout2_get_status.rst @@ -0,0 +1,16 @@ +test_flatout2_get_status +======================== + +Here are the results for the test method. + +.. code-block:: json + + { + "info": { + "hostname": "TestServer", + "timestamp": "-4022673819270859506", + "flags": "679480060", + "status": "3422879744", + "config": "c00000000000081000086124" + } + } diff --git a/docs/tests/protocols/test_flatout2/test_get_status.rst b/docs/tests/protocols/test_flatout2/test_get_status.rst new file mode 100644 index 0000000..a91a13c --- /dev/null +++ b/docs/tests/protocols/test_flatout2/test_get_status.rst @@ -0,0 +1,17 @@ +test_get_status +=============== + +Here are the results for the test method. + +.. code-block:: json + + { + "info": { + "hostname": "TestServer", + "timestamp": "1234567890", + "flags": "2818572332", + "status": "6094848", + "config": "000000000810000861240400101440" + }, + "players": [] + } \ No newline at end of file diff --git a/docs/tests/protocols/test_warcraft3/index.rst b/docs/tests/protocols/test_warcraft3/index.rst new file mode 100644 index 0000000..7bdf91e --- /dev/null +++ b/docs/tests/protocols/test_warcraft3/index.rst @@ -0,0 +1,7 @@ +.. _test_warcraft3: + +test_warcraft3 +============== + +.. toctree:: + test_warcraft3_status diff --git a/docs/tests/protocols/test_warcraft3/test_warcraft3_status.rst b/docs/tests/protocols/test_warcraft3/test_warcraft3_status.rst new file mode 100644 index 0000000..ceb0953 --- /dev/null +++ b/docs/tests/protocols/test_warcraft3/test_warcraft3_status.rst @@ -0,0 +1,24 @@ +test_warcraft3_status +===================== + +Here are the results for the test method. + +.. code-block:: json + + { + "game_version": "PX3W 26", + "hostname": "Lokales Spiel (Banane)", + "map_name": "Map name unavailable", + "game_type": "Custom Game", + "num_players": 1, + "max_players": 2, + "raw": { + "product": "PX3W", + "version": 26, + "host_counter": 6, + "entry_key": 52398541, + "settings_raw": "0103490701015501a955010fc791334d8b6171735d47736f857b656f5569736f456f655d293329555b6973697367616d6b476d616565732f477733790143616f8b616f650101272d35cd8315819ba93f8be953214553e7c513a1bd9b4b", + "game_flags": 9, + "remaining_data": "" + } + } diff --git a/opengsq/protocols/__init__.py b/opengsq/protocols/__init__.py index b9513b7..91e7083 100644 --- a/opengsq/protocols/__init__.py +++ b/opengsq/protocols/__init__.py @@ -3,6 +3,7 @@ from opengsq.protocols.doom3 import Doom3 from opengsq.protocols.eos import EOS from opengsq.protocols.fivem import FiveM +from opengsq.protocols.flatout2 import Flatout2 from opengsq.protocols.gamespy1 import GameSpy1 from opengsq.protocols.gamespy2 import GameSpy2 from opengsq.protocols.gamespy3 import GameSpy3 @@ -27,4 +28,5 @@ from opengsq.protocols.unreal2 import Unreal2 from opengsq.protocols.ut3 import UT3 from opengsq.protocols.vcmp import Vcmp +from opengsq.protocols.warcraft3 import Warcraft3 from opengsq.protocols.won import WON \ No newline at end of file diff --git a/opengsq/protocols/flatout2.py b/opengsq/protocols/flatout2.py new file mode 100644 index 0000000..e3081d1 --- /dev/null +++ b/opengsq/protocols/flatout2.py @@ -0,0 +1,168 @@ +from __future__ import annotations + +import os +from opengsq.binary_reader import BinaryReader +from opengsq.exceptions import InvalidPacketException +from opengsq.protocol_base import ProtocolBase +from opengsq.protocol_socket import UdpClient +from opengsq.responses.flatout2 import Status + + +class Flatout2(ProtocolBase): + """ + This class represents the Flatout 2 Protocol. It provides methods to interact with Flatout 2 game servers. + The protocol uses broadcast packets to discover and query servers. + """ + + full_name = "Flatout 2 Protocol" + FLATOUT2_PORT = 23757 # Default broadcast port for Flatout 2 + + # Protocol specific constants + REQUEST_HEADER = b"\x22\x00" + RESPONSE_HEADER = b"\x59\x00" + GAME_IDENTIFIER = b"FO14" + SESSION_ID = b"\x99\x72\xcc\x8f" + COMMAND_QUERY = b"\x18\x0c" + PACKET_END = b"\x2e\x55\x19\xb4\xe1\x4f\x81\x4a" + + def __init__(self, host: str, port: int = FLATOUT2_PORT, timeout: float = 5.0): + """ + Initialize the Flatout 2 protocol handler. + + :param host: The broadcast address (usually '255.255.255.255' for LAN) + :param port: The port to use (default: 23757) + :param timeout: Connection timeout in seconds + """ + if port != self.FLATOUT2_PORT: + raise ValueError(f"Flatout 2 protocol requires port {self.FLATOUT2_PORT}") + super().__init__(host, self.FLATOUT2_PORT, timeout) + self._allow_broadcast = True + + async def get_status(self) -> Status: + """ + Asynchronously retrieves the status of Flatout 2 servers via broadcast. + Expects a response packet with server information. + + :return: A Status object containing the status of the game server. + """ + # Build the request packet + request_data = ( + self.REQUEST_HEADER + # Protocol header + self.SESSION_ID + # Session ID + b"\x00" * 4 + # Padding pre-identifier + self.GAME_IDENTIFIER + # "FO14" + b"\x00" * 8 + # Padding post-identifier + self.COMMAND_QUERY + # Query command + b"\x00\x00\x22\x00" + # Command data + self.PACKET_END # Standard packet end + ) + + # Send broadcast and receive response + data = await UdpClient.communicate(self, request_data, source_port=self.FLATOUT2_PORT) + + # Debug output for packet analysis + print(f"Response header: {data[:2].hex()}, Expected: {self.RESPONSE_HEADER.hex()}") + print(f"Session ID: {data[2:6].hex()}") + print(f"Game ID: {data[10:14]}, Expected: {self.GAME_IDENTIFIER}") + print(f"Full packet length: {len(data)}") + + # Verify response packet + if not self._verify_packet(data): + raise InvalidPacketException("Invalid response packet received") + + br = BinaryReader(data) + return self._parse_response(br) + + def _verify_packet(self, data: bytes) -> bool: + """ + Verifies that a packet is a valid Flatout 2 response. + + :param data: The packet data to verify + :return: True if the packet is valid, False otherwise + """ + if len(data) < 14: # Minimum length for header + session ID + padding + game ID + print(f"Packet too short: {len(data)} bytes") + return False + + # Check response header + header_matches = data.startswith(self.RESPONSE_HEADER) + if not header_matches: + print(f"Header mismatch: got {data[:2].hex()}, expected {self.RESPONSE_HEADER.hex()}") + return False + + # Check game identifier (position 10-14, after session ID and padding) + game_id = data[10:14] + game_id_matches = game_id == self.GAME_IDENTIFIER + if not game_id_matches: + print(f"Game ID mismatch: got {game_id}, expected {self.GAME_IDENTIFIER}") + return False + + return True + + def _read_utf16_string(self, br: BinaryReader) -> str: + """ + Reads a UTF-16 encoded string from the binary reader. + + :param br: The binary reader to read from + :return: The decoded string + """ + bytes_list = [] + while True: + # Read two bytes at a time (UTF-16) + char_bytes = br.read_bytes(2) + if char_bytes == b"\x00\x00": # End of string + break + bytes_list.extend(char_bytes) + + return bytes(bytes_list).decode('utf-16-le').strip() + + def _parse_response(self, br: BinaryReader) -> Status: + """ + Parses the binary response into a Status object. + The response contains UTF-16 encoded strings and various server information. + + :param br: BinaryReader containing the response data + :return: A Status object containing the parsed information + """ + # Skip header (2), session ID (4), padding (4), game ID (4), padding (8), command (4), data (2), and packet end (8) + br.read_bytes(36) # Skip to the server data section + + info = {} + + try: + # Read server name (UTF-16 encoded) + server_name = self._read_utf16_string(br) + info["hostname"] = server_name + + # Read server information + timestamp = br.read_long_long() # Server timestamp + info["timestamp"] = str(timestamp) + + server_flags = br.read_long(unsigned=True) # Server configuration flags + info["flags"] = str(server_flags) + + # Skip reserved bytes + br.read_bytes(8) + + # Read server status + status_flags = br.read_long(unsigned=True) + info["status"] = str(status_flags) + + # Read server configuration + config = br.read_bytes(12) # Remaining configuration data + info["config"] = config.hex() + + except Exception as e: + print(f"Error parsing response: {e}") + + return Status(info=info) + + +if __name__ == "__main__": + import asyncio + + async def main_async(): + # Use broadcast address for LAN discovery + flatout2 = Flatout2(host="255.255.255.255", port=23757, timeout=5.0) + status = await flatout2.get_status() + print(status) \ No newline at end of file diff --git a/opengsq/protocols/warcraft3.py b/opengsq/protocols/warcraft3.py new file mode 100644 index 0000000..3e6886c --- /dev/null +++ b/opengsq/protocols/warcraft3.py @@ -0,0 +1,213 @@ +from __future__ import annotations + +from opengsq.protocol_base import ProtocolBase +from opengsq.protocol_socket import UdpClient +from opengsq.binary_reader import BinaryReader +from opengsq.responses.warcraft3 import Status +from enum import IntFlag, IntEnum + +class GameFlags(IntFlag): + """Game flags based on the Go implementation""" + CUSTOM_GAME = 0x000001 + SINGLE_PLAYER = 0x000005 + LADDER_1V1 = 0x000010 + LADDER_2V2 = 0x000020 + LADDER_3V3 = 0x000040 + LADDER_4V4 = 0x000080 + TEAM_LADDER = 0x000020 + SAVED_GAME = 0x000200 + TYPE_MASK = 0x0002F5 + SIGNED_MAP = 0x000008 + PRIVATE_GAME = 0x000800 + CREATOR_USER = 0x002000 + CREATOR_BLIZZARD = 0x004000 + CREATOR_MASK = 0x006000 + MAP_TYPE_MELEE = 0x008000 + MAP_TYPE_SCENARIO = 0x010000 + MAP_TYPE_MASK = 0x018000 + SIZE_SMALL = 0x020000 + SIZE_MEDIUM = 0x040000 + SIZE_LARGE = 0x080000 + SIZE_MASK = 0x0E0000 + OBS_FULL = 0x100000 + OBS_ON_DEFEAT = 0x200000 + OBS_NONE = 0x400000 + OBS_MASK = 0x700000 + FILTER_MASK = 0x7FE000 + +class GameSettingFlags(IntFlag): + """Game setting flags based on the Go implementation""" + SPEED_SLOW = 0x00000000 + SPEED_NORMAL = 0x00000001 + SPEED_FAST = 0x00000002 + SPEED_MASK = 0x0000000F + TERRAIN_HIDDEN = 0x00000100 + TERRAIN_EXPLORED = 0x00000200 + TERRAIN_VISIBLE = 0x00000400 + TERRAIN_DEFAULT = 0x00000800 + TERRAIN_MASK = 0x00000F00 + OBS_NONE = 0x00000000 + OBS_ENABLED = 0x00001000 + OBS_ON_DEFEAT = 0x00002000 + OBS_FULL = 0x00003000 + OBS_REFEREES = 0x40000000 + OBS_MASK = 0x40003000 + TEAMS_TOGETHER = 0x00004000 + TEAMS_FIXED = 0x00060000 + SHARED_CONTROL = 0x01000000 + RANDOM_HERO = 0x02000000 + RANDOM_RACE = 0x04000000 + +class SlotStatus(IntEnum): + """Slot status based on the Go implementation""" + OPEN = 0x00 + CLOSED = 0x01 + OCCUPIED = 0x02 + +class Warcraft3(ProtocolBase): + """ + This class represents the Warcraft 3 Protocol. It provides methods to interact with Warcraft 3 game servers. + Based on the gowarcraft3 implementation. + """ + + full_name = "Warcraft 3 Protocol" + WARCRAFT3_PORT = 6112 # Default port for Warcraft 3 + PROTOCOL_SIG = 0xF7 # Protocol signature + CURRENT_GAME_VERSION = 26 # Current game version (changed from 10032 to 26) + + # Packet types + PID_SEARCH_GAME = 0x2F + PID_GAME_INFO = 0x30 + + def __init__(self, host: str, port: int = WARCRAFT3_PORT, timeout: float = 5.0): + """ + Initialize the Warcraft 3 protocol handler. + + :param host: The server address + :param port: The port to use (default: 6112) + :param timeout: Connection timeout in seconds + """ + if port != self.WARCRAFT3_PORT: + raise ValueError(f"Warcraft 3 protocol requires port {self.WARCRAFT3_PORT}") + super().__init__(host, self.WARCRAFT3_PORT, timeout) + + def _create_search_game_packet(self) -> bytes: + """Creates a search game packet based on the Go implementation""" + packet = bytearray() + packet.extend([ + self.PROTOCOL_SIG, # Protocol signature + self.PID_SEARCH_GAME, # Packet type + 0x10, 0x00, # Packet size (16 bytes) + 0x50, 0x58, 0x33, 0x57, # Product "PX3W" (reversed "W3XP") + # Game version (little endian) + self.CURRENT_GAME_VERSION & 0xFF, + (self.CURRENT_GAME_VERSION >> 8) & 0xFF, + (self.CURRENT_GAME_VERSION >> 16) & 0xFF, + (self.CURRENT_GAME_VERSION >> 24) & 0xFF, + # Host counter + 0x00, 0x00, 0x00, 0x00 + ]) + return bytes(packet) + + async def get_status(self) -> Status: + """ + Asynchronously retrieves the status of the game server. + + :return: A Status object containing the status of the game server. + """ + request = self._create_search_game_packet() + response = await UdpClient.communicate(self, request, source_port=self.WARCRAFT3_PORT) + + if not response or len(response) < 4: + raise Exception("Invalid response") + + br = BinaryReader(response) + + # Validate protocol signature + if int.from_bytes(br.read_bytes(1), 'little') != self.PROTOCOL_SIG: + raise Exception("Invalid protocol signature") + + # Validate packet type + if int.from_bytes(br.read_bytes(1), 'little') != self.PID_GAME_INFO: + raise Exception("Invalid response packet type") + + # Read packet size + packet_size = int.from_bytes(br.read_bytes(2), 'little') + if len(response) < packet_size: + raise Exception("Incomplete response") + + # Read game version info + product = br.read_bytes(4).decode('ascii') # Product code (WAR3/W3XP) + version = int.from_bytes(br.read_bytes(4), 'little') # Game version + host_counter = int.from_bytes(br.read_bytes(4), 'little') # Host counter + entry_key = int.from_bytes(br.read_bytes(4), 'little') # Entry key + + # Read game name (null-terminated) + game_name = "" + while True: + char = int.from_bytes(br.read_bytes(1), 'little') + if char == 0: + break + game_name += chr(char) + + # Skip unknown byte + br.read_bytes(1) + + # Read game settings string (null-terminated) + settings_raw = bytearray() + while True: + byte = int.from_bytes(br.read_bytes(1), 'little') + if byte == 0: + break + settings_raw.append(byte) + + # Read remaining fields + slots_total = int.from_bytes(br.read_bytes(4), 'little') + game_flags = GameFlags(int.from_bytes(br.read_bytes(4), 'little')) + slots_used = int.from_bytes(br.read_bytes(4), 'little') + slots_available = int.from_bytes(br.read_bytes(4), 'little') + uptime_seconds = int.from_bytes(br.read_bytes(4), 'little') + port = int.from_bytes(br.read_bytes(2), 'little') + + # Store raw data for debugging + raw = { + 'product': product, + 'version': version, + 'host_counter': host_counter, + 'entry_key': entry_key, + 'settings_raw': settings_raw.hex(), + 'game_flags': game_flags, + 'remaining_data': br.read_bytes(br.remaining_bytes()).hex() + } + + return Status( + game_version=f"{product} {version}", + hostname=game_name, + map_name=self._get_map_name_from_settings(settings_raw), + game_type=self._get_game_type(game_flags), + num_players=slots_used, + max_players=slots_total, + raw=raw + ) + + def _get_map_name_from_settings(self, settings_raw: bytearray) -> str: + """Map name parsing is skipped due to encoding complexity""" + return "Map name unavailable" + + def _get_game_type(self, flags: GameFlags) -> str: + """Convert game flags to a readable game type""" + if flags & GameFlags.CUSTOM_GAME: + return "Custom Game" + elif flags & GameFlags.SINGLE_PLAYER: + return "Single Player" + elif flags & GameFlags.LADDER_1V1: + return "Ladder 1v1" + elif flags & GameFlags.LADDER_2V2: + return "Ladder 2v2" + elif flags & GameFlags.LADDER_3V3: + return "Ladder 3v3" + elif flags & GameFlags.LADDER_4V4: + return "Ladder 4v4" + elif flags & GameFlags.SAVED_GAME: + return "Saved Game" + return "Unknown" \ No newline at end of file diff --git a/opengsq/responses/flatout2.py b/opengsq/responses/flatout2.py new file mode 100644 index 0000000..ea62937 --- /dev/null +++ b/opengsq/responses/flatout2.py @@ -0,0 +1,27 @@ +from dataclasses import dataclass +from typing import Dict, List, Any + + +@dataclass +class Status: + """ + Represents the status response from a Flatout 2 server. + Contains server information and player data. + """ + info: Dict[str, Any] + players: List[Dict[str, Any]] + + def __str__(self) -> str: + """ + Returns a human-readable string representation of the server status. + """ + output = ["Server Information:"] + for key, value in self.info.items(): + output.append(f"{key}: {value}") + + if self.players: + output.append("\nPlayers:") + for player in self.players: + output.append(f"{player['name']} - Score: {player['score']}") + + return "\n".join(output) \ No newline at end of file diff --git a/opengsq/responses/flatout2/__init__.py b/opengsq/responses/flatout2/__init__.py new file mode 100644 index 0000000..f4d7d9c --- /dev/null +++ b/opengsq/responses/flatout2/__init__.py @@ -0,0 +1 @@ +from .status import Status \ No newline at end of file diff --git a/opengsq/responses/flatout2/status.py b/opengsq/responses/flatout2/status.py new file mode 100644 index 0000000..9279149 --- /dev/null +++ b/opengsq/responses/flatout2/status.py @@ -0,0 +1,36 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Dict + + +@dataclass +class Status: + """ + Represents the status of a Flatout 2 server. + """ + + info: Dict[str, str] + """ + Server information dictionary containing: + - hostname: Server name (UTF-16 encoded) + - timestamp: Server timestamp + - flags: Server configuration flags + - status: Server status flags + - config: Additional configuration data in hex format + """ + + def __str__(self) -> str: + """ + Returns a human-readable string representation of the server status. + + :return: Formatted server status string + """ + result = [] + + # Add server info + result.append("Server Information:") + for key, value in self.info.items(): + result.append(f" {key}: {value}") + + return "\n".join(result) \ No newline at end of file diff --git a/opengsq/responses/warcraft3/__init__.py b/opengsq/responses/warcraft3/__init__.py new file mode 100644 index 0000000..77a37e0 --- /dev/null +++ b/opengsq/responses/warcraft3/__init__.py @@ -0,0 +1,3 @@ +from opengsq.responses.warcraft3.status import Status + +__all__ = ["Status"] \ No newline at end of file diff --git a/opengsq/responses/warcraft3/status.py b/opengsq/responses/warcraft3/status.py new file mode 100644 index 0000000..45ef9ce --- /dev/null +++ b/opengsq/responses/warcraft3/status.py @@ -0,0 +1,15 @@ +from dataclasses import dataclass +from typing import Dict, Any + +@dataclass +class Status: + """ + Represents the status of a Warcraft 3 game server. + """ + game_version: str + hostname: str + map_name: str + game_type: str + num_players: int + max_players: int + raw: Dict[str, Any] \ No newline at end of file diff --git a/tests/protocols/test_flatout2.py b/tests/protocols/test_flatout2.py new file mode 100644 index 0000000..748b64c --- /dev/null +++ b/tests/protocols/test_flatout2.py @@ -0,0 +1,146 @@ +import pytest +from unittest.mock import AsyncMock, patch + +from opengsq.protocols.flatout2 import Flatout2 +from opengsq.exceptions import InvalidPacketException +from ..result_handler import ResultHandler + +handler = ResultHandler(__file__) +handler.enable_save = True + +@pytest.mark.asyncio +async def test_flatout2_get_status(): + """Test successful server status query via broadcast""" + # Example response packet based on actual server response + response_data = bytes.fromhex( + "5900" # Response header + "9972cc8f" # Session ID + "00000000" # Padding pre-identifier + "464f3134" # "FO14" Game Identifier + "0000000000000000" # Padding post-identifier + "19437312" # Command + "0000" # Additional data + "2e5519b4e14f814a" # Packet end marker + "47006f006d00620069000000" # Server name "Gombi" in UTF-16 + "0eb518737997" # Timestamp + "2cc8fc0a80281000" # Server flags + "0000000000000000" # Reserved bytes + "05ccc000" # Status flags + "000000000810000861240400101440" # Configuration data + ) + + print(f"\nTest packet structure:") + print(f"Header: {response_data[:2].hex()}") + print(f"Session ID: {response_data[2:6].hex()}") + print(f"Game ID: {response_data[10:14]}") + print(f"Full packet: {response_data.hex()}") + + # Mock the UdpClient.communicate method + mock_communicate = AsyncMock(return_value=response_data) + + with patch("opengsq.protocol_socket.UdpClient.communicate", mock_communicate): + flatout2 = Flatout2(host="255.255.255.255") + status = await flatout2.get_status() + + # Verify the mock was called with broadcast parameters + mock_communicate.assert_called_once() + call_args = mock_communicate.call_args + assert call_args is not None + args, kwargs = call_args + assert kwargs.get('source_port') == Flatout2.FLATOUT2_PORT + + # Verify server info + assert "hostname" in status.info + assert status.info["hostname"] # Check that hostname is not empty + assert "timestamp" in status.info + assert "flags" in status.info + assert "status" in status.info + assert "config" in status.info + + # Save the result for documentation + await handler.save_result("test_flatout2_get_status", status) + + +@pytest.mark.asyncio +async def test_flatout2_invalid_header(): + """Test handling of invalid packet header""" + # Response with invalid header (0x22 instead of 0x59) + invalid_response = bytes.fromhex( + "2200" # Wrong header + "9972cc8f" # Session ID + "00000000" # Padding pre-identifier + "464f3134" # "FO14" Game Identifier + "0000000000000000" # Padding post-identifier + "19437312" # Command + "0000" # Additional data + "2e5519b4e14f814a" # Packet end marker + "47006f006d00620069000000" # Server name "Gombi" in UTF-16 + "0eb518737997" # Rest of the data... + "2cc8fc0a802810000000000000000000000005ccc00000000000" + "081000086124040010144000" + ) + + # Mock the UdpClient.communicate method + mock_communicate = AsyncMock(return_value=invalid_response) + + with patch("opengsq.protocol_socket.UdpClient.communicate", mock_communicate): + flatout2 = Flatout2(host="255.255.255.255") + with pytest.raises(InvalidPacketException): + await flatout2.get_status() + + +@pytest.mark.asyncio +async def test_flatout2_invalid_game_id(): + """Test handling of invalid game identifier""" + # Response with wrong game ID (BADID instead of FO14) + invalid_game_id = bytes.fromhex( + "5900" # Response header + "9972cc8f" # Session ID + "00000000" # Padding pre-identifier + "42414449" # Wrong game ID "BADI" + "0000000000000000" # Padding post-identifier + "19437312" # Command + "0000" # Additional data + "2e5519b4e14f814a" # Packet end marker + "47006f006d00620069000000" # Server name "Gombi" in UTF-16 + "0eb518737997" # Rest of the data... + "2cc8fc0a802810000000000000000000000005ccc00000000000" + "081000086124040010144000" + ) + + # Mock the UdpClient.communicate method + mock_communicate = AsyncMock(return_value=invalid_game_id) + + with patch("opengsq.protocol_socket.UdpClient.communicate", mock_communicate): + flatout2 = Flatout2(host="255.255.255.255") + with pytest.raises(InvalidPacketException): + await flatout2.get_status() + + +@pytest.mark.asyncio +async def test_flatout2_wrong_port(): + """Test handling of wrong port number""" + with pytest.raises(ValueError) as exc_info: + Flatout2(host="255.255.255.255", port=27015) # Wrong port + assert str(exc_info.value) == f"Flatout 2 protocol requires port {Flatout2.FLATOUT2_PORT}" + + +def test_status_string_representation(): + """Test the string representation of the Status class""" + from opengsq.responses.flatout2 import Status + + # Create a test status object + info = { + "hostname": "Test Server", + "timestamp": "123456789", + "flags": "1234", + "status": "5678", + "config": "abcdef" + } + + status = Status(info=info) + str_repr = str(status) + + # Verify the string representation + assert "Server Information:" in str_repr + assert "hostname:" in str_repr # Just check that the hostname field exists \ No newline at end of file diff --git a/tests/protocols/test_warcraft3.py b/tests/protocols/test_warcraft3.py new file mode 100644 index 0000000..c354480 --- /dev/null +++ b/tests/protocols/test_warcraft3.py @@ -0,0 +1,26 @@ +import pytest +from opengsq.protocols.warcraft3 import Warcraft3 +from ..result_handler import ResultHandler + +handler = ResultHandler(__file__) +handler.enable_save = True + +@pytest.mark.asyncio +async def test_warcraft3_status(): + warcraft3 = Warcraft3(host="10.10.101.4", port=6112) # Replace with your test server + result = await warcraft3.get_status() + + print("\nWarcraft 3 Server Details:") + print(f"Server Name: {result.hostname}") + print(f"Game Version: {result.game_version}") + print(f"Map: {result.map_name}") + print(f"Game Type: {result.game_type}") + print(f"Players: {result.num_players}/{result.max_players}") + + # Example raw data that might be useful for debugging + print("\nRaw Data:") + if hasattr(result, 'raw'): + for key, value in result.raw.items(): + print(f"{key}: {value}") + + await handler.save_result("test_warcraft3_status", result) \ No newline at end of file