Game Testing

Testing Multiplayer Games: Network Simulation, Latency, and Concurrency

HelpMeTest

HelpMeTest

— 6 min read
Testing Multiplayer Games: Network Simulation, Latency, and Concurrency

Multiplayer game testing is categorically harder than single-player. You have multiple clients maintaining synchronized state over unreliable networks, with varying latency and packet loss. A feature that works perfectly in a local LAN test can break under real internet conditions.

The Multiplayer Testing Stack

Multiplayer testing has multiple layers:

  1. Unit — Test game logic (damage calculation, inventory management) in isolation
  2. Network simulation — Inject latency, packet loss, and jitter to test client resilience
  3. State synchronization — Verify all clients agree on game state
  4. Concurrency — Test behaviors when multiple players act simultaneously
  5. Load — Test server under many concurrent connections
  6. Security — Test anti-cheat and input validation

Network Simulation

Using tc netem

Linux's traffic control is the most reliable way to simulate network conditions in tests:

#!/bin/bash
<span class="hljs-comment"># simulate-network.sh

INTERFACE=<span class="hljs-variable">${1:-lo}
PROFILE=<span class="hljs-variable">${2:-"bad-3g"}

<span class="hljs-function">reset_network() {
    tc qdisc del dev <span class="hljs-variable">$INTERFACE root 2>/dev/null <span class="hljs-pipe">|| <span class="hljs-literal">true
}

<span class="hljs-keyword">case <span class="hljs-variable">$PROFILE <span class="hljs-keyword">in
  <span class="hljs-string">"perfect")
    reset_network
    <span class="hljs-pipe">;;
  <span class="hljs-string">"good-broadband")
    tc qdisc add dev <span class="hljs-variable">$INTERFACE root netem delay 20ms 5ms loss 0.1%
    <span class="hljs-pipe">;;
  <span class="hljs-string">"bad-3g")
    tc qdisc add dev <span class="hljs-variable">$INTERFACE root netem delay 150ms 50ms loss 2% corrupt 0.5%
    <span class="hljs-pipe">;;
  <span class="hljs-string">"satellite")
    tc qdisc add dev <span class="hljs-variable">$INTERFACE root netem delay 600ms 50ms loss 1%
    <span class="hljs-pipe">;;
  <span class="hljs-string">"high-jitter")
    <span class="hljs-comment"># Realistic home WiFi with interference
    tc qdisc add dev <span class="hljs-variable">$INTERFACE root netem delay 50ms 40ms distribution pareto loss 3%
    <span class="hljs-pipe">;;
  <span class="hljs-string">"packet-burst-loss")
    <span class="hljs-comment"># Simulate burst packet loss (common on mobile)
    tc qdisc add dev <span class="hljs-variable">$INTERFACE root netem loss gemodel 1% 10% 50% 5%
    <span class="hljs-pipe">;;
<span class="hljs-keyword">esac

<span class="hljs-built_in">echo <span class="hljs-string">"Applied '$PROFILE' network profile to <span class="hljs-variable">$INTERFACE"

Python Network Simulation Tests

import subprocess
import contextlib
import time
import socket
import pytest

@contextlib.contextmanager
def network_conditions(latency_ms: int = 0, loss_pct: float = 0, 
                        jitter_ms: int = 0, interface: str = "lo"):
    """Apply network conditions for the duration of a test."""
    conditions = []
    if latency_ms > 0:
        conditions.append(f"delay {latency_ms}ms {jitter_ms}ms")
    if loss_pct > 0:
        conditions.append(f"loss {loss_pct}%")
    
    if conditions:
        cmd = f"tc qdisc add dev {interface} root netem {' '.join(conditions)}"
        subprocess.run(cmd, shell=True, check=True)
    
    try:
        yield
    finally:
        subprocess.run(f"tc qdisc del dev {interface} root", shell=True)

@pytest.mark.network
def test_client_reconnects_after_packet_loss():
    """Client should automatically reconnect within 30 seconds under 10% packet loss."""
    game_client = GameClient(server_host="localhost", server_port=7777)
    game_client.connect()
    
    player_id = game_client.join_match()
    assert player_id is not None
    
    with network_conditions(loss_pct=10.0, latency_ms=50):
        # Connection should remain stable (might lag, but not disconnect)
        time.sleep(30)
        
        # Send a few game actions
        for _ in range(5):
            game_client.send_action("move", direction="forward")
            time.sleep(1)
        
        connection_state = game_client.get_connection_state()
        assert connection_state in ("connected", "reconnecting"), \
            f"Client entered unexpected state: {connection_state}"
    
    # After packet loss ends, should be fully connected
    time.sleep(5)
    assert game_client.is_connected(), "Client did not recover after packet loss"
    game_client.disconnect()

