"""Tests for partial update splitting and reassembly.""" import pytest from src.shared.models import GameState, Snake, Food, Position from src.server.partial_update import PartialUpdateEncoder from src.client.partial_state_tracker import PartialStateTracker from src.shared.binary_codec import BinaryCodec class TestPartialUpdateSplitting: """Test splitting game state into partial updates.""" def test_small_state_single_packet(self): """Test small state fits in one packet.""" # Create small game state state = GameState( snakes=[ Snake( player_id="player1", body=[Position(5, 5), Position(6, 5), Position(7, 5)], color_index=0, player_name="Alice" ) ], food=[Food(position=Position(10, 10))], game_running=True ) encoder = PartialUpdateEncoder() packets = encoder.split_state_update(state, update_id=1, max_packet_size=1280) # Should have metadata + one snake packet assert len(packets) >= 2 def test_many_snakes_multiple_packets(self): """Test many snakes split into multiple packets.""" # Create state with many snakes snakes = [] for i in range(32): snake = Snake( player_id=f"player{i}", body=[Position(i, j) for j in range(10)], # 10-segment snake color_index=i % 32, player_name=f"Player{i}" ) snakes.append(snake) state = GameState( snakes=snakes, food=[Food(position=Position(15, 15))], game_running=True ) encoder = PartialUpdateEncoder() packets = encoder.split_state_update(state, update_id=100, max_packet_size=1280) # Should have at least metadata packet + snake packet assert len(packets) >= 2 # All packets should be under size limit for packet in packets: assert len(packet) < 1280 def test_very_long_snake_splitting(self): """Test very long snake is split into segments.""" # Create snake with 500 segments body = [Position(i % 40, i // 40) for i in range(500)] snake = Snake( player_id="long_player", body=body, color_index=0, player_name="LongSnake" ) state = GameState( snakes=[snake], food=[], game_running=True ) encoder = PartialUpdateEncoder() packets = encoder.split_state_update(state, update_id=50, max_packet_size=1280) # Should have metadata + at least one snake packet assert len(packets) >= 2 # All packets under limit for packet in packets: assert len(packet) < 1280 def test_name_caching(self): """Test player name is only sent once.""" snake = Snake( player_id="player1", body=[Position(5, 5), Position(6, 5)], color_index=0, player_name="Alice" ) state = GameState(snakes=[snake], food=[], game_running=True) encoder = PartialUpdateEncoder() # First update - should include name packets1 = encoder.split_state_update(state, update_id=1) # Second update - name should be cached packets2 = encoder.split_state_update(state, update_id=2) # Second update packets should be smaller (no name) total_size1 = sum(len(p) for p in packets1) total_size2 = sum(len(p) for p in packets2) assert total_size2 <= total_size1 class TestPartialStateReassembly: """Test reassembling partial updates on client.""" def test_single_packet_reassembly(self): """Test reassembling single packet.""" # Create and encode state state = GameState( snakes=[ Snake( player_id="player1", body=[Position(5, 5), Position(6, 5)], color_index=0, player_name="Alice", direction=(1, 0), alive=True ) ], food=[Food(position=Position(10, 10))], game_running=True ) encoder = PartialUpdateEncoder() packets = encoder.split_state_update(state, update_id=1) # Reassemble tracker = PartialStateTracker() for packet in packets: tracker.process_packet(1, packet) reassembled = tracker.get_game_state() # Verify assert reassembled.game_running == True assert len(reassembled.snakes) >= 1 assert len(reassembled.food) == 1 def test_multiple_packet_reassembly(self): """Test reassembling from multiple packets.""" # Create state with multiple snakes snakes = [ Snake( player_id=f"player{i}", body=[Position(i, j) for j in range(5)], color_index=i, player_name=f"Player{i}" ) for i in range(10) ] state = GameState(snakes=snakes, food=[], game_running=True) encoder = PartialUpdateEncoder() packets = encoder.split_state_update(state, update_id=10) # Reassemble tracker = PartialStateTracker() for packet in packets: tracker.process_packet(10, packet) reassembled = tracker.get_game_state() # Should have all snakes assert len(reassembled.snakes) >= len(snakes) def test_packet_loss_resilience(self): """Test handling of lost packets.""" # Create state snakes = [ Snake( player_id=f"player{i}", body=[Position(i, j) for j in range(5)], color_index=i, player_name=f"Player{i}" ) for i in range(10) ] state = GameState(snakes=snakes, food=[], game_running=True) encoder = PartialUpdateEncoder() packets = encoder.split_state_update(state, update_id=20) # Simulate packet loss - skip middle packet if len(packets) > 2: lost_packet_idx = len(packets) // 2 packets_received = packets[:lost_packet_idx] + packets[lost_packet_idx + 1:] else: packets_received = packets # Reassemble tracker = PartialStateTracker() for packet in packets_received: tracker.process_packet(20, packet) reassembled = tracker.get_game_state() # Should have partial state (some snakes) assert len(reassembled.snakes) > 0 # But not all (due to loss) if len(packets) > 2: assert len(reassembled.snakes) < len(snakes) def test_name_caching_on_client(self): """Test client caches player names.""" snake = Snake( player_id="player1", body=[Position(5, 5)], color_index=0, player_name="Alice" ) state1 = GameState(snakes=[snake], food=[], game_running=True) encoder = PartialUpdateEncoder() packets1 = encoder.split_state_update(state1, update_id=1) # Process first update tracker = PartialStateTracker() for packet in packets1: tracker.process_packet(1, packet) result1 = tracker.get_game_state() assert result1.snakes[0].player_name == "Alice" # Second update without name state2 = GameState(snakes=[snake], food=[], game_running=True) packets2 = encoder.split_state_update(state2, update_id=2) # Process second update for packet in packets2: tracker.process_packet(2, packet) result2 = tracker.get_game_state() # Name should still be available from cache player_hash = BinaryCodec.player_id_hash("player1") assert player_hash in tracker.player_name_cache assert tracker.player_name_cache[player_hash] == "Alice" def test_update_id_transition(self): """Test transitioning between update IDs.""" snake1 = Snake(player_id="p1", body=[Position(1, 1)], color_index=0) snake2 = Snake(player_id="p2", body=[Position(2, 2)], color_index=1) state1 = GameState(snakes=[snake1], food=[], game_running=True) state2 = GameState(snakes=[snake2], food=[], game_running=True) encoder = PartialUpdateEncoder() # Encode both states packets1 = encoder.split_state_update(state1, update_id=1) packets2 = encoder.split_state_update(state2, update_id=2) # Process tracker = PartialStateTracker() for packet in packets1: tracker.process_packet(1, packet) result1 = tracker.get_game_state() for packet in packets2: tracker.process_packet(2, packet) result2 = tracker.get_game_state() # Should have transitioned to new update assert tracker.current_update_id == 2 class TestPacketSizeConstraints: """Test packet size constraints.""" def test_all_packets_under_mtu(self): """Test all packets respect MTU limit.""" # Create maximum state snakes = [ Snake( player_id=f"player{i}", body=[Position((i + j) % 40, j % 30) for j in range(20)], color_index=i % 32, player_name=f"VeryLongName{i:04d}" ) for i in range(32) ] state = GameState( snakes=snakes, food=[Food(position=Position(i, i)) for i in range(10)], game_running=True ) encoder = PartialUpdateEncoder() packets = encoder.split_state_update(state, update_id=999, max_packet_size=1280) # All packets must be under MTU for i, packet in enumerate(packets): assert len(packet) < 1280, f"Packet {i} exceeds MTU: {len(packet)} bytes" def test_compression_benefit(self): """Test compression reduces packet size.""" # Create repetitive state (compresses well) snake = Snake( player_id="player1", body=[Position(5, i) for i in range(100)], # Straight line color_index=0, player_name="Test" ) state = GameState(snakes=[snake], food=[], game_running=True) encoder = PartialUpdateEncoder() packets = encoder.split_state_update(state, update_id=1) # Packets should benefit from compression # Delta encoding + compression should keep size reasonable for packet in packets: # Uncompressed would be ~200 bytes for 100 positions # With delta + compression should be much smaller assert len(packet) < 150 if __name__ == "__main__": pytest.main([__file__, "-v"])