Files
transcience/tests/unit/test_net_codec.gd
T
claude 050b8251a7
ci / verify (push) Successful in 48s
Stage 3: inventory, ground loot, and two loot visibilities
Four always-on-screen slots, items as data, and loot tables on enemies and
bosses. Health potions drop rarely from trash and always from the Warden;
the Warden also drops a Warden's Ration, one per living player, which does
nothing at all.

The ration is not filler. Player-instanced loot is a separate code path from
shared loot -- a distinct entity per owner, filtered per peer in the snapshot
encoder -- and the cheapest way to keep that path honest is to have something
in the game that exercises it on every boss kill.

Item actions ride the input frame rather than becoming new client messages.
InputFrame gained BTN_USE, BTN_DROP and a slot byte, which buys the packet-loss
redundancy, the replay guard on last_input_tick, ordering against movement on
the same tick, and a rate limit of one action per tick -- all of which a
separate RPC would have needed bolted back on. The cost is that anything in
the frame which must not repeat has to be edge-triggered, since frames are
resent and a starved server coasts on the last one it holds.

Instanced loot is enforced in NetCodec.encode_snapshot, beside the actor
interest radius: a peer is never told another player's copy exists. Hiding it
client-side would have been the same mistake as relying on fog to hide enemies.

Inventories live on the character and are written to the store on every
transaction, so a crash between "picked it up" and "wrote it down" cannot lose
or duplicate an item. Anything dropped becomes world-shared whatever it was
before, and a potion used at full health is refused rather than spent.

tools/diag_loot.tscn covers drop -> snapshot -> pick up -> persist -> use ->
drop plus both visibilities on the wire, for the same reason diag_progression
exists: bots are poor shots and almost never produce a drop. It asserts each
input frame was actually consumed, after an early version silently dropped its
first press and every later check passed for the wrong reason.

check.sh clean, 266 tests, SMOKE PASS, all three diagnostics green.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-04 21:16:15 +02:00

405 lines
16 KiB
GDScript