@pytest.mark.network
def test_gameplay_playable_at_150ms_latency():
    """Game should be playable (no timeout errors) at 150ms latency."""
    game_client = GameClient(server_host="localhost", server_port=7777)
    
    with network_conditions(latency_ms=150, jitter_ms=30):
        game_client.connect()
        game_client.join_match()
        
        errors = []
        for i in range(20):
            try:
                result = game_client.send_action_and_wait("attack", timeout=5.0)
                assert result["acknowledged"]
            except TimeoutError as e:
                errors.append(f"Action {i}: timeout")
            time.sleep(0.5)
        
        assert len(errors) == 0, \
            f"Actions timed out at 150ms latency: {errors}"

State Synchronization Testing

All clients must agree on game state. Divergence causes cheating opportunities and "rubber-banding" bugs.

class MultiplayerStateTestHarness:
    def __init__(self, server_host: str, num_clients: int):
        self.clients = [
            GameClient(server_host, f"player-{i}") 
            for i in range(num_clients)
        ]
    
    def connect_all(self):
        for client in self.clients:
            client.connect()
    
    def get_all_states(self) -> list:
        """Get game state snapshot from each client."""
        return [client.get_game_state() for client in self.clients]
    
    def assert_states_synchronized(self, tolerance: float = 0.001):
        """Verify all clients report the same game state."""
        states = self.get_all_states()
        
        reference = states[0]
        for i, state in enumerate(states[1:], start=1):
            # Check all entities
            for entity_id, ref_entity in reference["entities"].items():
                if entity_id not in state["entities"]:
                    raise AssertionError(f"Client {i} missing entity {entity_id}")
                
                client_entity = state["entities"][entity_id]
                
                # Position should match within tolerance (floating point)
                for axis in ["x", "y", "z"]:
                    diff = abs(ref_entity["position"][axis] - client_entity["position"][axis])
                    if diff > tolerance:
                        raise AssertionError(
                            f"Position desync: entity {entity_id} {axis}: "
                            f"client 0={ref_entity['position'][axis]}, "
                            f"client {i}={client_entity['position'][axis]} "
                            f"(diff={diff:.4f})"
                        )
                
                # Health must match exactly (no floating point)
                assert ref_entity["health"] == client_entity["health"], \
                    f"Health desync: entity {entity_id}: client 0={ref_entity['health']}, " \
                    f"client {i}={client_entity['health']}"

def test_state_synchronized_after_combat():
    """All clients should agree on health values after a combat exchange."""
    harness = MultiplayerStateTestHarness(server_host="localhost", num_clients=4)
    harness.connect_all()
    
    match_id = harness.clients[0].create_match()
    for client in harness.clients[1:]:
        client.join_match(match_id)
    
    # Allow initial state to synchronize
    time.sleep(1.0)
    harness.assert_states_synchronized()
    
    # Client 0 attacks client 1
    harness.clients[0].attack_player("player-1")
    
    # Wait for state to propagate (3 ticks at 20Hz = 150ms)
    time.sleep(0.5)
    
    # All clients should see the same health values
    harness.assert_states_synchronized()

Concurrency Testing

What happens when two players click simultaneously? Race conditions in game servers can cause inconsistent state.

import threading
import concurrent.futures

def test_simultaneous_loot_pickup():
    """Two players grabbing the same loot simultaneously — only one should get it."""
    server = GameServer()
    match = server.create_match()
    
    player_a = GameClient("localhost")
    player_b = GameClient("localhost")
    player_a.join_match(match.id)
    player_b.join_match(match.id)
    
    # Spawn a loot item
    loot_id = match.spawn_loot(item="legendary_sword", at_position=(10, 0, 10))
    
    # Both players attempt to pick it up simultaneously
    results = []
    
    def pickup_attempt(client, player_name):
        result = client.pickup_item(loot_id)
        results.append({"player": player_name, "success": result.success, 
                        "item": result.item_id})
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
        future_a = executor.submit(pickup_attempt, player_a, "A")
        future_b = executor.submit(pickup_attempt, player_b, "B")
        concurrent.futures.wait([future_a, future_b])
    
    # Exactly one player should have gotten the item
    successes = [r for r in results if r["success"]]
    assert len(successes) == 1, \
        f"Expected exactly 1 player to get the loot, got {len(successes)}: {results}"
    
    # The item should not be present on the ground anymore
    assert not match.item_exists_in_world(loot_id), \
        "Loot item still exists in world after pickup"
    
    # The winning player should have the item in inventory
    winner = successes[0]["player"]
    winner_client = player_a if winner == "A" else player_b
    assert winner_client.has_item_in_inventory(loot_id)

