| # Lint as: python3 | |
| # Copyright 2020 Google Inc. All rights reserved. | |
| # | |
| # Licensed under the Apache License, Version 2.0 (the "License"); | |
| # you may not use this file except in compliance with the License. | |
| # You may obtain a copy of the License at | |
| # | |
| # http://www.apache.org/licenses/LICENSE-2.0 | |
| # | |
| # Unless required by applicable law or agreed to in writing, software | |
| # distributed under the License is distributed on an "AS IS" BASIS, | |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | |
| # See the License for the specific language governing permissions and | |
| # limitations under the License. | |
| """Unit tests for flexbuffers.py.""" | |
| import array | |
| import os.path | |
| import struct | |
| import unittest | |
| from flatbuffers import flexbuffers | |
| Type = flexbuffers.Type | |
| LOG2 = {1: 0, 2: 1, 4: 2, 8: 3} | |
| GOLD_FLEXBUFFER_OBJ = { | |
| 'bar': [1, 2, 3], | |
| 'bar3': [1, 2, 3], | |
| 'bool': True, | |
| 'bools': [True, False, True, False], | |
| 'foo': 100.0, | |
| 'mymap': {'foo': 'Fred'}, | |
| 'vec': [-100, 'Fred', 4.0, b'M', False, 4.0], | |
| } | |
| GOLD_FLEXBUFFER_FILE = 'gold_flexbuffer_example.bin' | |
| def read_test_file(name): | |
| with open(os.path.join(os.path.dirname(__file__), name), 'rb') as f: | |
| return f.read() | |
| def packed_type(type_, i): | |
| return (type_ << 2) | LOG2[i] | |
| def uint_size(value): | |
| """Returns number of bytes (power of two) to represent unsigned value.""" | |
| assert value >= 0 | |
| n = 8 | |
| while not value < (1 << n): | |
| n *= 2 | |
| return n // 8 | |
| def int_size(value): | |
| """Returns number of bytes (power of two) to represent signed value.""" | |
| n = 8 | |
| while not -(1 << (n - 1)) <= value < (1 << (n - 1)): | |
| n *= 2 | |
| return n // 8 | |
| def uint_sizes(value): | |
| return tuple(1 << i for i in range(LOG2[uint_size(value)], 4)) | |
| def int_sizes(value): | |
| return tuple(1 << i for i in range(LOG2[int_size(value)], 4)) | |
| def int_bytes(value, byte_width): | |
| return struct.pack( | |
| '<%s' % {1: 'b', 2: 'h', 4: 'i', 8: 'q'}[byte_width], value | |
| ) | |
| def uint_bytes(value, byte_width): | |
| return struct.pack( | |
| '<%s' % {1: 'B', 2: 'H', 4: 'I', 8: 'Q'}[byte_width], value | |
| ) | |
| def float_bytes(value, byte_width): | |
| return struct.pack('<%s' % {4: 'f', 8: 'd'}[byte_width], value) | |
| def min_value(type_, byte_width): | |
| assert byte_width > 0 | |
| if type_ in (Type.INT, Type.INDIRECT_INT): | |
| return -(1 << (8 * byte_width - 1)) | |
| elif type_ in (Type.UINT, Type.INDIRECT_UINT): | |
| return 0 | |
| else: | |
| raise ValueError('Unsupported type %s' % type_) | |
| def max_value(type_, byte_width): | |
| assert byte_width > 0 | |
| if type_ in (Type.INT, Type.INDIRECT_INT): | |
| return (1 << (8 * byte_width - 1)) - 1 | |
| elif type_ in (Type.UINT, Type.INDIRECT_UINT): | |
| return (1 << 8 * byte_width) - 1 | |
| else: | |
| raise ValueError('Unsupported type %s' % type_) | |
| def str_bytes(value, byte_width): | |
| value_bytes = value.encode('utf-8') | |
| return [*uint_bytes(len(value_bytes), byte_width), *value_bytes, 0] | |
| def key_bytes(value): | |
| return [*value.encode('ascii'), 0] | |
| def encode_type(type_, value, byte_width=None): | |
| fbb = flexbuffers.Builder() | |
| add = fbb.Adder(type_) | |
| if byte_width: | |
| add(value, byte_width) | |
| else: | |
| add(value) | |
| return fbb.Finish() | |
| INT_MIN_MAX_VALUES = ( | |
| min_value(Type.INT, 1), | |
| max_value(Type.INT, 1), | |
| min_value(Type.INT, 2), | |
| max_value(Type.INT, 2), | |
| min_value(Type.INT, 4), | |
| max_value(Type.INT, 4), | |
| min_value(Type.INT, 8), | |
| max_value(Type.INT, 8), | |
| ) | |
| UINT_MIN_MAX_VALUES = ( | |
| 0, | |
| max_value(Type.UINT, 1), | |
| max_value(Type.UINT, 2), | |
| max_value(Type.UINT, 4), | |
| max_value(Type.UINT, 8), | |
| ) | |
| class UtilTest(unittest.TestCase): | |
| """Tests to check FlexBuffer utility functions.""" | |
| def _test_type_predicate(self, pred, types): | |
| for type_ in types: | |
| with self.subTest(type=type_, pred=pred): | |
| self.assertTrue(pred(type_)) | |
| for type_ in set(Type).difference(types): | |
| with self.subTest(type=type_, pred=pred): | |
| self.assertFalse(pred(type_)) | |
| def test_inline_types(self): | |
| self._test_type_predicate( | |
| Type.IsInline, (Type.NULL, Type.INT, Type.UINT, Type.FLOAT, Type.BOOL) | |
| ) | |
| def test_typed_vector(self): | |
| self._test_type_predicate( | |
| Type.IsTypedVector, | |
| ( | |
| Type.VECTOR_INT, | |
| Type.VECTOR_UINT, | |
| Type.VECTOR_FLOAT, | |
| Type.VECTOR_KEY, | |
| Type.VECTOR_STRING_DEPRECATED, | |
| Type.VECTOR_BOOL, | |
| ), | |
| ) | |
| self._test_type_predicate( | |
| Type.IsTypedVectorElementType, | |
| (Type.INT, Type.UINT, Type.FLOAT, Type.KEY, Type.STRING, Type.BOOL), | |
| ) | |
| with self.assertRaises(ValueError): | |
| Type.ToTypedVectorElementType(Type.VECTOR) | |
| self.assertIs(Type.ToTypedVectorElementType(Type.VECTOR_INT), Type.INT) | |
| self.assertIs(Type.ToTypedVectorElementType(Type.VECTOR_UINT), Type.UINT) | |
| self.assertIs(Type.ToTypedVectorElementType(Type.VECTOR_FLOAT), Type.FLOAT) | |
| self.assertIs(Type.ToTypedVectorElementType(Type.VECTOR_KEY), Type.KEY) | |
| self.assertIs( | |
| Type.ToTypedVectorElementType(Type.VECTOR_STRING_DEPRECATED), | |
| Type.STRING, | |
| ) | |
| self.assertIs(Type.ToTypedVectorElementType(Type.VECTOR_BOOL), Type.BOOL) | |
| with self.assertRaises(ValueError): | |
| Type.ToTypedVector(Type.VECTOR) | |
| self.assertIs(Type.ToTypedVector(Type.INT), Type.VECTOR_INT) | |
| self.assertIs(Type.ToTypedVector(Type.UINT), Type.VECTOR_UINT) | |
| self.assertIs(Type.ToTypedVector(Type.FLOAT), Type.VECTOR_FLOAT) | |
| self.assertIs(Type.ToTypedVector(Type.KEY), Type.VECTOR_KEY) | |
| self.assertIs( | |
| Type.ToTypedVector(Type.STRING), Type.VECTOR_STRING_DEPRECATED | |
| ) | |
| self.assertIs(Type.ToTypedVector(Type.BOOL), Type.VECTOR_BOOL) | |
| def test_fixed_typed_vector(self): | |
| self._test_type_predicate( | |
| Type.IsFixedTypedVector, | |
| ( | |
| Type.VECTOR_INT2, | |
| Type.VECTOR_UINT2, | |
| Type.VECTOR_FLOAT2, | |
| Type.VECTOR_INT3, | |
| Type.VECTOR_UINT3, | |
| Type.VECTOR_FLOAT3, | |
| Type.VECTOR_INT4, | |
| Type.VECTOR_UINT4, | |
| Type.VECTOR_FLOAT4, | |
| ), | |
| ) | |
| self._test_type_predicate( | |
| Type.IsFixedTypedVectorElementType, (Type.INT, Type.UINT, Type.FLOAT) | |
| ) | |
| self.assertEqual( | |
| Type.ToFixedTypedVectorElementType(Type.VECTOR_INT2), (Type.INT, 2) | |
| ) | |
| self.assertEqual( | |
| Type.ToFixedTypedVectorElementType(Type.VECTOR_UINT2), (Type.UINT, 2) | |
| ) | |
| self.assertEqual( | |
| Type.ToFixedTypedVectorElementType(Type.VECTOR_FLOAT2), (Type.FLOAT, 2) | |
| ) | |
| self.assertEqual( | |
| Type.ToFixedTypedVectorElementType(Type.VECTOR_INT3), (Type.INT, 3) | |
| ) | |
| self.assertEqual( | |
| Type.ToFixedTypedVectorElementType(Type.VECTOR_UINT3), (Type.UINT, 3) | |
| ) | |
| self.assertEqual( | |
| Type.ToFixedTypedVectorElementType(Type.VECTOR_FLOAT3), (Type.FLOAT, 3) | |
| ) | |
| self.assertEqual( | |
| Type.ToFixedTypedVectorElementType(Type.VECTOR_INT4), (Type.INT, 4) | |
| ) | |
| self.assertEqual( | |
| Type.ToFixedTypedVectorElementType(Type.VECTOR_UINT4), (Type.UINT, 4) | |
| ) | |
| self.assertEqual( | |
| Type.ToFixedTypedVectorElementType(Type.VECTOR_FLOAT4), (Type.FLOAT, 4) | |
| ) | |
| # Invalid size | |
| for type_ in Type.INT, Type.UINT, Type.FLOAT: | |
| with self.assertRaises(ValueError): | |
| Type.ToTypedVector(type_, 1) | |
| with self.assertRaises(ValueError): | |
| Type.ToTypedVector(type_, 5) | |
| # Invalid element type | |
| for length in 1, 2, 3, 4, 5: | |
| with self.assertRaises(ValueError): | |
| Type.ToTypedVector(Type.STRING, length) | |
| self.assertIs(Type.ToTypedVector(Type.INT, 2), Type.VECTOR_INT2) | |
| self.assertIs(Type.ToTypedVector(Type.INT, 3), Type.VECTOR_INT3) | |
| self.assertIs(Type.ToTypedVector(Type.INT, 4), Type.VECTOR_INT4) | |
| self.assertIs(Type.ToTypedVector(Type.UINT, 2), Type.VECTOR_UINT2) | |
| self.assertIs(Type.ToTypedVector(Type.UINT, 3), Type.VECTOR_UINT3) | |
| self.assertIs(Type.ToTypedVector(Type.UINT, 4), Type.VECTOR_UINT4) | |
| self.assertIs(Type.ToTypedVector(Type.FLOAT, 2), Type.VECTOR_FLOAT2) | |
| self.assertIs(Type.ToTypedVector(Type.FLOAT, 3), Type.VECTOR_FLOAT3) | |
| self.assertIs(Type.ToTypedVector(Type.FLOAT, 4), Type.VECTOR_FLOAT4) | |
| def test_width(self): | |
| for x in range(1 << 10): | |
| self.assertEqual(flexbuffers.BitWidth.U(x), LOG2[uint_size(x)]) | |
| for x in range(-(1 << 10), 1 << 10): | |
| self.assertEqual(flexbuffers.BitWidth.I(x), LOG2[int_size(x)]) | |
| def test_padding(self): | |
| self.assertEqual(flexbuffers._PaddingBytes(0, 4), 0) | |
| self.assertEqual(flexbuffers._PaddingBytes(0, 8), 0) | |
| self.assertEqual(flexbuffers._PaddingBytes(0, 16), 0) | |
| self.assertEqual(flexbuffers._PaddingBytes(1, 8), 7) | |
| self.assertEqual(flexbuffers._PaddingBytes(17, 8), 7) | |
| self.assertEqual(flexbuffers._PaddingBytes(42, 2), 0) | |
| class DecoderTest(unittest.TestCase): | |
| """Tests to check FlexBuffer decoding functions. | |
| Common variable names used in the tests for compactness: | |
| bw: byte_width | |
| ebw: element_byte_width | |
| kbw: key_byte_width | |
| vbw: value_byte_width | |
| tbw: type_byte_width | |
| Having '_ignored' suffix means that variable doesn't affect the constructed | |
| byte buffer size. | |
| """ | |
| def test_null(self): | |
| for bw in 1, 2, 4, 8: | |
| for ebw_ignored in 1, 2, 4, 8: | |
| with self.subTest(bw=bw, ebw_ignored=ebw_ignored): | |
| data = bytes([ | |
| *uint_bytes(0, bw), | |
| packed_type(Type.NULL, ebw_ignored), | |
| bw, | |
| ]) | |
| root = flexbuffers.GetRoot(data) | |
| self.assertTrue(root.IsNull) | |
| self.assertEqual(root.AsBool, False) | |
| self.assertEqual(root.AsInt, 0) | |
| self.assertEqual(root.AsFloat, 0.0) | |
| for prop in ( | |
| type(root).AsKey, | |
| type(root).AsString, | |
| type(root).AsBlob, | |
| type(root).AsVector, | |
| type(root).AsTypedVector, | |
| type(root).AsFixedTypedVector, | |
| type(root).AsMap, | |
| ): | |
| with self.assertRaises(TypeError): | |
| prop.fget(root) | |
| self.assertEqual(root.Value, None) | |
| self.assertIsNone(flexbuffers.Loads(data)) | |
| def test_bool(self): | |
| for value in False, True: | |
| for bw in 1, 2, 4, 8: | |
| for ebw_ignored in 1, 2, 4, 8: | |
| with self.subTest(bw=bw, ebw_ignored=ebw_ignored): | |
| data = bytes([ | |
| *uint_bytes(int(value), bw), | |
| packed_type(Type.BOOL, ebw_ignored), | |
| bw, | |
| ]) | |
| root = flexbuffers.GetRoot(data) | |
| self.assertTrue(root.IsBool) | |
| self.assertEqual(root.AsBool, value) | |
| self.assertEqual(root.AsInt, int(value)) | |
| self.assertEqual(root.AsFloat, float(value)) | |
| for prop in ( | |
| type(root).AsKey, | |
| type(root).AsString, | |
| type(root).AsBlob, | |
| type(root).AsVector, | |
| type(root).AsTypedVector, | |
| type(root).AsFixedTypedVector, | |
| type(root).AsMap, | |
| ): | |
| with self.assertRaises(TypeError): | |
| prop.fget(root) | |
| self.assertEqual(root.Value, value) | |
| self.assertEqual(flexbuffers.Loads(data), value) | |
| def test_mutate_bool(self): | |
| root = flexbuffers.GetRoot(flexbuffers.Dumps(True)) | |
| self.assertTrue(root.IsBool) | |
| self.assertTrue(root.AsBool) | |
| self.assertTrue(root.MutateBool(False)) | |
| self.assertTrue(root.IsBool) | |
| self.assertFalse(root.AsBool) | |
| self.assertTrue(root.MutateBool(True)) | |
| self.assertTrue(root.IsBool) | |
| self.assertTrue(root.AsBool) | |
| def _check_int(self, data, value): | |
| root = flexbuffers.GetRoot(data) | |
| self.assertTrue(root.IsInt) | |
| self.assertEqual(root.AsInt, value) | |
| self.assertEqual(root.AsBool, bool(value)) | |
| self.assertEqual(root.AsFloat, float(value)) | |
| for prop in ( | |
| type(root).AsKey, | |
| type(root).AsString, | |
| type(root).AsBlob, | |
| type(root).AsVector, | |
| type(root).AsTypedVector, | |
| type(root).AsFixedTypedVector, | |
| type(root).AsMap, | |
| ): | |
| with self.assertRaises(TypeError): | |
| prop.fget(root) | |
| self.assertEqual(root.Value, value) | |
| self.assertEqual(flexbuffers.Loads(data), value) | |
| def test_int(self): | |
| for value in (0, 1, -1, 15, -17, *INT_MIN_MAX_VALUES): | |
| for bw in int_sizes(value): | |
| for ebw_ignored in 1, 2, 4, 8: | |
| with self.subTest(value=value, bw=bw, ebw_ignored=ebw_ignored): | |
| data = bytes([ | |
| *int_bytes(value, bw), | |
| packed_type(Type.INT, ebw_ignored), | |
| bw, | |
| ]) | |
| self._check_int(data, value) | |
| def test_indirect_int(self): | |
| for value in (0, 1, -1, 15, -17, *INT_MIN_MAX_VALUES): | |
| for bw in 1, 2, 4, 8: | |
| for ebw in int_sizes(value): | |
| with self.subTest(value=value, bw=bw, ebw=ebw): | |
| data = bytes([ | |
| # Int | |
| *int_bytes(value, ebw), | |
| # Root | |
| *uint_bytes(ebw, bw), | |
| packed_type(Type.INDIRECT_INT, ebw), | |
| bw, | |
| ]) | |
| self._check_int(data, value) | |
| def test_uint(self): | |
| for value in (1, *UINT_MIN_MAX_VALUES): | |
| for bw in uint_sizes(value): | |
| for ebw_ignored in 1, 2, 4, 8: | |
| with self.subTest(value=value, bw=bw, ebw_ignored=ebw_ignored): | |
| data = bytes([ | |
| *uint_bytes(value, bw), | |
| packed_type(Type.UINT, ebw_ignored), | |
| bw, | |
| ]) | |
| self._check_int(data, value) | |
| def test_inidirect_uint(self): | |
| for value in (1, *UINT_MIN_MAX_VALUES): | |
| for bw in 1, 2, 4, 8: | |
| for ebw in uint_sizes(value): | |
| with self.subTest(value=value, bw=bw, ebw=ebw): | |
| data = bytes([ | |
| # UInt | |
| *uint_bytes(value, ebw), | |
| # Root | |
| *uint_bytes(ebw, bw), | |
| packed_type(Type.INDIRECT_UINT, ebw), | |
| bw, | |
| ]) | |
| self._check_int(data, value) | |
| def test_mutate_ints(self): | |
| # Signed | |
| for type_ in Type.INT, Type.INDIRECT_INT: | |
| with self.subTest(type=type_): | |
| root = flexbuffers.GetRoot(encode_type(type_, 56)) | |
| self.assertEqual(root.AsInt, 56) | |
| for new_value in 0, 1, -1, -128, 127: | |
| self.assertTrue(root.MutateInt(new_value)) | |
| self.assertEqual(root.AsInt, new_value) | |
| for new_value in -129, 128: | |
| self.assertFalse(root.MutateInt(new_value)) | |
| # Unsigned | |
| for type_ in Type.UINT, Type.INDIRECT_UINT: | |
| with self.subTest(type=type_): | |
| root = flexbuffers.GetRoot(encode_type(type_, 1)) | |
| self.assertEqual(root.AsInt, 1) | |
| for new_value in 0, 1, 255: | |
| self.assertTrue(root.MutateInt(new_value)) | |
| self.assertEqual(root.AsInt, new_value) | |
| self.assertFalse(root.MutateInt(256)) | |
| # Inside vector | |
| fbb = flexbuffers.Builder() | |
| fbb.VectorFromElements([13, 0, -15]) | |
| data = fbb.Finish() | |
| self.assertEqual(flexbuffers.Loads(data), [13, 0, -15]) | |
| self.assertTrue(flexbuffers.GetRoot(data).AsVector[0].MutateInt(0)) | |
| self.assertTrue(flexbuffers.GetRoot(data).AsVector[1].MutateInt(-7)) | |
| self.assertTrue(flexbuffers.GetRoot(data).AsVector[2].MutateInt(45)) | |
| self.assertEqual(flexbuffers.Loads(data), [0, -7, 45]) | |
| # Inside map | |
| fbb = flexbuffers.Builder() | |
| fbb.MapFromElements({'x': -7, 'y': 46}) | |
| data = fbb.Finish() | |
| self.assertEqual(flexbuffers.Loads(data), {'x': -7, 'y': 46}) | |
| self.assertTrue(flexbuffers.GetRoot(data).AsMap['x'].MutateInt(14)) | |
| self.assertTrue(flexbuffers.GetRoot(data).AsMap['y'].MutateInt(-1)) | |
| self.assertEqual(flexbuffers.Loads(data), {'x': 14, 'y': -1}) | |
| def _check_float(self, data, value): | |
| root = flexbuffers.GetRoot(data) | |
| self.assertTrue(root.IsFloat) | |
| self.assertAlmostEqual(root.AsFloat, value) | |
| for prop in ( | |
| type(root).AsKey, | |
| type(root).AsString, | |
| type(root).AsBlob, | |
| type(root).AsVector, | |
| type(root).AsTypedVector, | |
| type(root).AsFixedTypedVector, | |
| type(root).AsMap, | |
| ): | |
| with self.assertRaises(TypeError): | |
| prop.fget(root) | |
| self.assertAlmostEqual(root.Value, value) | |
| self.assertAlmostEqual(flexbuffers.Loads(data), value) | |
| def test_float(self): | |
| for value in -1.0, 0.0, 1.0, 3.141592, 1.5e6: | |
| for bw in 4, 8: | |
| for ebw_ignored in 1, 2, 4, 8: | |
| with self.subTest(value=value, bw=bw, ebw_ignored=ebw_ignored): | |
| data = bytes([ | |
| *float_bytes(value, bw), | |
| packed_type(Type.FLOAT, ebw_ignored), | |
| bw, | |
| ]) | |
| self._check_float(data, value) | |
| def test_indirect_float(self): | |
| for value in -1.0, 0.0, 1.0, 3.141592, 1.5e6: | |
| for bw in 1, 2, 4, 8: | |
| for ebw in 4, 8: | |
| with self.subTest(value=value, bw=bw, ebw=ebw): | |
| data = bytes([ | |
| # Float | |
| *float_bytes(value, ebw), | |
| # Root | |
| *uint_bytes(ebw, bw), | |
| packed_type(Type.INDIRECT_FLOAT, ebw), | |
| bw, | |
| ]) | |
| self._check_float(data, value) | |
| def test_mutate_float(self): | |
| for type_ in Type.FLOAT, Type.INDIRECT_FLOAT: | |
| for bw in 4, 8: | |
| value = 3.141592 | |
| root = flexbuffers.GetRoot(encode_type(type_, value, bw)) | |
| self.assertAlmostEqual(root.AsFloat, value) | |
| value = 2.71828 | |
| self.assertTrue(root.MutateFloat(value)) | |
| self.assertAlmostEqual(root.AsFloat, value, places=5) | |
| # Inside vector | |
| data = flexbuffers.Dumps([2.4, 1.5, -7.2]) | |
| self.assertTrue(flexbuffers.GetRoot(data).AsVector[0].MutateFloat(0.0)) | |
| self.assertTrue(flexbuffers.GetRoot(data).AsVector[1].MutateFloat(15.2)) | |
| self.assertTrue(flexbuffers.GetRoot(data).AsVector[2].MutateFloat(-5.1)) | |
| for a, b in zip(flexbuffers.Loads(data), [0.0, 15.2, -5.1]): | |
| self.assertAlmostEqual(a, b) | |
| def test_string(self): | |
| for value in 'red', 'green', 'blue', 'flatbuffers + flexbuffers': | |
| value_bytes = value.encode('utf-8') | |
| for bw in 1, 2, 4, 8: | |
| for lbw in 1, 2, 4, 8: | |
| with self.subTest(bw=bw, lbw=lbw): | |
| data = bytes([ | |
| # String | |
| *uint_bytes(len(value_bytes), lbw), | |
| *value_bytes, | |
| 0, | |
| # Root | |
| *uint_bytes(len(value_bytes) + 1, bw), # offset | |
| packed_type(Type.STRING, lbw), | |
| bw, | |
| ]) | |
| root = flexbuffers.GetRoot(data) | |
| self.assertTrue(root.IsString) | |
| self.assertEqual(root.AsString, value) | |
| self.assertEqual(root.Value, value) | |
| self.assertEqual(root.AsInt, len(value)) | |
| self.assertEqual(flexbuffers.Loads(data), value) | |
| def test_mutate_string(self): | |
| data = encode_type(Type.STRING, '12345') | |
| root = flexbuffers.GetRoot(data) | |
| self.assertTrue(root.IsString) | |
| self.assertEqual(root.AsString, '12345') | |
| self.assertFalse(root.MutateString('543210')) | |
| self.assertTrue(root.MutateString('54321')) | |
| self.assertTrue(root.IsString) | |
| self.assertEqual(root.AsString, '54321') | |
| self.assertTrue(root.MutateString('543')) | |
| self.assertTrue(root.IsString) | |
| self.assertEqual(root.AsString, '543') | |
| self.assertFalse(root.MutateString('54321')) | |
| def test_empty_blob(self): | |
| for bw in 1, 2, 4, 8: | |
| for lbw in 1, 2, 4, 8: | |
| with self.subTest(bw=bw, lbw=lbw): | |
| data = bytes([ | |
| # Blob | |
| *uint_bytes(0, lbw), | |
| # Root | |
| *uint_bytes(0, bw), | |
| packed_type(Type.BLOB, lbw), | |
| bw, | |
| ]) | |
| root = flexbuffers.GetRoot(data) | |
| self.assertTrue(root.IsBlob) | |
| self.assertEqual(root.AsBlob, bytes()) | |
| self.assertEqual(root.Value, bytes()) | |
| self.assertEqual(flexbuffers.Loads(data), bytes()) | |
| def test_blob(self): | |
| for blob in [], [215], [23, 75, 124, 0, 45, 15], 255 * [0]: | |
| for bw in 1, 2, 4, 8: | |
| for lbw in 1, 2, 4, 8: | |
| with self.subTest(blob=blob, bw=bw, lbw=lbw): | |
| data = bytes([ | |
| # Blob | |
| *uint_bytes(len(blob), lbw), | |
| *blob, | |
| # Root | |
| *uint_bytes(len(blob), bw), | |
| packed_type(Type.BLOB, lbw), | |
| bw, | |
| ]) | |
| root = flexbuffers.GetRoot(data) | |
| self.assertTrue(root.IsBlob) | |
| self.assertEqual(root.AsBlob, bytes(blob)) | |
| self.assertEqual(root.Value, bytes(blob)) | |
| self.assertEqual(flexbuffers.Loads(data), bytes(blob)) | |
| def test_key(self): | |
| for value in '', 'x', 'color': | |
| for bw in 1, 2, 4, 8: | |
| with self.subTest(value=value, bw=bw): | |
| value_bytes = value.encode('ascii') | |
| data = bytes([ | |
| # Key | |
| *value_bytes, | |
| 0, | |
| # Root | |
| *uint_bytes(len(value_bytes) + 1, bw), | |
| packed_type(Type.KEY, 1), | |
| bw, | |
| ]) | |
| root = flexbuffers.GetRoot(data) | |
| self.assertTrue(root.IsKey) | |
| self.assertEqual(root.AsKey, value) | |
| self.assertEqual(root.Value, value) | |
| self.assertEqual(flexbuffers.Loads(data), value) | |
| def _check_fixed_typed_vector(self, data, vector, type_): | |
| self.assertEqual(flexbuffers.Loads(data), vector) | |
| root = flexbuffers.GetRoot(data) | |
| self.assertTrue(root.IsFixedTypedVector) | |
| v = root.AsFixedTypedVector | |
| self.assertEqual(len(v), len(vector)) | |
| self.assertIs(v.ElementType, type_) | |
| self.assertEqual([e.Value for e in v], vector) | |
| self.assertSequenceEqual(v.Value, vector) | |
| self.assertEqual(root.AsInt, len(vector)) | |
| def test_fixed_typed_vector_float(self): | |
| for type_, vector in ( | |
| (Type.VECTOR_FLOAT2, [-75.0, 34.89]), | |
| (Type.VECTOR_FLOAT3, [-75.0, 34.89, 12.0]), | |
| (Type.VECTOR_FLOAT4, [-75.0, 34.89, -1.0, 1.0]), | |
| ): | |
| for bw in 1, 2, 4, 8: | |
| for ebw in 4, 8: | |
| with self.subTest(type=type_, vector=vector, bw=bw, ebw=ebw): | |
| data = bytes([ | |
| # FixedTypedVector | |
| *b''.join(float_bytes(e, ebw) for e in vector), | |
| # Root | |
| *uint_bytes(len(vector) * ebw, bw), | |
| packed_type(type_, ebw), | |
| bw, | |
| ]) | |
| for a, b in zip(flexbuffers.Loads(data), vector): | |
| self.assertAlmostEqual(a, b, places=2) | |
| def test_fixed_typed_vector_int(self): | |
| for type_, vector in ( | |
| (Type.VECTOR_INT2, [0, -13]), | |
| (Type.VECTOR_INT3, [127, 0, -13]), | |
| (Type.VECTOR_INT4, [127, 0, -13, 0]), | |
| ): | |
| for bw in 1, 2, 4, 8: | |
| for ebw in 1, 2, 4, 8: | |
| with self.subTest(type=type_, vector=vector, bw=bw, ebw=ebw): | |
| data = bytes([ | |
| # FixedTypeVector | |
| *b''.join(int_bytes(e, ebw) for e in vector), | |
| # Root | |
| *uint_bytes(ebw * len(vector), bw), | |
| packed_type(type_, ebw), | |
| bw, | |
| ]) | |
| self._check_fixed_typed_vector(data, vector, Type.INT) | |
| def test_fixed_typed_vector_uint(self): | |
| for type_, vector in ( | |
| (Type.VECTOR_UINT2, [0, 13]), | |
| (Type.VECTOR_UINT3, [127, 0, 13]), | |
| (Type.VECTOR_UINT4, [127, 0, 13, 0]), | |
| ): | |
| for bw in 1, 2, 4, 8: | |
| for ebw in 1, 2, 4, 8: | |
| with self.subTest(type=type_, vector=vector, bw=bw, ebw=ebw): | |
| data = bytes([ | |
| # FixedTypeVector | |
| *b''.join(uint_bytes(e, ebw) for e in vector), | |
| # Root | |
| *uint_bytes(ebw * len(vector), bw), | |
| packed_type(type_, ebw), | |
| bw, | |
| ]) | |
| self._check_fixed_typed_vector(data, vector, Type.UINT) | |
| def _check_typed_vector(self, data, vector, type_): | |
| self.assertEqual(flexbuffers.Loads(data), vector) | |
| root = flexbuffers.GetRoot(data) | |
| self.assertTrue(root.IsTypedVector) | |
| v = root.AsTypedVector | |
| self.assertIs(v.ElementType, type_) | |
| self.assertEqual(len(v), len(vector)) | |
| self.assertEqual([e.Value for e in v], vector) | |
| self.assertSequenceEqual(v.Value, vector) | |
| self.assertEqual(root.AsInt, len(vector)) | |
| def test_empty_typed_vector(self): | |
| for type_ in ( | |
| Type.VECTOR_BOOL, | |
| Type.VECTOR_INT, | |
| Type.VECTOR_UINT, | |
| Type.VECTOR_FLOAT, | |
| Type.VECTOR_KEY, | |
| Type.VECTOR_STRING_DEPRECATED, | |
| ): | |
| for bw in 1, 2, 4, 8: | |
| for ebw in 1, 2, 4, 8: | |
| with self.subTest(type=type_, bw=bw, ebw=ebw): | |
| data = bytes([ | |
| # TypedVector[type_] | |
| *uint_bytes(0, ebw), | |
| # Root | |
| *uint_bytes(0, bw), | |
| packed_type(type_, ebw), | |
| bw, | |
| ]) | |
| element_type = Type.ToTypedVectorElementType(type_) | |
| if element_type == Type.STRING: | |
| element_type = Type.KEY | |
| self._check_typed_vector(data, [], element_type) | |
| def test_typed_vector_bool(self): | |
| vector = [True, False, False, False, True] | |
| for bw in 1, 2, 4, 8: | |
| for ebw in 1, 2, 4, 8: | |
| with self.subTest(bw=bw, ebw=ebw): | |
| data = bytes([ | |
| # TypedVector[Type.BOOL] | |
| *uint_bytes(len(vector), ebw), | |
| *b''.join(uint_bytes(int(e), ebw) for e in vector), | |
| # Root | |
| *uint_bytes(len(vector) * ebw, bw), | |
| packed_type(Type.VECTOR_BOOL, ebw), | |
| bw, | |
| ]) | |
| self._check_typed_vector(data, vector, Type.BOOL) | |
| def test_typed_vector_int(self): | |
| vector = [-100, 200, -300] | |
| for bw in 1, 2, 4, 8: | |
| for ebw in 2, 4, 8: | |
| with self.subTest(bw=bw, ebw=ebw): | |
| data = bytes([ | |
| # TypedVector[Type.INT] | |
| *uint_bytes(len(vector), ebw), | |
| *b''.join(int_bytes(e, ebw) for e in vector), | |
| # Root | |
| *uint_bytes(len(vector) * ebw, bw), | |
| packed_type(Type.VECTOR_INT, ebw), | |
| bw, | |
| ]) | |
| self._check_typed_vector(data, vector, Type.INT) | |
| def test_typed_vector_uint(self): | |
| vector = [100, 200, 300, 400, 0] | |
| for bw in 1, 2, 4, 8: | |
| for ebw in 2, 4, 8: | |
| with self.subTest(bw=bw, ebw=ebw): | |
| data = bytes([ | |
| # TypedVector[Type.UINT] | |
| *uint_bytes(len(vector), ebw), | |
| *b''.join(int_bytes(e, ebw) for e in vector), | |
| # Root | |
| *uint_bytes(len(vector) * ebw, bw), | |
| packed_type(Type.VECTOR_UINT, ebw), | |
| bw, | |
| ]) | |
| self._check_typed_vector(data, vector, Type.UINT) | |
| def test_typed_vector_float(self): | |
| vector = [3.64, -6.36, 3.14, 634.0, -42.0] | |
| for bw in 1, 2, 4, 8: | |
| for ebw in 4, 8: | |
| with self.subTest(bw=bw, ebw=ebw): | |
| data = bytes([ | |
| # TypedVector[Type.FLOAT] | |
| *uint_bytes(len(vector), ebw), | |
| *b''.join(float_bytes(e, ebw) for e in vector), | |
| # Root | |
| *uint_bytes(ebw * len(vector), bw), | |
| packed_type(Type.VECTOR_FLOAT, ebw), | |
| bw, | |
| ]) | |
| for a, b in zip(flexbuffers.Loads(data), vector): | |
| self.assertAlmostEqual(a, b, places=2) | |
| def test_typed_vector_key(self): | |
| vector = ['red', 'green', 'blue'] | |
| for bw in 1, 2, 4, 8: | |
| for ebw in 1, 2, 4, 8: | |
| with self.subTest(bw=bw, ebw=ebw): | |
| data = bytes([ | |
| # Keys | |
| *key_bytes(vector[0]), | |
| *key_bytes(vector[1]), | |
| *key_bytes(vector[2]), | |
| # TypedVector[Type.KEY] | |
| *uint_bytes(len(vector), ebw), | |
| *uint_bytes(15 + 1 * ebw, ebw), # offset to vector[0] | |
| *uint_bytes(11 + 2 * ebw, ebw), # offset to vector[1] | |
| *uint_bytes(5 + 3 * ebw, ebw), # offset to vector[2] | |
| # Root | |
| *uint_bytes(len(vector) * ebw, bw), # offset to vector | |
| packed_type(Type.VECTOR_KEY, ebw), | |
| bw, | |
| ]) | |
| self._check_typed_vector(data, vector, Type.KEY) | |
| def test_typed_vector_string(self): | |
| vector = ['red', 'green', 'blue'] | |
| for bw in 1, 2, 4, 8: | |
| for ebw in 1, 2, 4, 8: | |
| with self.subTest(bw=bw, ebw=ebw): | |
| data = bytes([ | |
| # Strings | |
| *str_bytes(vector[0], 1), # 5 bytes | |
| *str_bytes(vector[1], 1), # 7 bytes | |
| *str_bytes(vector[2], 1), # 6 bytes | |
| # TypedVector[Type.STRING] | |
| *uint_bytes(len(vector), ebw), | |
| *uint_bytes(17 + 1 * ebw, ebw), # offset to vector[0] | |
| *uint_bytes(12 + 2 * ebw, ebw), # offset to vector[1] | |
| *uint_bytes(5 + 3 * ebw, ebw), # offset to vector[2] | |
| # Root | |
| *uint_bytes(len(vector) * ebw, bw), # offset to vector | |
| packed_type(Type.VECTOR_STRING_DEPRECATED, ebw), | |
| bw, | |
| ]) | |
| # We have to pass Type.KEY because of Type.VECTOR_STRING_DEPRECATED. | |
| self._check_typed_vector(data, vector, Type.KEY) | |
| def test_typed_vector_string_deprecated(self): | |
| # Check FlexBuffersDeprecatedTest() inside test.cpp for details. | |
| vector = [300 * 'A', 'test'] | |
| fbb = flexbuffers.Builder() | |
| with fbb.TypedVector(): | |
| for e in vector: | |
| fbb.String(e) | |
| data = fbb.Finish() | |
| # We have to pass Type.KEY because of Type.VECTOR_STRING_DEPRECATED. | |
| self._check_typed_vector(data, vector, Type.KEY) | |
| def test_typed_vector_invalid(self): | |
| fbb = flexbuffers.Builder() | |
| with self.assertRaises(RuntimeError): | |
| fbb.TypedVectorFromElements(['string', 423]) | |
| def test_empty_vector(self): | |
| for bw in 1, 2, 4, 8: | |
| for ebw in 1, 2, 4, 8: | |
| data = bytes([ | |
| *uint_bytes(0, ebw), | |
| # Root | |
| *uint_bytes(0, bw), | |
| packed_type(Type.VECTOR, ebw), | |
| bw, | |
| ]) | |
| root = flexbuffers.GetRoot(data) | |
| self.assertTrue(root.IsVector) | |
| self.assertEqual(len(root.AsVector), 0) | |
| self.assertEqual(flexbuffers.Loads(data), []) | |
| def test_vector1(self): | |
| vector = [300, 400, 500] | |
| for bw in 1, 2, 4, 8: | |
| for ebw in 2, 4, 8: | |
| for tbw_ignored in 1, 2, 4, 8: | |
| with self.subTest(bw=bw, ebw=ebw, ignore=tbw_ignored): | |
| data = bytes([ | |
| # Vector length | |
| *uint_bytes(len(vector), ebw), | |
| # Vector elements | |
| *int_bytes(vector[0], ebw), | |
| *int_bytes(vector[1], ebw), | |
| *int_bytes(vector[2], ebw), | |
| # Vector types | |
| packed_type(Type.INT, tbw_ignored), | |
| packed_type(Type.INT, tbw_ignored), | |
| packed_type(Type.INT, tbw_ignored), | |
| # Root | |
| *uint_bytes(ebw * len(vector) + len(vector), bw), | |
| packed_type(Type.VECTOR, ebw), | |
| bw, | |
| ]) | |
| root = flexbuffers.GetRoot(data) | |
| self.assertTrue(root.IsVector) | |
| self.assertFalse(root.IsMap) | |
| v = root.AsVector | |
| self.assertEqual(len(v), len(vector)) | |
| for i in range(len(v)): | |
| self.assertTrue(v[i].IsInt) | |
| self.assertEqual(v[i].AsInt, vector[i]) | |
| for i, e in enumerate(v): | |
| self.assertTrue(e.IsInt) | |
| self.assertEqual(e.AsInt, vector[i]) | |
| with self.assertRaises(IndexError): | |
| v[-1].AsInt # pylint: disable=pointless-statement | |
| with self.assertRaises(IndexError): | |
| v[3].AsInt # pylint: disable=pointless-statement | |
| with self.assertRaises(TypeError): | |
| root.AsMap # pylint: disable=pointless-statement | |
| self.assertEqual(root.AsInt, len(vector)) | |
| self.assertEqual(root.AsFloat, float(len(vector))) | |
| self.assertEqual(flexbuffers.Loads(data), vector) | |
| def test_vector2(self): | |
| vector = [1984, 'August', True] | |
| for bw in 1, 2, 4, 8: | |
| with self.subTest(bw=bw): | |
| data = bytes([ | |
| *str_bytes(vector[1], 1), | |
| # Vector | |
| *uint_bytes(len(vector), 2), | |
| *int_bytes(vector[0], 2), | |
| *uint_bytes(11, 2), # offset to 'August' | |
| *uint_bytes(int(vector[2]), 2), | |
| packed_type(Type.INT, 2), | |
| packed_type(Type.STRING, 1), | |
| packed_type(Type.BOOL, 2), | |
| # Root | |
| *uint_bytes(2 * len(vector) + len(vector), bw), # offset to vector | |
| packed_type(Type.VECTOR, 2), | |
| bw, | |
| ]) | |
| self.assertEqual(flexbuffers.Loads(data), vector) | |
| root = flexbuffers.GetRoot(data) | |
| self.assertTrue(root.IsVector) | |
| v = root.AsVector | |
| self.assertTrue(v[0].IsInt) | |
| self.assertEqual(v[0].AsInt, 1984) | |
| self.assertTrue(v[1].IsString) | |
| self.assertEqual(v[1].AsString, 'August') | |
| self.assertTrue(v[2].IsBool) | |
| self.assertTrue(v[2].AsBool) | |
| self.assertEqual(v.Value, vector) | |
| self.assertEqual(root.AsInt, len(vector)) | |
| def test_empty_map(self): | |
| for bw in 1, 2, 4, 8: | |
| for kbw in 1, 2, 4, 8: | |
| for vbw in 1, 2, 4, 8: | |
| data = bytes([ | |
| *uint_bytes(0, kbw), # Keys length | |
| *uint_bytes(0, vbw), | |
| *uint_bytes(kbw, vbw), | |
| *uint_bytes(0, vbw), # Values length | |
| # Root | |
| *uint_bytes(0, bw), | |
| packed_type(Type.MAP, vbw), | |
| bw, | |
| ]) | |
| root = flexbuffers.GetRoot(data) | |
| self.assertTrue(root.IsMap) | |
| self.assertEqual(len(root.AsMap), 0) | |
| self.assertEqual(flexbuffers.Loads(data), {}) | |
| def test_map(self): | |
| value = {'foo': 13, 'bar': 14} | |
| for bw in 1, 2, 4, 8: | |
| for kbw in 1, 2, 4, 8: | |
| for vbw in 1, 2, 4, 8: | |
| with self.subTest(kbw=kbw, vbw=vbw, bw=bw): | |
| data = bytes([ | |
| *key_bytes('foo'), # 4 bytes | |
| *key_bytes('bar'), # 4 bytes | |
| # Map | |
| *uint_bytes(len(value), kbw), | |
| *uint_bytes(4 + 1 * kbw, kbw), # offset to 'bar' | |
| *uint_bytes(8 + 2 * kbw, kbw), # offset to 'foo' | |
| *uint_bytes(len(value) * kbw, vbw), # offset to keys | |
| *uint_bytes(kbw, vbw), | |
| *uint_bytes(len(value), vbw), | |
| *int_bytes(value['bar'], vbw), | |
| *int_bytes(value['foo'], vbw), | |
| packed_type(Type.INT, vbw), | |
| packed_type(Type.INT, vbw), | |
| # Root | |
| *uint_bytes( | |
| vbw * len(value) + len(value), bw | |
| ), # offset to values | |
| packed_type(Type.MAP, vbw), | |
| bw, | |
| ]) | |
| root = flexbuffers.GetRoot(data) | |
| self.assertTrue(root.IsMap) | |
| m = root.AsMap | |
| self.assertEqual(len(m), 2) | |
| self.assertEqual(m[0].AsInt, 14) | |
| self.assertEqual(m[1].AsInt, 13) | |
| self.assertEqual(m['bar'].AsInt, 14) | |
| self.assertEqual(m['foo'].AsInt, 13) | |
| for invalid_key in 'a', 'b', 'no': | |
| with self.assertRaises(KeyError): | |
| m[invalid_key] # pylint: disable=pointless-statement | |
| values = m.Values | |
| self.assertEqual(len(values), 2) | |
| self.assertEqual(values[0].AsInt, 14) | |
| self.assertEqual(values[1].AsInt, 13) | |
| keys = m.Keys | |
| self.assertEqual(len(keys), 2) | |
| self.assertEqual(len(keys[0].AsKey), 3) | |
| self.assertEqual(keys[0].AsKey, 'bar') | |
| self.assertEqual(len(keys[1].AsKey), 3) | |
| self.assertEqual(keys[1].AsKey, 'foo') | |
| keys = [key.AsKey for key in keys] | |
| self.assertEqual(sorted(keys), keys) | |
| self.assertEqual(root.AsInt, len(value)) | |
| self.assertEqual(flexbuffers.Loads(data), value) | |
| def test_alignment(self): | |
| value = ['test', 7] | |
| data = bytes([ | |
| *key_bytes('test'), # 5 bytes: 'test' and \0 | |
| 0, | |
| 0, | |
| 0, # 3 bytes: alignment | |
| # Vector | |
| *uint_bytes(len(value), byte_width=8), | |
| *uint_bytes(16, byte_width=8), | |
| *uint_bytes(7, byte_width=8), | |
| packed_type(Type.KEY, 1), | |
| packed_type(Type.INT, 8), | |
| # Root | |
| *uint_bytes(8 * len(value) + len(value), 1), | |
| packed_type(Type.VECTOR, 8), | |
| 1, | |
| ]) | |
| self.assertEqual(flexbuffers.Loads(data), value) | |
| class EncoderTest(unittest.TestCase): | |
| """Tests to check FlexBuffer encoding functions.""" | |
| def test_null(self): | |
| def encode_null(): | |
| fbb = flexbuffers.Builder() | |
| fbb.Null() | |
| return fbb.Finish() | |
| self.assertIsNone(flexbuffers.Loads(encode_null())) | |
| def test_bool(self): | |
| for value in False, True: | |
| data = encode_type(Type.BOOL, value) | |
| self.assertEqual(flexbuffers.Loads(data), value) | |
| def test_int(self): | |
| for byte_width in 1, 2, 4, 8: | |
| for type_ in Type.INT, Type.INDIRECT_INT, Type.UINT, Type.INDIRECT_UINT: | |
| with self.subTest(byte_width=byte_width, type=type_): | |
| value = min_value(type_, byte_width) | |
| data = encode_type(type_, value) | |
| self.assertEqual(flexbuffers.Loads(data), value) | |
| value = max_value(type_, byte_width) | |
| data = encode_type(type_, value) | |
| self.assertEqual(flexbuffers.Loads(data), value) | |
| def test_float(self): | |
| for value in 3.141592, 7.62, 999.99: | |
| for type_ in Type.FLOAT, Type.INDIRECT_FLOAT: | |
| with self.subTest(value=value, type=type_): | |
| data = encode_type(type_, value) | |
| self.assertEqual(flexbuffers.Loads(data), value) | |
| data = encode_type(type_, value, 4) | |
| self.assertAlmostEqual(flexbuffers.Loads(data), value, places=4) | |
| data = encode_type(type_, value, 8) | |
| self.assertEqual(flexbuffers.Loads(data), value) | |
| def test_string(self): | |
| for value in '', 'x', 'color', 'hello world': | |
| with self.subTest(value=value): | |
| data = encode_type(Type.STRING, value) | |
| self.assertEqual(flexbuffers.Loads(data), value) | |
| def test_blob(self): | |
| for value in bytes(), bytes([240, 12, 143, 7]), bytes(1000 * [17]): | |
| with self.subTest(value=value): | |
| data = encode_type(Type.BLOB, value) | |
| self.assertEqual(flexbuffers.Loads(data), value) | |
| def test_key(self): | |
| for value in '', 'color', 'hello world': | |
| with self.subTest(value=value): | |
| data = encode_type(Type.KEY, value) | |
| self.assertEqual(flexbuffers.Loads(data), value) | |
| with self.assertRaises(ValueError): | |
| encode_type(Type.KEY, (b'\x00' * 10).decode('ascii')) | |
| def test_vector(self): | |
| def encode_vector(elements, element_type): | |
| fbb = flexbuffers.Builder() | |
| with fbb.Vector(): | |
| add = fbb.Adder(element_type) | |
| for e in elements: | |
| add(e) | |
| return fbb.Finish() | |
| def encode_vector_from_elements(elements): | |
| fbb = flexbuffers.Builder() | |
| fbb.VectorFromElements(elements) | |
| return fbb.Finish() | |
| for elements in [], [1435], [56, 23, 0, 6783]: | |
| data = encode_vector(elements, Type.INT) | |
| self.assertEqual(flexbuffers.Loads(data), elements) | |
| data = encode_vector_from_elements(elements) | |
| self.assertEqual(flexbuffers.Loads(data), elements) | |
| # Elements of different type: one by one | |
| elements = [56.0, 'flexbuffers', 0, False, 75123] | |
| fbb = flexbuffers.Builder() | |
| with fbb.Vector(): | |
| fbb.Float(elements[0]) | |
| fbb.String(elements[1]) | |
| fbb.UInt(elements[2], 8) | |
| fbb.Bool(elements[3]) | |
| fbb.Int(elements[4]) | |
| data = fbb.Finish() | |
| self.assertEqual(flexbuffers.Loads(data), elements) | |
| # Elements of different type: all at once | |
| fbb = flexbuffers.Builder() | |
| fbb.VectorFromElements(elements) | |
| data = fbb.Finish() | |
| self.assertEqual(flexbuffers.Loads(data), elements) | |
| def test_nested_vectors(self): | |
| fbb = flexbuffers.Builder() | |
| with fbb.Vector(): | |
| fbb.String('begin') | |
| fbb.IndirectInt(42) | |
| with fbb.Vector(): | |
| for i in range(5): | |
| fbb.Int(i) | |
| fbb.String('end') | |
| data = fbb.Finish() | |
| self.assertEqual( | |
| flexbuffers.Loads(data), ['begin', 42, [0, 1, 2, 3, 4], 'end'] | |
| ) | |
| def test_big_vector(self): | |
| n = 10 * 1000 | |
| fbb = flexbuffers.Builder() | |
| with fbb.Vector(): | |
| for i in range(n): | |
| fbb.Int(i) | |
| self.assertEqual(flexbuffers.Loads(fbb.Finish()), list(range(n))) | |
| def test_typed_vector(self): | |
| def encode_typed_vector_from_elements(elements, element_type=None): | |
| fbb = flexbuffers.Builder() | |
| fbb.TypedVectorFromElements(elements, element_type) | |
| return fbb.Finish() | |
| for elements in [], [False], [True], [False, True, True, False, False]: | |
| data = encode_typed_vector_from_elements(elements, Type.BOOL) | |
| self.assertEqual(flexbuffers.Loads(data), elements) | |
| data = encode_typed_vector_from_elements(elements) | |
| self.assertEqual(flexbuffers.Loads(data), elements) | |
| for elements in [], [23455], [351, -2, 0, 6783, 0, -10]: | |
| data = encode_typed_vector_from_elements(elements, Type.INT) | |
| self.assertEqual(flexbuffers.Loads(data), elements) | |
| data = encode_typed_vector_from_elements(elements) | |
| self.assertEqual(flexbuffers.Loads(data), elements) | |
| for elements in [], [23455], [351, 2, 0, 6783, 0, 10]: | |
| data = encode_typed_vector_from_elements(elements) | |
| self.assertEqual(flexbuffers.Loads(data), elements) | |
| data = encode_typed_vector_from_elements(elements, Type.INT) | |
| self.assertEqual(flexbuffers.Loads(data), elements) | |
| data = encode_typed_vector_from_elements(elements, Type.UINT) | |
| self.assertEqual(flexbuffers.Loads(data), elements) | |
| for elements in [], [7.0], [52.0, 51.2, 70.0, -4.0]: | |
| data = encode_typed_vector_from_elements(elements, Type.FLOAT) | |
| self.assertEqual(flexbuffers.Loads(data), elements) | |
| data = encode_typed_vector_from_elements(elements) | |
| self.assertEqual(flexbuffers.Loads(data), elements) | |
| for elements in [], ['color'], ['x', 'y']: | |
| data = encode_typed_vector_from_elements(elements, Type.KEY) | |
| self.assertEqual(flexbuffers.Loads(data), elements) | |
| data = encode_typed_vector_from_elements(elements) | |
| self.assertEqual(flexbuffers.Loads(data), elements) | |
| def test_typed_vector_from_array(self): | |
| def encode_array(typecode, values): | |
| fbb = flexbuffers.Builder() | |
| fbb.VectorFromElements(array.array(typecode, values)) | |
| return fbb.Finish() | |
| values = [1.0, 3.14, -2.54, 0.0] | |
| data = encode_array('f', values) | |
| for a, b in zip(flexbuffers.Loads(data), values): | |
| self.assertAlmostEqual(a, b, places=2) | |
| values = [1.0, 3.14, -2.54, 0.0] | |
| data = encode_array('d', values) | |
| self.assertEqual(flexbuffers.Loads(data), values) | |
| values = [1, -7, 9, 26, 12] | |
| data = encode_array('i', values) | |
| self.assertEqual(flexbuffers.Loads(data), values) | |
| values = [0, 1, 2, 3, 4, 5, 6] | |
| data = encode_array('I', values) | |
| self.assertEqual(flexbuffers.Loads(data), values) | |
| def test_fixed_typed_vector(self): | |
| def encode_fixed_typed_vector(elements, element_type=None): | |
| fbb = flexbuffers.Builder() | |
| fbb.FixedTypedVectorFromElements(elements, element_type) | |
| return fbb.Finish() | |
| for elements in ( | |
| (-2, 2), | |
| (1, 2, 3), | |
| (100, -100, 200, -200), | |
| (4.0, 7.0), | |
| (0.0, 1.0, 8.0), | |
| (9.0, 7.0, 1.0, 5.5), | |
| ): | |
| with self.subTest(elements=elements): | |
| data = encode_fixed_typed_vector(elements) | |
| self.assertSequenceEqual(flexbuffers.Loads(data), elements) | |
| elements = [-170, 432, 0, -7] | |
| data = encode_fixed_typed_vector(elements, Type.INT) | |
| self.assertSequenceEqual(flexbuffers.Loads(data), elements) | |
| with self.assertRaises(ValueError): | |
| encode_fixed_typed_vector([]) # Invalid input length | |
| with self.assertRaises(ValueError): | |
| encode_fixed_typed_vector([1]) # Invalid input length | |
| with self.assertRaises(ValueError): | |
| encode_fixed_typed_vector([1, 2, 3, 4, 5]) # Invalid input length | |
| with self.assertRaises(TypeError): | |
| encode_fixed_typed_vector([1, 1.0]) # Invalid input types | |
| with self.assertRaises(TypeError): | |
| encode_fixed_typed_vector(['', '']) # Invalid input types | |
| def test_map_builder(self): | |
| def get_keys(data): | |
| return [key.AsKey for key in flexbuffers.GetRoot(data).AsMap.Keys] | |
| # Empty map | |
| fbb = flexbuffers.Builder() | |
| with fbb.Map(): | |
| pass | |
| data = fbb.Finish() | |
| self.assertEqual(flexbuffers.Loads(data), {}) | |
| # Two-element map of Int | |
| fbb = flexbuffers.Builder() | |
| with fbb.Map(): | |
| fbb.Int('y', -2) | |
| fbb.Int('x', 10) | |
| data = fbb.Finish() | |
| self.assertEqual(flexbuffers.Loads(data), {'x': 10, 'y': -2}) | |
| # Multiple-element map of vectors | |
| fbb = flexbuffers.Builder() | |
| with fbb.Map(): | |
| with fbb.Vector('v'): | |
| fbb.Int(45) | |
| with fbb.TypedVector('tv'): | |
| fbb.Int(-7) | |
| fbb.FixedTypedVectorFromElements('ftv', [-2.0, 1.0]) | |
| data = fbb.Finish() | |
| self.assertEqual( | |
| flexbuffers.Loads(data), {'v': [45], 'tv': [-7], 'ftv': [-2.0, 1.0]} | |
| ) | |
| keys = get_keys(data) | |
| self.assertEqual(sorted(keys), keys) | |
| # Multiple-element map of different types | |
| fbb = flexbuffers.Builder() | |
| with fbb.Map(): | |
| fbb.Null('n') | |
| fbb.Bool('b', False) | |
| fbb.Int('i', -27) | |
| fbb.UInt('u', 27) | |
| fbb.Float('f', -0.85) | |
| fbb.String('s', 'String') | |
| fbb.Blob('bb', b'data') | |
| fbb.IndirectInt('ii', -9500) | |
| fbb.IndirectUInt('iu', 540) | |
| fbb.IndirectFloat('if', 0.0) | |
| fbb.VectorFromElements('v', [2, 1, 0.0]) | |
| fbb.TypedVectorFromElements('tv', [2, 1, 0]) | |
| fbb.FixedTypedVectorFromElements('ftv', [2.0, -6.0]) | |
| data = fbb.Finish() | |
| self.assertEqual( | |
| flexbuffers.Loads(data), | |
| { | |
| 'n': None, | |
| 'b': False, | |
| 'i': -27, | |
| 'u': 27, | |
| 'f': -0.85, | |
| 's': 'String', | |
| 'bb': b'data', | |
| 'ii': -9500, | |
| 'iu': 540, | |
| 'if': 0.0, | |
| 'v': [2, 1, 0.0], | |
| 'tv': [2, 1, 0], | |
| 'ftv': [2.0, -6.0], | |
| }, | |
| ) | |
| keys = get_keys(data) | |
| self.assertEqual(sorted(keys), keys) | |
| def test_map_python(self): | |
| maps = [ | |
| {}, | |
| {'key': 'value'}, | |
| {'x': None, 'y': 3400, 'z': -7040}, | |
| {'zzz': 100, 'aaa': 5.0, 'ccc': ['Test', 32, False, None, True]}, | |
| { | |
| 'name': ['John', 'Smith'], | |
| 'valid': True, | |
| 'note': None, | |
| 'address': { | |
| 'lines': [175, 'Alhambra'], | |
| 'city': 'San Francisco', | |
| 'zip': 94123, | |
| }, | |
| }, | |
| ] | |
| for m in maps: | |
| self.assertEqual(flexbuffers.Loads(flexbuffers.Dumps(m)), m) | |
| def test_gold_from_file(self): | |
| data = read_test_file(GOLD_FLEXBUFFER_FILE) | |
| self.assertEqual(flexbuffers.Loads(data), GOLD_FLEXBUFFER_OBJ) | |
| def test_gold_from_builder(self): | |
| fbb = flexbuffers.Builder() | |
| with fbb.Map(): | |
| with fbb.Vector('vec'): | |
| fbb.Int(-100) | |
| fbb.String('Fred') | |
| fbb.IndirectFloat(4.0) | |
| i_f = fbb.LastValue | |
| fbb.Blob(bytes([77])) | |
| fbb.Bool(False) | |
| fbb.ReuseValue(i_f) | |
| vec = [1, 2, 3] | |
| fbb.VectorFromElements('bar', vec) | |
| fbb.FixedTypedVectorFromElements('bar3', [1, 2, 3]) | |
| fbb.VectorFromElements('bools', [True, False, True, False]) | |
| fbb.Bool('bool', True) | |
| fbb.Float('foo', 100) | |
| with fbb.Map('mymap'): | |
| fbb.String('foo', 'Fred') | |
| data = fbb.Finish() | |
| self.assertEqual(flexbuffers.Loads(data), GOLD_FLEXBUFFER_OBJ) | |
| def test_min_bit_width(self): | |
| fbb = flexbuffers.Builder(force_min_bit_width=flexbuffers.BitWidth.W8) | |
| fbb.TypedVectorFromElements([0, 1, 0, 1, 0]) | |
| data = fbb.Finish() | |
| root = flexbuffers.GetRoot(data) | |
| self.assertTrue(root.IsTypedVector) | |
| self.assertEqual(root.AsTypedVector.ByteWidth, 1) | |
| fbb = flexbuffers.Builder(force_min_bit_width=flexbuffers.BitWidth.W32) | |
| fbb.TypedVectorFromElements([0, 1, 0, 1, 0]) | |
| data = fbb.Finish() | |
| root = flexbuffers.GetRoot(data) | |
| self.assertTrue(root.IsTypedVector) | |
| self.assertEqual(root.AsTypedVector.ByteWidth, 4) | |
| def test_share_keys(self): | |
| def encode_key_vector(value, count, share_keys): | |
| fbb = flexbuffers.Builder(share_keys=share_keys) | |
| with fbb.Vector(): | |
| for _ in range(count): | |
| fbb.Key(value) | |
| return fbb.Finish(), fbb.KeyPool.Elements | |
| data, pool = encode_key_vector('test', 10, share_keys=False) | |
| self.assertEqual(len(pool), 0) | |
| self.assertEqual(len(data), 74) | |
| self.assertEqual(flexbuffers.Loads(data), 10 * ['test']) | |
| data, pool = encode_key_vector('test', 10, share_keys=True) | |
| self.assertEqual(len(pool), 1) | |
| self.assertEqual(pool[0], 'test'.encode('ascii')) | |
| self.assertEqual(len(data), 29) | |
| self.assertEqual(flexbuffers.Loads(data), 10 * ['test']) | |
| def test_share_strings(self): | |
| def encode_string_vector(value, count, share_strings): | |
| fbb = flexbuffers.Builder(share_strings=share_strings) | |
| with fbb.Vector(): | |
| for _ in range(count): | |
| fbb.String(value) | |
| return fbb.Finish(), fbb.StringPool.Elements | |
| data, pool = encode_string_vector('test', 10, share_strings=False) | |
| self.assertEqual(len(pool), 0) | |
| self.assertEqual(len(data), 84) | |
| self.assertEqual(flexbuffers.Loads(data), 10 * ['test']) | |
| data, pool = encode_string_vector('test', 10, share_strings=True) | |
| self.assertEqual(len(pool), 1) | |
| self.assertEqual(pool[0], 'test'.encode('utf-8')) | |
| self.assertEqual(len(data), 30) | |
| self.assertEqual(flexbuffers.Loads(data), 10 * ['test']) | |
| def test_invalid_stack_size(self): | |
| fbb = flexbuffers.Builder() | |
| with self.assertRaises(RuntimeError): | |
| fbb.Finish() | |
| fbb.Int(100) | |
| fbb.Int(200) | |
| with self.assertRaises(RuntimeError): | |
| fbb.Finish() | |
| fbb.Clear() | |
| fbb.Int(420) | |
| fbb.Finish() | |
| if __name__ == '__main__': | |
| unittest.main() |