transcience/tests/unit/test_server_authority.gd

extends GutTest
## The security tests. Each one describes something a modified client would try
## and asserts that the authoritative world does not let it happen.
##
## The design intent is that these are boring to write, because the client has
## no message that expresses the cheat in the first place -- it can only send
## intent. These tests pin that property down so a future "just let the client
## send its position, it is simpler" change fails loudly.

var world: SimWorld
const PEER := 7


func before_each() -> void:
	world = SimWorld.new(1)
	world.add_player(PEER, "tester")


func _send(frame_tick: int, move := Vector2.ZERO, buttons := 0, aim := 0.0) -> void:
	var frames: Array[InputFrame] = [InputFrame.make(frame_tick, move, aim, buttons)]
	world.queue_input(PEER, frames)


## Drive the player for [param ticks] ticks with one fresh input per tick.
func _drive(ticks: int, move := Vector2.ZERO, buttons := 0, aim := 0.0) -> void:
	for _i in ticks:
		_send(world.tick + 1, move, buttons, aim)
		world.step()


func test_player_cannot_outrun_the_configured_speed() -> void:
	var start: Vector2 = world.players[PEER].pos
	_drive(60, Vector2(1.0, 0.0))
	var travelled: float = world.players[PEER].pos.distance_to(start)
	assert_almost_eq(travelled, SimConfig.PLAYER_SPEED, 1.0,
		"one second of held input must cover exactly one second of movement")


func test_replayed_input_is_dropped() -> void:
	_drive(5, Vector2.RIGHT)
	var pos_after: Vector2 = world.players[PEER].pos
	var acked: int = world.players[PEER].last_input_tick
	# Re-send an already-consumed tick, the classic replay attack.
	_send(acked, Vector2.RIGHT)
	assert_eq(world.players[PEER].input_queue.size(), 0)
	world.step()
	# The held input coasts one more tick, which is expected; what matters is
	# that the stale frame did not stack a second move on top of it.
	assert_almost_eq(world.players[PEER].pos.distance_to(pos_after),
		SimConfig.PLAYER_SPEED * SimConfig.TICK_DELTA, 0.001)


func test_input_from_the_far_future_is_dropped() -> void:
	_send(world.tick + SimConfig.INPUT_MAX_LEAD + 50, Vector2.RIGHT)
	assert_eq(world.players[PEER].input_queue.size(), 0,
		"a client cannot buy a head start by claiming a future tick")


func test_ancient_input_is_dropped() -> void:
	world.tick = 10000
	_send(1, Vector2.RIGHT)
	assert_eq(world.players[PEER].input_queue.size(), 0)


func test_input_flood_cannot_grow_the_queue_without_bound() -> void:
	for i in 500:
		_send(world.tick + 1 + i, Vector2.RIGHT)
	assert_lte(world.players[PEER].input_queue.size(), SimConfig.INPUT_MAX_AGE,
		"a flood of inputs must not become unbounded server memory")


## The shot sound used to ride on BULLET_SPAWN, and a bullet born inside an
## enemy is resolved and removed within the same tick -- so no spawn event was
## ever emitted and the shot was silent. Firing is now its own event.
func test_firing_is_reported_even_when_the_bullet_dies_instantly() -> void:
	var p: SimPlayer = world.players[PEER]
	p.pos = Vector2.ZERO
	p.aim = 0.0
	# The inert practice dummy, sitting exactly where the muzzle is, so the
	# bullet is consumed on the tick it is created. Deliberately not a turret:
	# a turret's own ring would fill the event list with its spawns and the
	# setup check below would pass for the wrong reason.
	world.spawn_enemy(Content.dummy(), Vector2(SimConfig.PLAYER_MUZZLE_OFFSET, 0.0))
	_drive(1, Vector2.ZERO, InputFrame.BTN_FIRE)
	assert_eq(_events_of(SimEvent.Type.PLAYER_FIRED).size(), 1,
		"the shot happened and has to be reported")
	assert_eq(_events_of(SimEvent.Type.BULLET_SPAWN).size(), 0,
		"setup check: the bullet really did die on the tick it spawned")


func test_firing_is_reported_once_per_shot() -> void:
	_drive(60, Vector2.ZERO, InputFrame.BTN_FIRE)
	var fired := _events_of(SimEvent.Type.PLAYER_FIRED).size()
	var expected := 60 / SimConfig.PLAYER_FIRE_COOLDOWN
	assert_almost_eq(float(fired), float(expected), 1.0,
		"one report per shot the cooldown actually allowed")


func test_fire_rate_is_enforced_by_the_server() -> void:
	# Hold fire every single tick; the server still applies its own cooldown.
	_drive(60, Vector2.ZERO, InputFrame.BTN_FIRE)
	var expected := 60 / SimConfig.PLAYER_FIRE_COOLDOWN
	assert_almost_eq(float(world.pool.live_count), float(expected), 2.0,
		"holding fire must not fire faster than the cooldown allows")


