# -*- coding: utf-8 -*- # Run test_umsgpack.py with your Python interpreter of choice to test the # correctness of u-msgpack-python! # # $ python2 test_umsgpack.py # $ python3 test_umsgpack.py # $ pypy test_umsgpack.py # $ pypy3 test_umsgpack.py # import sys import struct import unittest import io from collections import OrderedDict, namedtuple import umsgpack single_test_vectors = [ # None [ "nil", None, b"\xc0" ], # Booleans [ "bool false", False, b"\xc2" ], [ "bool true", True, b"\xc3" ], # + 7-bit uint [ "7-bit uint", 0x00, b"\x00" ], [ "7-bit uint", 0x10, b"\x10" ], [ "7-bit uint", 0x7f, b"\x7f" ], # - 5-bit int [ "5-bit sint", -1, b"\xff" ], [ "5-bit sint", -16, b"\xf0" ], [ "5-bit sint", -32, b"\xe0" ], # 8-bit uint [ "8-bit uint", 0x80, b"\xcc\x80" ], [ "8-bit uint", 0xf0, b"\xcc\xf0" ], [ "8-bit uint", 0xff, b"\xcc\xff" ], # 16-bit uint [ "16-bit uint", 0x100, b"\xcd\x01\x00" ], [ "16-bit uint", 0x2000, b"\xcd\x20\x00" ], [ "16-bit uint", 0xffff, b"\xcd\xff\xff" ], # 32-bit uint [ "32-bit uint", 0x10000, b"\xce\x00\x01\x00\x00" ], [ "32-bit uint", 0x200000, b"\xce\x00\x20\x00\x00" ], [ "32-bit uint", 0xffffffff, b"\xce\xff\xff\xff\xff" ], # 64-bit uint [ "64-bit uint", 0x100000000, b"\xcf\x00\x00\x00\x01\x00\x00\x00\x00" ], [ "64-bit uint", 0x200000000000, b"\xcf\x00\x00\x20\x00\x00\x00\x00\x00" ], [ "64-bit uint", 0xffffffffffffffff, b"\xcf\xff\xff\xff\xff\xff\xff\xff\xff" ], # 8-bit int [ "8-bit int", -33, b"\xd0\xdf" ], [ "8-bit int", -100, b"\xd0\x9c" ], [ "8-bit int", -128, b"\xd0\x80" ], # 16-bit int [ "16-bit int", -129, b"\xd1\xff\x7f" ], [ "16-bit int", -2000, b"\xd1\xf8\x30" ], [ "16-bit int", -32768, b"\xd1\x80\x00" ], # 32-bit int [ "32-bit int", -32769, b"\xd2\xff\xff\x7f\xff" ], [ "32-bit int", -1000000000, b"\xd2\xc4\x65\x36\x00" ], [ "32-bit int", -2147483648, b"\xd2\x80\x00\x00\x00" ], # 64-bit int [ "64-bit int", -2147483649, b"\xd3\xff\xff\xff\xff\x7f\xff\xff\xff" ], [ "64-bit int", -1000000000000000002, b"\xd3\xf2\x1f\x49\x4c\x58\x9b\xff\xfe" ], [ "64-bit int", -9223372036854775808, b"\xd3\x80\x00\x00\x00\x00\x00\x00\x00" ], # 64-bit float [ "64-bit float", 0.0, b"\xcb\x00\x00\x00\x00\x00\x00\x00\x00" ], [ "64-bit float", 2.5, b"\xcb\x40\x04\x00\x00\x00\x00\x00\x00" ], [ "64-bit float", float(10**35), b"\xcb\x47\x33\x42\x61\x72\xc7\x4d\x82" ], # Fixstr String [ "fix string", u"", b"\xa0" ], [ "fix string", u"a", b"\xa1\x61" ], [ "fix string", u"abc", b"\xa3\x61\x62\x63" ], [ "fix string", u"a"*31, b"\xbf" + b"\x61"*31 ], # 8-bit String [ "8-bit string", u"b"*32, b"\xd9\x20" + b"b"*32 ], [ "8-bit string", u"c"*100, b"\xd9\x64" + b"c"*100 ], [ "8-bit string", u"d"*255, b"\xd9\xff" + b"d"*255 ], # 16-bit String [ "16-bit string", u"b"*256, b"\xda\x01\x00" + b"b"*256 ], [ "16-bit string", u"c"*65535, b"\xda\xff\xff" + b"c"*65535 ], # 32-bit String [ "32-bit string", u"b"*65536, b"\xdb\x00\x01\x00\x00" + b"b"*65536 ], # Wide character String [ "wide char string", u"Allagbé", b"\xa8Allagb\xc3\xa9" ], [ "wide char string", u"По оживлённым берегам", b"\xd9\x28\xd0\x9f\xd0\xbe\x20\xd0\xbe\xd0\xb6\xd0\xb8\xd0\xb2\xd0\xbb\xd1\x91\xd0\xbd\xd0\xbd\xd1\x8b\xd0\xbc\x20\xd0\xb1\xd0\xb5\xd1\x80\xd0\xb5\xd0\xb3\xd0\xb0\xd0\xbc" ], # 8-bit Binary [ "8-bit binary", b"\x80"*1, b"\xc4\x01" + b"\x80"*1 ], [ "8-bit binary", b"\x80"*32, b"\xc4\x20" + b"\x80"*32 ], [ "8-bit binary", b"\x80"*255, b"\xc4\xff" + b"\x80"*255 ], # 16-bit Binary [ "16-bit binary", b"\x80"*256, b"\xc5\x01\x00" + b"\x80"*256 ], # 32-bit Binary [ "32-bit binary", b"\x80"*65536, b"\xc6\x00\x01\x00\x00" + b"\x80"*65536 ], # Fixext 1 [ "fixext 1", umsgpack.Ext(0x05, b"\x80"*1), b"\xd4\x05" + b"\x80"*1 ], # Fixext 2 [ "fixext 2", umsgpack.Ext(0x05, b"\x80"*2), b"\xd5\x05" + b"\x80"*2 ], # Fixext 4 [ "fixext 4", umsgpack.Ext(0x05, b"\x80"*4), b"\xd6\x05" + b"\x80"*4 ], # Fixext 8 [ "fixext 8", umsgpack.Ext(0x05, b"\x80"*8), b"\xd7\x05" + b"\x80"*8 ], # Fixext 16 [ "fixext 16", umsgpack.Ext(0x05, b"\x80"*16), b"\xd8\x05" + b"\x80"*16 ], # 8-bit Ext [ "8-bit ext", umsgpack.Ext(0x05, b"\x80"*255), b"\xc7\xff\x05" + b"\x80"*255 ], # 16-bit Ext [ "16-bit ext", umsgpack.Ext(0x05, b"\x80"*256), b"\xc8\x01\x00\x05" + b"\x80"*256 ], # 32-bit Ext [ "32-bit ext", umsgpack.Ext(0x05, b"\x80"*65536), b"\xc9\x00\x01\x00\x00\x05" + b"\x80"*65536 ], # Empty Array [ "empty array", [], b"\x90" ], # Empty Map [ "empty map", {}, b"\x80" ], ] composite_test_vectors = [ # Fix Array [ "fix array", [ 5, u"abc", True ], b"\x93\x05\xa3\x61\x62\x63\xc3" ], # 16-bit Array [ "16-bit array", [ 0x05 ]*16, b"\xdc\x00\x10" + b"\x05"*16 ], [ "16-bit array", [ 0x05 ]*65535, b"\xdc\xff\xff" + b"\x05"*65535 ], # 32-bit Array [ "32-bit array", [ 0x05 ]*65536, b"\xdd\x00\x01\x00\x00" + b"\x05"*65536 ], # Fix Map [ "fix map", OrderedDict([(1, True), (2, u"abc"), (3, b"\x80")]), b"\x83\x01\xc3\x02\xa3\x61\x62\x63\x03\xc4\x01\x80" ], [ "fix map", { u"abc" : 5 }, b"\x81\xa3\x61\x62\x63\x05" ], [ "fix map", { b"\x80" : 0xffff }, b"\x81\xc4\x01\x80\xcd\xff\xff" ], [ "fix map", { True : None }, b"\x81\xc3\xc0" ], # 16-bit Map [ "16-bit map", OrderedDict([(k, 0x05) for k in range(16)]), b"\xde\x00\x10" + b"".join([struct.pack("B", i) + b"\x05" for i in range(16)])], [ "16-bit map", OrderedDict([(k, 0x05) for k in range(6000)]), b"\xde\x17\x70" + b"".join([struct.pack("B", i) + b"\x05" for i in range(128)]) + b"".join([b"\xcc" + struct.pack("B", i) + b"\x05" for i in range(128, 256)]) + b"".join([b"\xcd" + struct.pack(">H", i) + b"\x05" for i in range(256, 6000)]) ], # Complex Array [ "complex array", [ True, 0x01, umsgpack.Ext(0x03, b"foo"), 0xff, OrderedDict([(1, False), (2, u"abc")]), b"\x80", [1, 2, 3], u"abc" ], b"\x98\xc3\x01\xc7\x03\x03\x66\x6f\x6f\xcc\xff\x82\x01\xc2\x02\xa3\x61\x62\x63\xc4\x01\x80\x93\x01\x02\x03\xa3\x61\x62\x63" ], # Complex Map [ "complex map", OrderedDict([(1, [OrderedDict([(1, 2), (3, 4)]), {}]), (2, 1), (3, [False, u"def"]), (4, OrderedDict([(0x100000000, u"a"), (0xffffffff, u"b")]))]), b"\x84\x01\x92\x82\x01\x02\x03\x04\x80\x02\x01\x03\x92\xc2\xa3\x64\x65\x66\x04\x82\xcf\x00\x00\x00\x01\x00\x00\x00\x00\xa1\x61\xce\xff\xff\xff\xff\xa1\x62" ], # Map with Tuple Keys [ "map with tuple keys", OrderedDict([((u"foo", False, 3), True), ((3e6, -5), u"def")]), b"\x82\x93\xa3\x66\x6f\x6f\xc2\x03\xc3\x92\xcb\x41\x46\xe3\x60\x00\x00\x00\x00\xfb\xa3\x64\x65\x66" ], # Map with Complex Tuple Keys [ "map with complex tuple keys", {(u"foo", (1,2,3), 3) : -5}, b"\x81\x93\xa3\x66\x6f\x6f\x93\x01\x02\x03\x03\xfb" ] ] pack_exception_test_vectors = [ # Unsupported type exception [ "unsupported type", set([1,2,3]), umsgpack.UnsupportedTypeException ], [ "unsupported type", -2**(64-1)-1, umsgpack.UnsupportedTypeException ], [ "unsupported type", 2**64, umsgpack.UnsupportedTypeException ], ] unpack_exception_test_vectors = [ # Type errors [ "type error unpack unicode string", u"\x01", TypeError ], [ "type error unpack boolean", True, TypeError ], # Insufficient data to unpack object [ "insufficient data 8-bit uint", b"\xcc", umsgpack.InsufficientDataException ], [ "insufficient data 16-bit uint", b"\xcd\xff", umsgpack.InsufficientDataException ], [ "insufficient data 32-bit uint", b"\xce\xff", umsgpack.InsufficientDataException ], [ "insufficient data 64-bit uint", b"\xcf\xff", umsgpack.InsufficientDataException ], [ "insufficient data 8-bit int", b"\xd0", umsgpack.InsufficientDataException ], [ "insufficient data 16-bit int", b"\xd1\xff", umsgpack.InsufficientDataException ], [ "insufficient data 32-bit int", b"\xd2\xff", umsgpack.InsufficientDataException ], [ "insufficient data 64-bit int", b"\xd3\xff", umsgpack.InsufficientDataException ], [ "insufficient data 32-bit float", b"\xca\xff", umsgpack.InsufficientDataException ], [ "insufficient data 64-bit float", b"\xcb\xff", umsgpack.InsufficientDataException ], [ "insufficient data fixstr", b"\xa1", umsgpack.InsufficientDataException ], [ "insufficient data 8-bit string", b"\xd9", umsgpack.InsufficientDataException ], [ "insufficient data 8-bit string", b"\xd9\x01", umsgpack.InsufficientDataException ], [ "insufficient data 16-bit string", b"\xda\x01\x00", umsgpack.InsufficientDataException ], [ "insufficient data 32-bit string", b"\xdb\x00\x01\x00\x00", umsgpack.InsufficientDataException ], [ "insufficient data 8-bit binary", b"\xc4", umsgpack.InsufficientDataException ], [ "insufficient data 8-bit binary", b"\xc4\x01", umsgpack.InsufficientDataException ], [ "insufficient data 16-bit binary", b"\xc5\x01\x00", umsgpack.InsufficientDataException ], [ "insufficient data 32-bit binary", b"\xc6\x00\x01\x00\x00", umsgpack.InsufficientDataException ], [ "insufficient data fixarray", b"\x91", umsgpack.InsufficientDataException ], [ "insufficient data fixarray", b"\x92\xc2", umsgpack.InsufficientDataException ], [ "insufficient data 16-bit array", b"\xdc\x00\xf0\xc2\xc3", umsgpack.InsufficientDataException ], [ "insufficient data 32-bit array", b"\xdd\x00\x01\x00\x00\xc2\xc3", umsgpack.InsufficientDataException ], [ "insufficient data fixmap", b"\x81", umsgpack.InsufficientDataException ], [ "insufficient data fixmap", b"\x82\xc2\xc3", umsgpack.InsufficientDataException ], [ "insufficient data 16-bit map", b"\xde\x00\xf0\xc2\xc3", umsgpack.InsufficientDataException ], [ "insufficient data 32-bit map", b"\xdf\x00\x01\x00\x00\xc2\xc3", umsgpack.InsufficientDataException ], [ "insufficient data fixext 1", b"\xd4", umsgpack.InsufficientDataException ], [ "insufficient data fixext 1", b"\xd4\x05", umsgpack.InsufficientDataException ], [ "insufficient data fixext 2", b"\xd5\x05\x01", umsgpack.InsufficientDataException ], [ "insufficient data fixext 4", b"\xd6\x05\x01\x02\x03", umsgpack.InsufficientDataException ], [ "insufficient data fixext 8", b"\xd7\x05\x01\x02\x03", umsgpack.InsufficientDataException ], [ "insufficient data fixext 16", b"\xd8\x05\x01\x02\x03", umsgpack.InsufficientDataException ], [ "insufficient data ext 8-bit", b"\xc7\x05\x05\x01\x02\x03", umsgpack.InsufficientDataException ], [ "insufficient data ext 16-bit", b"\xc8\x01\x00\x05\x01\x02\x03", umsgpack.InsufficientDataException ], [ "insufficient data ext 32-bit", b"\xc9\x00\x01\x00\x00\x05\x01\x02\x03", umsgpack.InsufficientDataException ], # Unhashable key { 1 : True, { 1 : 1 } : False } [ "unhashable key", b"\x82\x01\xc3\x81\x01\x01\xc2", umsgpack.UnhashableKeyException ], # Unhashable key { [ 1, 2, {} ] : True } [ "unhashable key", b"\x81\x93\x01\x02\x80\xc3", umsgpack.UnhashableKeyException ], # Key duplicate { 1 : True, 1 : False } [ "duplicate key", b"\x82\x01\xc3\x01\xc2", umsgpack.DuplicateKeyException ], # Reserved code (0xc1) [ "reserved code", b"\xc1", umsgpack.ReservedCodeException ], # Invalid string (non utf-8) [ "invalid string", b"\xa1\x80", umsgpack.InvalidStringException ], ] compatibility_test_vectors = [ # Fix Raw [ "fix raw", b"", b"\xa0" ], [ "fix raw", u"", b"\xa0" ], [ "fix raw", b"a", b"\xa1\x61" ], [ "fix raw", u"a", b"\xa1\x61" ], [ "fix raw", b"abc", b"\xa3\x61\x62\x63" ], [ "fix raw", u"abc", b"\xa3\x61\x62\x63" ], [ "fix raw", b"a"*31, b"\xbf" + b"\x61"*31 ], [ "fix raw", u"a"*31, b"\xbf" + b"\x61"*31 ], # 16-bit Raw [ "16-bit raw", u"b"*32, b"\xda\x00\x20" + b"b"*32 ], [ "16-bit raw", b"b"*32, b"\xda\x00\x20" + b"b"*32 ], [ "16-bit raw", u"b"*256, b"\xda\x01\x00" + b"b"*256 ], [ "16-bit raw", b"b"*256, b"\xda\x01\x00" + b"b"*256 ], [ "16-bit raw", u"c"*65535, b"\xda\xff\xff" + b"c"*65535 ], [ "16-bit raw", b"c"*65535, b"\xda\xff\xff" + b"c"*65535 ], # 32-bit Raw [ "32-bit raw", u"b"*65536, b"\xdb\x00\x01\x00\x00" + b"b"*65536 ], [ "32-bit raw", b"b"*65536, b"\xdb\x00\x01\x00\x00" + b"b"*65536 ], ] CustomType = namedtuple('CustomType', ['x', 'y', 'z']) ext_handlers = { complex: lambda obj: umsgpack.Ext(0x20, struct.pack("ff", obj.real, obj.imag)), CustomType: lambda obj: umsgpack.Ext(0x30, umsgpack.packb(list(obj))), 0x20: lambda ext: complex(*struct.unpack("ff", ext.data)), 0x30: lambda ext: CustomType(*umsgpack.unpackb(ext.data)), } ext_handlers_test_vectors = [ [ "complex", complex(1, 2), b"\xd7\x20\x00\x00\x80\x3f\x00\x00\x00\x40" ], [ "custom type", CustomType(b"abc", 123, True), b"\xd7\x30\x93\xc4\x03\x61\x62\x63\x7b\xc3" ], ] # These are the only global variables that should be exported by umsgpack exported_vars_test_vector = [ "Ext", "InvalidString", "PackException", "UnpackException", "UnsupportedTypeException", "InsufficientDataException", "InvalidStringException", "ReservedCodeException", "UnhashableKeyException", "DuplicateKeyException", "KeyNotPrimitiveException", "KeyDuplicateException", "pack", "packb", "unpack", "unpackb", "dump", "dumps", "load", "loads", "version", "compatibility", ] ################################################################################ class TestUmsgpack(unittest.TestCase): def test_pack_single(self): for (name, obj, data) in single_test_vectors: obj_repr = repr(obj) print("\tTesting %s: object %s" % (name, obj_repr if len(obj_repr) < 24 else obj_repr[0:24] + "...")) self.assertEqual(umsgpack.packb(obj), data) def test_pack_composite(self): for (name, obj, data) in composite_test_vectors: obj_repr = repr(obj) print("\tTesting %s: object %s" % (name, obj_repr if len(obj_repr) < 24 else obj_repr[0:24] + "...")) self.assertEqual(umsgpack.packb(obj), data) def test_pack_exceptions(self): for (name, obj, exception) in pack_exception_test_vectors: obj_repr = repr(obj) print("\tTesting %s: object %s" % (name, obj_repr if len(obj_repr) < 24 else obj_repr[0:24] + "...")) with self.assertRaises(exception): umsgpack.packb(obj) def test_unpack_single(self): for (name, obj, data) in single_test_vectors: obj_repr = repr(obj) print("\tTesting %s: object %s" % (name, obj_repr if len(obj_repr) < 24 else obj_repr[0:24] + "...")) unpacked = umsgpack.unpackb(data) # In Python2, we have both int and long integer types, but which # one we end up with depends on the architecture (32-bit/64-bit) if sys.version_info[0] == 2: # Allow both {int,long} -> unpackb -> {int,long} if isinstance(obj, int) or isinstance(obj, long): self.assertTrue(isinstance(unpacked, int) or isinstance(unpacked, long)) else: self.assertTrue(isinstance(unpacked, type(obj))) # In Python3, we only have the int integer type else: self.assertTrue(isinstance(unpacked, type(obj))) self.assertEqual(unpacked, obj) def test_unpack_composite(self): for (name, obj, data) in composite_test_vectors: obj_repr = repr(obj) print("\tTesting %s: object %s" % (name, obj_repr if len(obj_repr) < 24 else obj_repr[0:24] + "...")) self.assertEqual(umsgpack.unpackb(data), obj) def test_unpack_exceptions(self): for (name, data, exception) in unpack_exception_test_vectors: print("\tTesting %s" % name) with self.assertRaises(exception): umsgpack.unpackb(data) def test_pack_compatibility(self): umsgpack.compatibility = True for (name, obj, data) in compatibility_test_vectors: obj_repr = repr(obj) print("\tTesting %s: object %s" % (name, obj_repr if len(obj_repr) < 24 else obj_repr[0:24] + "...")) self.assertEqual(umsgpack.packb(obj), data) umsgpack.compatibility = False def test_unpack_compatibility(self): umsgpack.compatibility = True for (name, obj, data) in compatibility_test_vectors: obj_repr = repr(obj) print("\tTesting %s: object %s" % (name, obj_repr if len(obj_repr) < 24 else obj_repr[0:24] + "...")) unpacked = umsgpack.unpackb(data) # Encoded raw should always unpack to bytes in compatibility mode, # so convert any string obj to bytes before comparison if sys.version_info[0] == 3 and isinstance(obj, str): _obj = obj.encode('utf-8') elif sys.version_info[0] == 2 and isinstance(obj, unicode): _obj = bytes(obj) else: _obj = obj self.assertTrue(isinstance(unpacked, type(_obj))) self.assertEqual(unpacked, _obj) umsgpack.compatibility = False def test_unpack_invalid_string(self): # Use last unpack exception test vector (an invalid string) (_, data, _) = unpack_exception_test_vectors[-1] obj = umsgpack.unpackb(data, allow_invalid_utf8=True) self.assertTrue(isinstance(obj, umsgpack.InvalidString)) self.assertEqual(obj, b"\x80") def test_unpack_ordered_dict(self): # Use last composite test vector (a map) (_, obj, data) = composite_test_vectors[-1] # Unpack with default options (unordered dict) unpacked = umsgpack.unpackb(data) self.assertTrue(isinstance(unpacked, dict)) # Unpack with unordered dict unpacked = umsgpack.unpackb(data, use_ordered_dict=False) self.assertTrue(isinstance(unpacked, dict)) # Unpack with ordered dict unpacked = umsgpack.unpackb(data, use_ordered_dict=True) self.assertTrue(isinstance(unpacked, OrderedDict)) self.assertEqual(unpacked, obj) def test_ext_exceptions(self): with self.assertRaises(TypeError): _ = umsgpack.Ext(-1, b"") with self.assertRaises(TypeError): _ = umsgpack.Ext(128, b"") with self.assertRaises(TypeError): _ = umsgpack.Ext(0, u"unicode string") def test_pack_ext_handler(self): for (name, obj, data) in ext_handlers_test_vectors: obj_repr = repr(obj) print("\tTesting %s: object %s" % (name, obj_repr if len(obj_repr) < 24 else obj_repr[0:24] + "...")) self.assertEqual(umsgpack.packb(obj, ext_handlers=ext_handlers), data) def test_unpack_ext_handler(self): for (name, obj, data) in ext_handlers_test_vectors: obj_repr = repr(obj) print("\tTesting %s: object %s" % (name, obj_repr if len(obj_repr) < 24 else obj_repr[0:24] + "...")) self.assertEqual(umsgpack.unpackb(data, ext_handlers=ext_handlers), obj) def test_streaming_writer(self): # Try first composite test vector (_, obj, data) = composite_test_vectors[0] writer = io.BytesIO() umsgpack.pack(obj, writer) self.assertTrue(writer.getvalue(), data) def test_streaming_reader(self): # Try first composite test vector (_, obj, data) = composite_test_vectors[0] reader = io.BytesIO(data) self.assertEqual(umsgpack.unpack(reader), obj) def test_namespacing(self): # Get a list of global variables from umsgpack module exported_vars = list(filter(lambda x: not x.startswith("_"), dir(umsgpack))) # Ignore imports exported_vars = list(filter(lambda x: x != "struct" and x != "collections" and x != "sys" and x != "io" and x != "xrange", exported_vars)) self.assertTrue(len(exported_vars) == len(exported_vars_test_vector)) for var in exported_vars_test_vector: self.assertTrue(var in exported_vars) if __name__ == '__main__': unittest.main()