172 lines
5 KiB
Python
172 lines
5 KiB
Python
from twnet_parser.packet import \
|
|
ChunkHeaderParser
|
|
from twnet_parser.chunk_header import \
|
|
ChunkHeader
|
|
|
|
def test_chunk_header_unpack_vital1() -> None:
|
|
parser = ChunkHeaderParser()
|
|
header: ChunkHeader = parser.parse_header7(b'\x40\x10\x0a')
|
|
|
|
assert header.flags.resend is False
|
|
assert header.flags.vital is True
|
|
|
|
assert header.size == 16
|
|
assert header.seq == 10
|
|
|
|
def cmp_bytes_as_bin(got: bytes, expected: bytes) -> None:
|
|
g_bin = [format(int(x), '#010b') for x in got]
|
|
e_bin = [format(int(x), '#010b') for x in expected]
|
|
assert g_bin == e_bin
|
|
|
|
def test_chunk_header_pack_vital1() -> None:
|
|
header: ChunkHeader = ChunkHeader()
|
|
|
|
header.flags.resend = False
|
|
header.flags.vital = True
|
|
header.size = 16
|
|
header.seq = 10
|
|
|
|
assert bytes([0b01000000, 0b00010000, 0b00001010]) == b'\x40\x10\x0a'
|
|
|
|
cmp_bytes_as_bin(header.pack(), b'\x40\x10\x0a')
|
|
assert header.pack() == b'\x40\x10\x0a'
|
|
|
|
def test_chunk_header_pack_resend() -> None:
|
|
header: ChunkHeader = ChunkHeader()
|
|
|
|
header.flags.resend = True
|
|
header.flags.vital = False
|
|
header.size = 16
|
|
header.seq = 0
|
|
|
|
assert bytes([0b10000000, 0b00010000]) == b'\x80\x10'
|
|
|
|
cmp_bytes_as_bin(header.pack(), b'\x80\x10')
|
|
assert header.pack() == b'\x80\x10'
|
|
|
|
def test_chunk_header_pack_no_flags1() -> None:
|
|
header: ChunkHeader = ChunkHeader()
|
|
|
|
header.flags.resend = False
|
|
header.flags.vital = False
|
|
header.size = 16
|
|
header.seq = 0
|
|
|
|
assert bytes([0b00000000, 0b00010000]) == b'\x00\x10'
|
|
|
|
cmp_bytes_as_bin(header.pack(), b'\x00\x10')
|
|
assert header.pack() == b'\x00\x10'
|
|
|
|
def test_chunk_header_pack_no_flags2() -> None:
|
|
header: ChunkHeader = ChunkHeader()
|
|
|
|
header.flags.resend = False
|
|
header.flags.vital = False
|
|
header.seq = 0
|
|
|
|
header.size = 17
|
|
assert bytes([0b00000000, 0b00010001]) == b'\x00\x11'
|
|
cmp_bytes_as_bin(header.pack(), b'\x00\x11')
|
|
assert header.pack() == b'\x00\x11'
|
|
|
|
header.size = 18
|
|
assert bytes([0b00000000, 0b00010010]) == b'\x00\x12'
|
|
cmp_bytes_as_bin(header.pack(), b'\x00\x12')
|
|
assert header.pack() == b'\x00\x12'
|
|
|
|
header.size = 19
|
|
assert bytes([0b00000000, 0b00010011]) == b'\x00\x13'
|
|
cmp_bytes_as_bin(header.pack(), b'\x00\x13')
|
|
assert header.pack() == b'\x00\x13'
|
|
|
|
header.size = 20
|
|
assert bytes([0b00000000, 0b00010100]) == b'\x00\x14'
|
|
cmp_bytes_as_bin(header.pack(), b'\x00\x14')
|
|
assert header.pack() == b'\x00\x14'
|
|
|
|
header.size = 21
|
|
assert bytes([0b00000000, 0b00010101]) == b'\x00\x15'
|
|
cmp_bytes_as_bin(header.pack(), b'\x00\x15')
|
|
assert header.pack() == b'\x00\x15'
|
|
|
|
def test_chunk_header_pack_no_flags_max_size() -> None:
|
|
header: ChunkHeader = ChunkHeader()
|
|
|
|
header.flags.resend = False
|
|
header.flags.vital = False
|
|
header.seq = 0
|
|
|
|
header.size = 4094
|
|
#
|
|
# from libtw2 docs
|
|
# https://github.com/heinrich5991/libtw2/blob/7885c99974ee445ce13297b72ae3e7c6ea3b969d/doc/packet7.md
|
|
#
|
|
# chunk7_header_nonvital:
|
|
# [ 1] flag_resend
|
|
# [ 1] flag_vital
|
|
# [ 6] <----------
|
|
# [ 2] padding |-- size
|
|
# [ 6] <----------
|
|
#
|
|
# FFss ssss PPss ssss
|
|
assert bytes([0b0011_1111, 0b0011_1110]) == b'\x3f\x3e'
|
|
cmp_bytes_as_bin(header.pack(), b'\x3f\x3e')
|
|
assert header.pack() == b'\x3f\x3e'
|
|
|
|
header.size = 4095
|
|
#
|
|
# from libtw2 docs
|
|
# https://github.com/heinrich5991/libtw2/blob/7885c99974ee445ce13297b72ae3e7c6ea3b969d/doc/packet7.md
|
|
#
|
|
# chunk7_header_nonvital:
|
|
# [ 1] flag_resend
|
|
# [ 1] flag_vital
|
|
# [ 6] <----------
|
|
# [ 2] padding |-- size
|
|
# [ 6] <----------
|
|
#
|
|
# FFss ssss PPss ssss
|
|
assert bytes([0b0011_1111, 0b0011_1111]) == b'\x3f\x3f'
|
|
cmp_bytes_as_bin(header.pack(), b'\x3f\x3f')
|
|
assert header.pack() == b'\x3f\x3f'
|
|
|
|
|
|
def test_chunk_header_pack_no_flags_out_of_bounds_size() -> None:
|
|
header: ChunkHeader = ChunkHeader()
|
|
|
|
header.flags.resend = False
|
|
header.flags.vital = False
|
|
header.seq = 0
|
|
|
|
header.size = 4096
|
|
#
|
|
# from libtw2 docs
|
|
# https://github.com/heinrich5991/libtw2/blob/7885c99974ee445ce13297b72ae3e7c6ea3b969d/doc/packet7.md
|
|
#
|
|
# chunk7_header_nonvital:
|
|
# [ 1] flag_resend
|
|
# [ 1] flag_vital
|
|
# [ 6] <----------
|
|
# [ 2] padding |-- size
|
|
# [ 6] <----------
|
|
#
|
|
# FFss ssss PPss ssss
|
|
assert bytes([0b0000_0000, 0b0000_0000]) == b'\x00\x00'
|
|
# ^
|
|
# does overflow to zero
|
|
# should we error here instead?
|
|
# TODO: https://gitlab.com/teeworlds-network/twnet_parser/-/issues/2
|
|
cmp_bytes_as_bin(header.pack(), b'\x00\x00')
|
|
assert header.pack() == b'\x00\x00'
|
|
|
|
|
|
def test_chunk_header_unpack_vital2() -> None:
|
|
parser = ChunkHeaderParser()
|
|
header: ChunkHeader = parser.parse_header7(b'\x40\x10\x09')
|
|
|
|
assert header.flags.resend is False
|
|
assert header.flags.vital is True
|
|
|
|
assert header.size == 16
|
|
assert header.seq == 9
|