func test_a_starved_player_eventually_stops_moving() -> void:
	_drive(3, Vector2.RIGHT)
	var pos_at_starve: Vector2 = world.players[PEER].pos
	# Send nothing at all for a long time, as a disconnecting client would.
	for _i in SimConfig.INPUT_MAX_AGE + 120:
		world.step()
	var coasted: float = world.players[PEER].pos.distance_to(pos_at_starve)
	assert_lt(coasted, SimConfig.PLAYER_SPEED * 1.0,
		"a silent client must coast briefly, then stop, not drift forever")


func test_enemy_bullets_damage_the_player_and_are_consumed() -> void:
	var p: SimPlayer = world.players[PEER]
	p.pos = Vector2.ZERO
	world.pool.spawn(Vector2(-1.0, 0.0), Vector2.ZERO, 6.0, 60, 25,
		SimConfig.TEAM_ENEMY, SimConfig.KIND_ORB)
	world.step()
	assert_eq(p.hp, SimConfig.PLAYER_MAX_HP - 25)
	assert_eq(world.pool.live_count, 0, "a bullet that hits must be consumed")


## The inverse of what this asserted before. There are no i-frames: in a bullet
## hell the wall is the threat, and letting one hit grant immunity to the rest
## of the wall inverts the whole risk model.
func test_every_bullet_in_a_wall_lands() -> void:
	var p: SimPlayer = world.players[PEER]
	p.pos = Vector2.ZERO
	for _i in 3:
		world.pool.spawn(Vector2.ZERO, Vector2.ZERO, 6.0, 60, 25,
			SimConfig.TEAM_ENEMY, SimConfig.KIND_ORB)
	world.step()
	assert_eq(p.hp, SimConfig.PLAYER_MAX_HP - 75,
		"three bullets is three hits, not one")


func test_consecutive_ticks_both_deal_damage() -> void:
	var p: SimPlayer = world.players[PEER]
	p.pos = Vector2.ZERO
	world.pool.spawn(Vector2.ZERO, Vector2.ZERO, 6.0, 60, 10,
		SimConfig.TEAM_ENEMY, SimConfig.KIND_ORB)
	world.step()
	world.pool.spawn(Vector2.ZERO, Vector2.ZERO, 6.0, 60, 10,
		SimConfig.TEAM_ENEMY, SimConfig.KIND_ORB)
	world.step()
	assert_eq(p.hp, SimConfig.PLAYER_MAX_HP - 20,
		"there is no window of immunity after a hit")


func test_a_replica_world_never_resolves_a_hit() -> void:
	world.authoritative = false
	var p: SimPlayer = world.players[PEER]
	p.pos = Vector2.ZERO
	world.pool.spawn(Vector2.ZERO, Vector2.ZERO, 6.0, 60, 25,
		SimConfig.TEAM_ENEMY, SimConfig.KIND_ORB)
	world.step()
	assert_eq(p.hp, SimConfig.PLAYER_MAX_HP,
		"only the server decides damage; a client replica must never apply it")


func _kill_player() -> SimPlayer:
	var p: SimPlayer = world.players[PEER]
	p.pos = Vector2.ZERO
	p.hp = 5
	p.spawn_grace = 0
	world.pool.spawn(Vector2.ZERO, Vector2.ZERO, 6.0, 60, 25,
		SimConfig.TEAM_ENEMY, SimConfig.KIND_ORB)
	world.step()
	assert_false(p.alive, "setup: the player should be down")
	return p


func _events_of(type: int) -> Array:
	return world.events.filter(func(e: Dictionary) -> bool: return int(e["t"]) == type)


func test_a_downed_player_stays_down_without_input() -> void:
	var p := _kill_player()
	world.drain_events()
	for _i in 600:
		world.step()
	assert_false(p.alive, "there is no timed respawn -- death waits for the player")
	assert_eq(_events_of(SimEvent.Type.RESPAWN_REQUESTED).size(), 0)


func test_a_downed_player_asking_to_respawn_is_reported_once_per_tick() -> void:
	_kill_player()
	_drive(SimConfig.RESPAWN_LOCKOUT_TICKS)
	world.drain_events()
	_drive(1, Vector2.ZERO, InputFrame.BTN_INTERACT)
	assert_eq(_events_of(SimEvent.Type.RESPAWN_REQUESTED).size(), 1,
		"the request is an event for the instance layer, not a local revive")
	# Crucially the world does NOT revive the player itself: only the server's
	# instance layer can, by moving them to the hub.
	assert_false(world.players[PEER].alive)


