Implemented a complete network multiplayer Snake game with the following features: Core Game: - Client-server architecture using asyncio for networking - Pygame-based rendering at 60 FPS - Server-authoritative game state with 10 TPS - Collision detection (walls, self, other players) - Food spawning and score tracking - Support for multiple players with color-coded snakes Server Discovery: - UDP multicast-based automatic server discovery (239.255.0.1:9999) - Server beacon broadcasts presence every 2 seconds - Client discovery with 3-second timeout - Server selection UI for multiple servers - Auto-connect for single server - Graceful fallback to manual connection Project Structure: - src/shared/ - Protocol, models, constants, discovery utilities - src/server/ - Game server, game logic, server beacon - src/client/ - Game client, renderer, discovery, server selector - tests/ - Unit tests for game logic, models, and discovery Command-line interface with argparse for both server and client. Comprehensive documentation in README.md and CLAUDE.md. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
160 lines
4.9 KiB
Python
160 lines
4.9 KiB
Python
"""Tests for server discovery functionality."""
|
|
|
|
import pytest
|
|
from src.shared.discovery import (
|
|
ServerInfo,
|
|
DiscoveryMessage,
|
|
create_multicast_socket,
|
|
get_local_ip,
|
|
)
|
|
|
|
|
|
class TestServerInfo:
|
|
"""Test suite for ServerInfo class."""
|
|
|
|
def test_server_info_creation(self) -> None:
|
|
"""Test creating ServerInfo."""
|
|
info = ServerInfo(
|
|
host="192.168.1.100",
|
|
port=8888,
|
|
server_name="Test Server",
|
|
players_count=3,
|
|
)
|
|
|
|
assert info.host == "192.168.1.100"
|
|
assert info.port == 8888
|
|
assert info.server_name == "Test Server"
|
|
assert info.players_count == 3
|
|
|
|
def test_server_info_serialization(self) -> None:
|
|
"""Test ServerInfo to_dict and from_dict."""
|
|
info = ServerInfo(
|
|
host="10.0.0.1",
|
|
port=9999,
|
|
server_name="Game Room",
|
|
players_count=2,
|
|
)
|
|
|
|
# Serialize
|
|
data = info.to_dict()
|
|
assert data["host"] == "10.0.0.1"
|
|
assert data["port"] == 9999
|
|
assert data["server_name"] == "Game Room"
|
|
assert data["players_count"] == 2
|
|
|
|
# Deserialize
|
|
info2 = ServerInfo.from_dict(data)
|
|
assert info2.host == info.host
|
|
assert info2.port == info.port
|
|
assert info2.server_name == info.server_name
|
|
assert info2.players_count == info.players_count
|
|
|
|
def test_server_info_host_override(self) -> None:
|
|
"""Test ServerInfo from_dict with host override."""
|
|
data = {
|
|
"host": "1.2.3.4",
|
|
"port": 8888,
|
|
"server_name": "Test",
|
|
"players_count": 1,
|
|
}
|
|
|
|
# Override host
|
|
info = ServerInfo.from_dict(data, host="5.6.7.8")
|
|
assert info.host == "5.6.7.8"
|
|
assert info.port == 8888
|
|
|
|
|
|
class TestDiscoveryMessage:
|
|
"""Test suite for DiscoveryMessage class."""
|
|
|
|
def test_create_discover_message(self) -> None:
|
|
"""Test creating DISCOVER message."""
|
|
msg = DiscoveryMessage.create_discover()
|
|
assert isinstance(msg, bytes)
|
|
|
|
# Parse it back
|
|
msg_type, msg_data = DiscoveryMessage.parse(msg)
|
|
assert msg_type == DiscoveryMessage.DISCOVER
|
|
assert msg_data is None or msg_data == {}
|
|
|
|
def test_create_announce_message(self) -> None:
|
|
"""Test creating SERVER_ANNOUNCE message."""
|
|
server_info = ServerInfo(
|
|
host="192.168.1.1",
|
|
port=8888,
|
|
server_name="Test Server",
|
|
players_count=5,
|
|
)
|
|
|
|
msg = DiscoveryMessage.create_announce(server_info)
|
|
assert isinstance(msg, bytes)
|
|
|
|
# Parse it back
|
|
msg_type, msg_data = DiscoveryMessage.parse(msg)
|
|
assert msg_type == DiscoveryMessage.SERVER_ANNOUNCE
|
|
assert msg_data is not None
|
|
assert msg_data["host"] == "192.168.1.1"
|
|
assert msg_data["port"] == 8888
|
|
assert msg_data["server_name"] == "Test Server"
|
|
assert msg_data["players_count"] == 5
|
|
|
|
def test_parse_invalid_message(self) -> None:
|
|
"""Test parsing invalid message."""
|
|
msg_type, msg_data = DiscoveryMessage.parse(b"invalid json")
|
|
assert msg_type is None
|
|
assert msg_data is None
|
|
|
|
def test_message_round_trip(self) -> None:
|
|
"""Test encoding and decoding messages."""
|
|
# Test DISCOVER
|
|
discover = DiscoveryMessage.create_discover()
|
|
msg_type, _ = DiscoveryMessage.parse(discover)
|
|
assert msg_type == DiscoveryMessage.DISCOVER
|
|
|
|
# Test ANNOUNCE
|
|
info = ServerInfo(
|
|
host="10.0.0.1",
|
|
port=7777,
|
|
server_name="Round Trip",
|
|
players_count=0,
|
|
)
|
|
announce = DiscoveryMessage.create_announce(info)
|
|
msg_type, msg_data = DiscoveryMessage.parse(announce)
|
|
assert msg_type == DiscoveryMessage.SERVER_ANNOUNCE
|
|
assert msg_data["server_name"] == "Round Trip"
|
|
|
|
|
|
class TestDiscoveryUtilities:
|
|
"""Test suite for discovery utility functions."""
|
|
|
|
def test_create_multicast_socket(self) -> None:
|
|
"""Test creating multicast socket."""
|
|
# Test without binding
|
|
sock = create_multicast_socket(bind=False)
|
|
assert sock is not None
|
|
sock.close()
|
|
|
|
# Test with binding (may fail in some environments)
|
|
try:
|
|
sock = create_multicast_socket(bind=True)
|
|
assert sock is not None
|
|
sock.close()
|
|
except Exception:
|
|
# Binding may fail in restricted environments
|
|
pytest.skip("Cannot bind to multicast group in this environment")
|
|
|
|
def test_get_local_ip(self) -> None:
|
|
"""Test getting local IP address."""
|
|
ip = get_local_ip()
|
|
assert isinstance(ip, str)
|
|
assert len(ip) > 0
|
|
|
|
# Should be a valid IP format
|
|
parts = ip.split(".")
|
|
assert len(parts) == 4
|
|
|
|
# Each part should be a number 0-255
|
|
for part in parts:
|
|
num = int(part)
|
|
assert 0 <= num <= 255
|