65 lines
2.2 KiB
Python
65 lines
2.2 KiB
Python
import pytest
|
|
|
|
from gns_snapshot_envelope import (
|
|
HEADER_SIZE,
|
|
MAGIC,
|
|
SnapshotEnvelopeError,
|
|
decode_snapshot,
|
|
encode_snapshot,
|
|
)
|
|
from packet_codec import MAX_MESSAGE_BYTES
|
|
|
|
|
|
def test_transform_and_npc_snapshots_round_trip_with_sequence():
|
|
transform = encode_snapshot("transform", b'{"type":"transform","x":1}', 7)
|
|
decoded_transform = decode_snapshot(transform)
|
|
assert decoded_transform is not None
|
|
assert decoded_transform.packet_type == "transform"
|
|
assert decoded_transform.sequence == 7
|
|
assert decoded_transform.payload == b'{"type":"transform","x":1}'
|
|
|
|
npc = encode_snapshot("npcState", b'{"type":"npcState","npcs":[]}', 9)
|
|
decoded_npc = decode_snapshot(npc)
|
|
assert decoded_npc is not None
|
|
assert decoded_npc.packet_type == "npcState"
|
|
assert decoded_npc.sequence == 9
|
|
|
|
|
|
def test_reliable_json_is_not_misidentified_as_snapshot_envelope():
|
|
assert decode_snapshot(b'{"type":"playerState"}') is None
|
|
|
|
|
|
def test_snapshot_envelope_rejects_invalid_family_sequence_and_size():
|
|
with pytest.raises(SnapshotEnvelopeError):
|
|
encode_snapshot("playerState", b"{}", 1)
|
|
with pytest.raises(SnapshotEnvelopeError):
|
|
encode_snapshot("transform", b"{}", 0)
|
|
with pytest.raises(SnapshotEnvelopeError):
|
|
encode_snapshot("transform", b"x" * MAX_MESSAGE_BYTES, 1)
|
|
|
|
|
|
def test_snapshot_decoder_rejects_truncated_or_corrupt_headers():
|
|
with pytest.raises(SnapshotEnvelopeError):
|
|
decode_snapshot(MAGIC)
|
|
|
|
valid = bytearray(encode_snapshot("transform", b"{}", 1))
|
|
valid[4] = 99
|
|
with pytest.raises(SnapshotEnvelopeError):
|
|
decode_snapshot(valid)
|
|
|
|
valid = bytearray(encode_snapshot("transform", b"{}", 1))
|
|
valid[5] = 99
|
|
with pytest.raises(SnapshotEnvelopeError):
|
|
decode_snapshot(valid)
|
|
|
|
valid = bytearray(encode_snapshot("transform", b"{}", 1))
|
|
valid[6:8] = b"\x00\x01"
|
|
with pytest.raises(SnapshotEnvelopeError):
|
|
decode_snapshot(valid)
|
|
|
|
|
|
def test_snapshot_header_leaves_payload_under_native_64k_cap():
|
|
payload = b"x" * (MAX_MESSAGE_BYTES - HEADER_SIZE)
|
|
message = encode_snapshot("transform", payload, 0xFFFFFFFF)
|
|
assert len(message) == MAX_MESSAGE_BYTES
|