def test_simultaneous_zone_capture():
    """Zone capture with simultaneous contestors — should have deterministic outcome."""
    results = []
    
    for _ in range(100):  # Run many times to catch race conditions
        match = create_test_match()
        player_a, player_b = create_two_players(match)
        
        zone_id = match.create_capture_zone()
        
        # Both players enter zone simultaneously
        with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
            executor.submit(player_a.enter_zone, zone_id)
            executor.submit(player_b.enter_zone, zone_id)
        
        time.sleep(5)  # Capture timer
        
        owner = match.get_zone_owner(zone_id)
        # Zone must have an owner — contested = neither is wrong game design,
        # but it must be deterministic: not sometimes A, sometimes B, sometimes neither
        results.append(owner)
    
    # Check that the outcomes are deterministic enough
    # (not all the same — that would indicate a bug — but consistent patterns)
    none_count = results.count(None)
    assert none_count < 10, \
        f"Zone had no owner in {none_count}/100 contests — contested state not resolved"

Anti-Cheat Input Validation

def test_server_rejects_teleport_hack():
    """Server should reject position updates that are physically impossible."""
    client = GameClient("localhost")
    client.connect_to_match()
    
    # Get current position
    initial_pos = client.get_player_position()
    
    # Try to teleport (move 1000 units in one frame)
    hacked_position = {
        "x": initial_pos["x"] + 1000,
        "y": initial_pos["y"],
        "z": initial_pos["z"]
    }
    
    result = client.send_raw_position_update(hacked_position)
    
    # Server should reject the update
    assert result.rejected, "Server accepted an impossible position update"
    
    # Player position should be unchanged or corrected
    actual_pos = client.get_player_position()
    distance = math.sqrt(
        (actual_pos["x"] - initial_pos["x"])**2 +
        (actual_pos["z"] - initial_pos["z"])**2
    )
    
    assert distance < 10, \
        f"Player teleported {distance:.1f} units — server did not correct position"

def test_server_rejects_instant_kill():
    """Server should validate damage values."""
    attacker = GameClient("localhost")
    victim = GameClient("localhost")
    
    attacker.connect_to_match()
    victim.connect_to_match()
    
    victim_max_health = victim.get_player_stats()["max_health"]
    
    # Try to deal 999999 damage (exceeds any weapon damage)
    result = attacker.send_raw_damage_packet(
        target_id=victim.player_id,
        damage=999999,
        weapon="pistol"  # A pistol can't do 999999 damage
    )
    
    assert result.rejected, "Server accepted invalid damage value"
    
    # Victim's health should be unchanged
    actual_health = victim.get_current_health()
    assert actual_health == victim_max_health, \
        f"Victim health changed from {victim_max_health} to {actual_health} after rejected damage"

Scalability Testing

def test_server_handles_100_concurrent_matches():
    """Server should handle 100 simultaneous 4-player matches (400 connections)."""
    import asyncio
    
    async def simulate_match(match_num: int) -> dict:
        clients = []
        errors = []
        
        try:
            # Create match
            server_client = AsyncGameClient("localhost")
            match_id = await server_client.create_match(max_players=4)
            
            # Join 4 players
            for i in range(4):
                client = AsyncGameClient("localhost")
                await client.join_match(match_id)
                clients.append(client)
            
            # Simulate 5 minutes of gameplay
            start = asyncio.get_event_loop().time()
            while asyncio.get_event_loop().time() - start < 300:
                for client in clients:
                    await client.send_action("random_action")
                await asyncio.sleep(0.1)  # 10Hz simulation
            
        except Exception as e:
            errors.append(str(e))
        finally:
            for client in clients:
                await client.disconnect()
        
        return {"match_num": match_num, "errors": errors}
    
    async def run_all_matches():
        tasks = [simulate_match(i) for i in range(100)]
        return await asyncio.gather(*tasks)
    
    results = asyncio.run(run_all_matches())
    
    failed_matches = [r for r in results if r["errors"]]
    assert len(failed_matches) < 5, \
        f"{len(failed_matches)}/100 matches had errors: {failed_matches[:3]}"

Multiplayer testing requires simulating adversarial network conditions from the start — not as an afterthought. The bugs you find at 150ms latency with 5% packet loss are exactly the bugs your worst-connected users will hit on launch day.

Read more