extends GutTest
## Codec round-trips. A field that encodes but decodes wrong shows up as a
## mysterious gameplay bug three layers away, so it is pinned down here.
var world: SimWorld
func before_each() -> void:
world = SimWorld.new(7)
var p := world.add_player(42, "ada")
p.pos = Vector2(120.5, -64.25)
p.aim = 1.25
p.hp = 73
p.escape_ticks = 30
p.last_input_tick = 555
world.spawn_enemy(Content.turret(), Vector2(-200.0, 100.0))
world.spawn_boss(Content.warden())
world.tick = 9001
func test_snapshot_round_trips_player_state() -> void:
var snap := NetCodec.decode_snapshot(NetCodec.encode_snapshot(world))
assert_eq(int(snap["tick"]), 9001)
var rec: Dictionary = snap["players"][0]
assert_eq(int(rec["peer"]), 42)
assert_almost_eq((rec["pos"] as Vector2).x, 120.5, 0.01)
assert_almost_eq((rec["pos"] as Vector2).y, -64.25, 0.01)
assert_almost_eq(float(rec["aim"]), 1.25, 0.001)
assert_eq(int(rec["hp"]), 73)
assert_eq(int(rec["last_input_tick"]), 555)
assert_true((int(rec["flags"]) & Protocol.F_ALIVE) != 0)
assert_true((int(rec["flags"]) & Protocol.F_ESCAPING) != 0)
assert_almost_eq(float(rec["escape"]), 30.0 / float(SimConfig.ESCAPE_CHANNEL_TICKS), 0.01)
func test_snapshot_carries_enemy_radius_for_late_joiners() -> void:
var snap := NetCodec.decode_snapshot(NetCodec.encode_snapshot(world))
var e: Dictionary = snap["enemies"][0]
assert_almost_eq(float(e["radius"]), Content.turret().radius, 0.5)
assert_eq(int(e["visual"]), Content.turret().visual)
func test_snapshot_carries_the_boss() -> void:
var snap := NetCodec.decode_snapshot(NetCodec.encode_snapshot(world))
assert_not_null(snap["boss"])
assert_eq(int(snap["boss"]["hp"]), Content.warden().max_hp)
func test_dead_enemies_are_not_sent() -> void:
for e in world.enemies.values():
e.alive = false
var snap := NetCodec.decode_snapshot(NetCodec.encode_snapshot(world))
assert_eq((snap["enemies"] as Array).size(), 0)
func test_event_round_trip_preserves_bullet_spawn() -> void:
var ev := {
"t": SimEvent.Type.BULLET_SPAWN, "uid": 8123,
"pos": Vector2(10.0, -20.0), "vel": Vector2(0.0, 150.0),
"r": 7.5, "life": 300, "kind": SimConfig.KIND_NEEDLE,
"team": SimConfig.TEAM_ENEMY, "accel": 12.0, "turn": 0.02,
}
var packet := NetCodec.decode_events(NetCodec.encode_events(77, [ev]))
assert_eq(int(packet["tick"]), 77)
var out: Dictionary = packet["events"][0]
assert_eq(int(out["uid"]), 8123)
assert_eq(out["pos"], Vector2(10.0, -20.0))
assert_eq(out["vel"], Vector2(0.0, 150.0))
assert_almost_eq(float(out["r"]), 7.5, 0.001)
assert_eq(int(out["life"]), 300)
assert_almost_eq(float(out["accel"]), 12.0, 0.001)
assert_almost_eq(float(out["turn"]), 0.02, 0.0001)
func test_server_only_events_never_reach_the_wire() -> void:
var events: Array[Dictionary] = [
{"t": SimEvent.Type.PORTAL_USED, "peer": 3},
{"t": SimEvent.Type.ESCAPE_COMPLETED, "peer": 3},
{"t": SimEvent.Type.PLAYER_DIED, "peer": 3},
]
var packet := NetCodec.decode_events(NetCodec.encode_events(1, events))
assert_eq((packet["events"] as Array).size(), 1,
"instance transfers are the server's business, not the client's")
assert_eq(int(packet["events"][0]["t"]), SimEvent.Type.PLAYER_DIED)
func test_input_round_trip() -> void:
var frames: Array[InputFrame] = [
InputFrame.make(10, Vector2(1, 0), 0.0, InputFrame.BTN_FIRE),
InputFrame.make(11, Vector2(0, 1), 1.0, 0),
]
var out := NetCodec.decode_inputs(NetCodec.encode_inputs(frames))
assert_eq(out.size(), 2)
assert_eq(out[0].tick, 10)
assert_eq(out[1].tick, 11)
func test_truncated_input_packet_is_rejected_not_read_past() -> void:
var frames: Array[InputFrame] = [InputFrame.make(10, Vector2.ONE, 0.0, 1)]
var data := NetCodec.encode_inputs(frames)
# Claim three frames but send one. A hostile client will try exactly this.
data[0] = 3
assert_eq(NetCodec.decode_inputs(data).size(), 0)
func test_empty_input_packet_is_safe() -> void:
assert_eq(NetCodec.decode_inputs(PackedByteArray()).size(), 0)
# --- Roster -----------------------------------------------------------------
func test_roster_round_trips() -> void:
var entries: Array[Dictionary] = [
{"peer": 7, "name": "ada", "kind": Protocol.InstanceKind.LOBBY,
"instance": 1, "alive": true},
{"peer": 9, "name": "grace", "kind": Protocol.InstanceKind.DUNGEON,
"instance": 4, "alive": false},
]
var out := NetCodec.decode_roster(NetCodec.encode_roster(entries))
assert_eq(out.size(), 2)
assert_eq(int(out[0]["peer"]), 7)
assert_eq(String(out[0]["name"]), "ada")
assert_eq(int(out[0]["kind"]), Protocol.InstanceKind.LOBBY)
assert_true(out[0]["alive"])
assert_eq(String(out[1]["name"]), "grace")
assert_eq(int(out[1]["kind"]), Protocol.InstanceKind.DUNGEON,
"the hub needs to know someone is already inside")
assert_eq(int(out[1]["instance"]), 4)
assert_false(out[1]["alive"])
func test_roster_survives_unicode_names() -> void:
var entries: Array[Dictionary] = [
{"peer": 1, "name": "ゆき", "kind": 0, "instance": 1, "alive": true}]
var out := NetCodec.decode_roster(NetCodec.encode_roster(entries))
assert_eq(String(out[0]["name"]), "ゆき")
func test_empty_roster_is_safe() -> void:
assert_eq(NetCodec.decode_roster(PackedByteArray()).size(), 0)
assert_eq(NetCodec.decode_roster(NetCodec.encode_roster([] as Array[Dictionary])).size(), 0)
# --- Snapshot extras --------------------------------------------------------
func test_snapshot_carries_spawn_grace_and_linkdead_flags() -> void:
var p: SimPlayer = world.players[42]
p.spawn_grace = 30
p.linkdead = true
var snap := NetCodec.decode_snapshot(NetCodec.encode_snapshot(world))
var flags := int(snap["players"][0]["flags"])
assert_true((flags & Protocol.F_SPAWN_GRACE) != 0)
assert_true((flags & Protocol.F_LINKDEAD) != 0)
func test_cleared_countdown_round_trips() -> void:
var snap := NetCodec.decode_snapshot(NetCodec.encode_snapshot(world, 17))
assert_eq(int(snap["cleared_countdown"]), 17)
func test_countdown_defaults_to_not_applicable() -> void:
var snap := NetCodec.decode_snapshot(NetCodec.encode_snapshot(world))
assert_eq(int(snap["cleared_countdown"]), Protocol.COUNTDOWN_NONE)
# --- Characters -------------------------------------------------------------
func test_character_roster_round_trips() -> void:
var rng := RandomNumberGenerator.new()
rng.seed = 5
var a := Character.create("Ada", rng)
a.grant_xp(Progression.total_xp_for_level(4))
var b := Character.create("Second", rng)
var out := NetCodec.decode_characters(
NetCodec.encode_characters([a, b] as Array[Character], a.id))
assert_eq(String(out["selected"]), a.id)
var chars: Array = out["characters"]
assert_eq(chars.size(), 2)
assert_eq(String(chars[0]["name"]), "Ada")
assert_eq(int(chars[0]["level"]), a.level)
assert_eq(int(chars[0]["max_hp"]), a.max_hp(),
"the roster shows each character's own health ceiling")
assert_true(chars[0]["active"])
## Colour is a character's only identity until cosmetics exist, so it has to
## survive the wire exactly rather than approximately.
func test_character_colour_survives_the_wire() -> void:
var rng := RandomNumberGenerator.new()
rng.seed = 11
var c := Character.create("Hue", rng)
var out := NetCodec.decode_characters(
NetCodec.encode_characters([c] as Array[Character], ""))
var got: Color = out["characters"][0]["colour"]
assert_eq(got.to_rgba32(), c.colour.to_rgba32())
func test_an_empty_character_list_is_safe() -> void:
var out := NetCodec.decode_characters(
NetCodec.encode_characters([] as Array[Character], ""))
assert_eq((out["characters"] as Array).size(), 0)
assert_eq(NetCodec.decode_characters(PackedByteArray())["characters"].size(), 0)
## Max health is per-character now, so a snapshot that assumed a constant would
## draw every levelled player's bar wrong.
func test_the_snapshot_carries_each_players_own_max_health() -> void:
var p: SimPlayer = world.players[42]
p.level = 5
p.max_hp = Progression.max_hp_for_level(5)
p.hp = 42
p.colour = Color(0.2, 0.6, 0.9)
var snap := NetCodec.decode_snapshot(NetCodec.encode_snapshot(world))
var rec: Dictionary = snap["players"][0]
assert_eq(int(rec["hp"]), 42)
assert_eq(int(rec["max_hp"]), Progression.max_hp_for_level(5))
assert_almost_eq((rec["colour"] as Color).b, 0.9, 0.02,
"party members are told apart by colour, so it rides the snapshot")
## The bar has to move on every kill, so experience rides the snapshot rather
## than waiting for a roster message that is only sent when the SET of
## characters changes.
func test_the_snapshot_carries_live_experience() -> void:
var p: SimPlayer = world.players[42]
p.total_xp = 1234
var snap := NetCodec.decode_snapshot(NetCodec.encode_snapshot(world))
assert_eq(int(snap["players"][0]["total_xp"]), 1234)
## The failure mode this file exists for, made explicit. A decoder that reads a
## field its encoder never wrote runs off the end of the buffer -- which is what
## happened when a snapshot change was applied to the character decoder by
## mistake.
func test_a_truncated_character_packet_does_not_read_past_the_end() -> void:
var rng := RandomNumberGenerator.new()
rng.seed = 21
var full := NetCodec.encode_characters(
[Character.create("Ada", rng), Character.create("Bee", rng)] as Array[Character], "")
for cut in [1, 4, 9, 17, 25, 33]:
if cut >= full.size():
continue
var out := NetCodec.decode_characters(full.slice(0, cut))
assert_lte((out["characters"] as Array).size(), 2,
"a packet cut at %d bytes must degrade, not invent entries" % cut)
func test_the_character_decoder_consumes_exactly_what_the_encoder_wrote() -> void:
var rng := RandomNumberGenerator.new()
rng.seed = 22
var chars := [Character.create("Solo", rng)] as Array[Character]
var data := NetCodec.encode_characters(chars, "")
# Append a sentinel: if the decoder reads the right number of bytes, the
# sentinel is untouched and the entry still decodes cleanly.
var padded := data.duplicate()
padded.append_array(PackedByteArray([0xAB, 0xCD]))
var out := NetCodec.decode_characters(padded)
assert_eq((out["characters"] as Array).size(), 1)
assert_eq(String(out["characters"][0]["name"]), "Solo")
## Truncation safety for the other string-carrying message, for the same reason.
func test_a_truncated_roster_packet_does_not_read_past_the_end() -> void:
var entries: Array[Dictionary] = [
{"peer": 1, "name": "someone", "kind": 0, "instance": 1, "alive": true},
{"peer": 2, "name": "another", "kind": 1, "instance": 3, "alive": true},
]
var full := NetCodec.encode_roster(entries)
for cut in range(1, full.size(), 3):
var out := NetCodec.decode_roster(full.slice(0, cut))
assert_lte(out.size(), 2,
"a roster cut at %d bytes must degrade, not invent entries" % cut)
assert_eq(NetCodec.decode_roster(full).size(), 2, "and the full packet still works")
# --- Inventory and loot -----------------------------------------------------
func test_the_snapshot_carries_the_observers_own_inventory() -> void:
var p: SimPlayer = world.players[42]
p.add_item(Items.HEALTH_POTION)
p.inventory[2] = Items.WARDENS_RATION
var snap := NetCodec.decode_snapshot(
NetCodec.encode_snapshot(world, Protocol.COUNTDOWN_NONE, 42))
var inv: Array = snap["inventory"]
assert_eq(inv.size(), SimConfig.INVENTORY_SLOTS)
assert_eq(Items.by_index(int(inv[0])), Items.HEALTH_POTION)
assert_eq(Items.by_index(int(inv[1])), Items.NONE)
assert_eq(Items.by_index(int(inv[2])), Items.WARDENS_RATION)
## Nobody needs to know what a party member is carrying, and the cheapest way to
## keep that true is to never put it on the wire.
func test_the_snapshot_never_carries_another_players_inventory() -> void:
var them := world.add_player(43, "them")
them.pos = Vector2(120.0, -64.0)
them.add_item(Items.HEALTH_POTION)
var snap := NetCodec.decode_snapshot(
NetCodec.encode_snapshot(world, Protocol.COUNTDOWN_NONE, 42))
assert_eq((snap["players"] as Array).size(), 2, "they should still be visible")
for inv_index in snap["inventory"]:
assert_eq(int(inv_index), 0, "but their bag must not be in this packet")
func test_ground_loot_round_trips() -> void:
var l := world.spawn_loot(Items.HEALTH_POTION, Vector2(64.0, -32.0))
var snap := NetCodec.decode_snapshot(NetCodec.encode_snapshot(world))
var got: Dictionary = snap["loot"][0]
assert_eq(int(got["id"]), l.id)
assert_almost_eq((got["pos"] as Vector2).x, 64.0, 0.01)
assert_almost_eq((got["pos"] as Vector2).y, -32.0, 0.01)
assert_eq(Items.by_index(int(got["item"])), Items.HEALTH_POTION)
## The wire is where player-instanced loot is actually enforced. A peer is never
## told another player's copy exists, so a modified client has nothing to
## reveal -- this is an interest-management rule, not a UI convention.
func test_another_players_instanced_loot_is_not_on_the_wire() -> void:
var p: SimPlayer = world.players[42]
world.spawn_loot(Items.WARDENS_RATION, p.pos + Vector2(10.0, 0.0), 43)
world.spawn_loot(Items.WARDENS_RATION, p.pos + Vector2(20.0, 0.0), 42)
world.spawn_loot(Items.HEALTH_POTION, p.pos + Vector2(30.0, 0.0))
var snap := NetCodec.decode_snapshot(
NetCodec.encode_snapshot(world, Protocol.COUNTDOWN_NONE, 42))
assert_eq((snap["loot"] as Array).size(), 2,
"my ration and the shared potion, never theirs")
func test_distant_loot_is_not_sent() -> void:
var p: SimPlayer = world.players[42]
world.spawn_loot(Items.HEALTH_POTION,
p.pos + Vector2(SimConfig.ACTOR_INTEREST_RADIUS + 200.0, 0.0))
var snap := NetCodec.decode_snapshot(
NetCodec.encode_snapshot(world, Protocol.COUNTDOWN_NONE, 42))
assert_eq((snap["loot"] as Array).size(), 0)
func test_item_events_round_trip() -> void:
var events: Array[Dictionary] = [
{"t": SimEvent.Type.ITEM_PICKED_UP, "peer": 42, "item": Items.HEALTH_POTION},
{"t": SimEvent.Type.ITEM_USED, "peer": 42, "item": Items.WARDENS_RATION},
{"t": SimEvent.Type.ITEM_DROPPED, "peer": 7, "item": Items.HEALTH_POTION},
]
var out: Array = NetCodec.decode_events(NetCodec.encode_events(1, events))["events"]
assert_eq(out.size(), 3)
assert_eq(int(out[0]["t"]), SimEvent.Type.ITEM_PICKED_UP)
assert_eq(out[0]["item"], Items.HEALTH_POTION)
assert_eq(int(out[1]["peer"]), 42)
assert_eq(out[1]["item"], Items.WARDENS_RATION)
assert_eq(int(out[2]["peer"]), 7)
assert_eq(out[2]["item"], Items.HEALTH_POTION)
## The events after the item ones must still decode. A fixed-width event whose
## body length is wrong desynchronises the whole rest of the packet, and that
## shows up as unrelated nonsense rather than as a decode error.
func test_events_after_an_item_event_still_decode() -> void:
var events: Array[Dictionary] = [
{"t": SimEvent.Type.ITEM_USED, "peer": 42, "item": Items.HEALTH_POTION},
{"t": SimEvent.Type.ENEMY_HIT, "id": 99, "dmg": 12, "hp": 400},
{"t": SimEvent.Type.PLAYER_DIED, "peer": 42},
]
var out: Array = NetCodec.decode_events(NetCodec.encode_events(1, events))["events"]
assert_eq(out.size(), 3)
assert_eq(int(out[1]["id"]), 99)
assert_eq(int(out[1]["hp"]), 400)
assert_eq(int(out[2]["t"]), SimEvent.Type.PLAYER_DIED)
func test_the_character_roster_carries_inventories() -> void:
var rng := RandomNumberGenerator.new()
rng.seed = 31
var c := Character.create("Packrat", rng)
c.inventory[0] = Items.HEALTH_POTION
c.inventory[3] = Items.WARDENS_RATION
var out := NetCodec.decode_characters(
NetCodec.encode_characters([c] as Array[Character], c.id))
var inv: Array = out["characters"][0]["inventory"]
assert_eq(Items.by_index(int(inv[0])), Items.HEALTH_POTION)
assert_eq(Items.by_index(int(inv[3])), Items.WARDENS_RATION)
## The snapshot is parsed positionally from the front, so a packet with no
## players, enemies or boss still has to land on the right byte for the
## inventory and loot that follow.
func test_an_empty_world_snapshot_still_decodes_its_tail() -> void:
var empty := SimWorld.new(1)
empty.spawn_loot(Items.HEALTH_POTION, Vector2(5.0, 5.0))
var snap := NetCodec.decode_snapshot(NetCodec.encode_snapshot(empty))
assert_eq((snap["players"] as Array).size(), 0)
assert_eq((snap["inventory"] as Array).size(), SimConfig.INVENTORY_SLOTS)
assert_eq((snap["loot"] as Array).size(), 1)
func test_an_input_frame_carries_its_slot_over_the_wire() -> void:
var frames: Array[InputFrame] = [
InputFrame.make(10, Vector2.ZERO, 0.0, InputFrame.BTN_USE, 3),
InputFrame.make(11, Vector2.ZERO, 0.0, InputFrame.BTN_DROP, 1),
]
var out := NetCodec.decode_inputs(NetCodec.encode_inputs(frames))
assert_eq(out.size(), 2)
assert_eq(out[0].slot, 3)
assert_true(out[0].pressed(InputFrame.BTN_USE))
assert_eq(out[1].slot, 1)
assert_true(out[1].pressed(InputFrame.BTN_DROP))