## The HUD disables its button for the same duration, but that is presentation.
## A client that ignores its own UI and spams the bit still has to wait.
func test_the_respawn_lockout_is_enforced_by_the_server() -> void:
	_kill_player()
	world.drain_events()
	_drive(SimConfig.RESPAWN_LOCKOUT_TICKS - 2, Vector2.ZERO, InputFrame.BTN_INTERACT)
	assert_eq(_events_of(SimEvent.Type.RESPAWN_REQUESTED).size(), 0,
		"holding the request down must not shorten the lockout")
	_drive(4, Vector2.ZERO, InputFrame.BTN_INTERACT)
	assert_gt(_events_of(SimEvent.Type.RESPAWN_REQUESTED).size(), 0,
		"and it must actually open once the lockout expires")


func test_the_lockout_is_reported_to_the_client() -> void:
	_kill_player()
	var snap := NetCodec.decode_snapshot(NetCodec.encode_snapshot(world))
	var wait: float = snap["players"][0]["respawn_wait"]
	assert_almost_eq(wait,
		float(SimConfig.RESPAWN_LOCKOUT_TICKS) / float(SimConfig.TICK_RATE), 0.3,
		"the HUD needs the countdown to render its disabled button")


func test_respawn_request_never_reaches_the_client() -> void:
	_kill_player()
	_drive(1, Vector2.ZERO, InputFrame.BTN_INTERACT)
	var packet := NetCodec.decode_events(NetCodec.encode_events(world.tick, world.events))
	for ev: Dictionary in packet["events"]:
		assert_ne(int(ev["t"]), SimEvent.Type.RESPAWN_REQUESTED,
			"where a dead player goes is the server's decision")


func test_spawn_grace_blocks_damage_and_firing() -> void:
	var p: SimPlayer = world.players[PEER]
	p.pos = Vector2.ZERO
	p.spawn_grace = 60
	world.pool.spawn(Vector2.ZERO, Vector2.ZERO, 6.0, 60, 25,
		SimConfig.TEAM_ENEMY, SimConfig.KIND_ORB)
	_drive(1, Vector2.ZERO, InputFrame.BTN_FIRE)
	assert_eq(p.hp, SimConfig.PLAYER_MAX_HP, "arrival protection must absorb the hit")
	assert_eq(world.pool.live_count, 1,
		"only the enemy bullet: a protected player cannot shoot either")


func test_spawn_grace_expires() -> void:
	var p: SimPlayer = world.players[PEER]
	p.spawn_grace = 5
	_drive(6)
	assert_eq(p.spawn_grace, 0)
	assert_false(p.invulnerable())
	_drive(1, Vector2.ZERO, InputFrame.BTN_FIRE)
	assert_eq(world.pool.live_count, 1, "the gun comes back once grace ends")


## A muzzle offset moves a bullet's birthplace away from its shooter, and the
## bullet pool samples geometry once per tick so it never sees that jump. With
## six Doubleshots the sideways offset reaches 45px -- wider than a 32px wall
## plus the 6px hitbox that holds you off it -- so standing flush against a wall
## and firing along it put live shots down the far side. No modified client
## needed, which is what made it worth a test rather than a comment.
func _wall_at_column(column: int) -> SimWorld:
	var g := MapGrid.new(10, 6, MapGrid.Kind.FLOOR)
	for ty in 6:
		g.set_tile(column, ty, MapGrid.Kind.WALL)
	var w := SimWorld.new(1)
	w.set_map(g)
	return w


func _fire_flush_against_the_wall(doubleshots: int) -> Dictionary:
	var w := _wall_at_column(4)
	var p := w.add_player(PEER, "tester")
	var face: float = w.map.origin.x + 4.0 * MapGrid.TILE
	p.pos = Vector2(face - SimConfig.PLAYER_RADIUS, w.map.tile_centre(3, 3).y)
	p.aim = -PI / 2.0   # along the wall, not into it
	p.spawn_grace = 0
	p.fire_cooldown = 0
	var ids: Array[StringName] = []
	for _i in doubleshots:
		ids.append(Upgrades.DOUBLESHOT)
	p.stats = PlayerStats.build(ids)

	w._fire_player_shot(p)
	var fired := 0
	var past := 0
	for i in w.pool.high_water:
		if w.pool.alive[i] == 0:
			continue
		fired += 1
		if w.pool.pos[i].x > face:
			past += 1
	return {"fired": fired, "past_the_face": past}


func test_no_stack_of_upgrades_can_put_a_shot_through_a_wall() -> void:
	for doubleshots in [0, 5, 6, 10]:
		var r := _fire_flush_against_the_wall(doubleshots)
		assert_eq(int(r["past_the_face"]), 0,
			"%d Doubleshot(s) put %d shot(s) on the far side of the wall"
				% [doubleshots, r["past_the_face"]])


func test_clamping_the_muzzle_does_not_swallow_the_shot() -> void:
	# The fix must not silently reduce your firepower for standing near cover:
	# a blocked muzzle falls back to the shooter's centre, it does not cancel.
	assert_eq(int(_fire_flush_against_the_wall(6)["fired"]), 7,
		"six Doubleshots is seven projectiles, wall or no wall")