tests/test_serialization.py

   1 #!/usr/bin/env python
   2 # -*- coding: utf-8 -*-
   3 import unittest
   4 from minecraft.networking.types import (
   5     Type, Boolean, UnsignedByte, Byte, Short, UnsignedShort,
   6     Integer, FixedPointInteger, Angle, VarInt, Long, Float, Double,
   7     ShortPrefixedByteArray, VarIntPrefixedByteArray, UUID,
   8     String as StringType, Position, TrailingByteArray, UnsignedLong,
   9 )
  10 from minecraft.networking.packets.clientbound.play import (
  11     MultiBlockChangePacket
  12 )
  13 from minecraft.networking.packets import PacketBuffer
  14 from minecraft.networking.connection import ConnectionContext
  15 from minecraft import SUPPORTED_PROTOCOL_VERSIONS, RELEASE_PROTOCOL_VERSIONS
  16
  17
  18 TEST_VERSIONS = list(RELEASE_PROTOCOL_VERSIONS)
  19 if SUPPORTED_PROTOCOL_VERSIONS[-1] not in TEST_VERSIONS:
  20     TEST_VERSIONS.append(SUPPORTED_PROTOCOL_VERSIONS[-1])
  21
  22 TEST_DATA = {
  23     Boolean: [True, False],
  24     UnsignedByte: [0, 125],
  25     Byte: [-22, 22],
  26     Short: [-340, 22, 350],
  27     UnsignedShort: [0, 400],
  28     UnsignedLong: [0, 400],
  29     Integer: [-1000, 1000],
  30     FixedPointInteger: [float(-13098.3435), float(-0.83), float(1000)],
  31     Angle: [0, 360.0, 720, 47.12947238973, -108.7],
  32     VarInt: [1, 250, 50000, 10000000],
  33     Long: [50000000],
  34     Float: [21.000301],
  35     Double: [36.004002],
  36     ShortPrefixedByteArray: [bytes(245)],
  37     VarIntPrefixedByteArray: [bytes(1234)],
  38     TrailingByteArray: [b'Q^jO<5*|+o  LGc('],
  39     UUID: ["12345678-1234-5678-1234-567812345678"],
  40     StringType: ["hello world"],
  41     Position: [(758, 0, 691), (-500, -12, -684)],
  42     MultiBlockChangePacket.ChunkSectionPos: [
  43         (x, y, z)
  44         for x in [-0x200000, -123, -1, 0, 123, 0x1FFFFF]
  45         for z in [-0x200000, -456, -1, 0, 456, 0x1FFFFF]
  46         for y in [-0x80000, -789, -1, 0, 789, 0x7FFFF]
  47     ]
  48 }
  49
  50
  51 class SerializationTest(unittest.TestCase):
  52     def test_serialization(self):
  53         for protocol_version in TEST_VERSIONS:
  54             context = ConnectionContext(protocol_version=protocol_version)
  55
  56             for data_type in Type.__subclasses__():
  57                 if data_type in TEST_DATA:
  58                     test_cases = TEST_DATA[data_type]
  59
  60                     for test_data in test_cases:
  61                         packet_buffer = PacketBuffer()
  62                         data_type.send_with_context(
  63                             test_data, packet_buffer, context)
  64                         packet_buffer.reset_cursor()
  65
  66                         deserialized = data_type.read_with_context(
  67                             packet_buffer, context)
  68                         if data_type is FixedPointInteger:
  69                             self.assertAlmostEqual(
  70                                 test_data, deserialized, delta=1.0/32.0)
  71                         elif data_type is Angle:
  72                             self.assertAlmostEqual(test_data % 360,
  73                                                    deserialized,
  74                                                    delta=360/256)
  75                         elif data_type is Float or data_type is Double:
  76                             self.assertAlmostEqual(test_data, deserialized, 3)
  77                         else:
  78                             self.assertEqual(test_data, deserialized)
  79
  80     def test_exceptions(self):
  81         base_type = Type()
  82         with self.assertRaises(NotImplementedError):
  83             base_type.read(None)
  84
  85         with self.assertRaises(NotImplementedError):
  86             base_type.read_with_context(None, None)
  87
  88         with self.assertRaises(NotImplementedError):
  89             base_type.send(None, None)
  90
  91         with self.assertRaises(NotImplementedError):
  92             base_type.send_with_context(None, None, None)
  93
  94         with self.assertRaises(TypeError):
  95             Position.read(None)
  96
  97         with self.assertRaises(TypeError):
  98             Position.send(None, None)
  99
 100         empty_socket = PacketBuffer()
 101         with self.assertRaises(Exception):
 102             VarInt.read(empty_socket)
 103
 104     def test_varint(self):
 105         self.assertEqual(VarInt.size(2), 1)
 106         self.assertEqual(VarInt.size(1250), 2)
 107
 108         with self.assertRaises(ValueError):
 109             VarInt.size(2 ** 90)
 110
 111         with self.assertRaises(ValueError):
 112             packet_buffer = PacketBuffer()
 113             VarInt.send(2 ** 49, packet_buffer)
 114             packet_buffer.reset_cursor()
 115             VarInt.read(packet_buffer)
 116
 117         packet_buffer = PacketBuffer()
 118         VarInt.send(50000, packet_buffer)
 119         packet_buffer.reset_cursor()
 120
 121         self.assertEqual(VarInt.read(packet_buffer), 50000)