You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
1371 lines
48 KiB
1371 lines
48 KiB
# coding=utf-8 |
|
# Copyright 2014 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. |
|
|
|
import os.path |
|
import sys |
|
PY_VERSION = sys.version_info[:2] |
|
|
|
import ctypes |
|
from collections import defaultdict |
|
import timeit |
|
import unittest |
|
|
|
|
|
from flatbuffers import compat |
|
from flatbuffers.compat import range_func as compat_range |
|
|
|
import flatbuffers |
|
from flatbuffers import number_types as N |
|
|
|
import MyGame # refers to generated code |
|
import MyGame.Example # refers to generated code |
|
import MyGame.Example.Any # refers to generated code |
|
import MyGame.Example.Color # refers to generated code |
|
import MyGame.Example.Monster # refers to generated code |
|
import MyGame.Example.Test # refers to generated code |
|
import MyGame.Example.Stat # refers to generated code |
|
import MyGame.Example.Vec3 # refers to generated code |
|
|
|
|
|
def assertRaises(test_case, fn, exception_class): |
|
''' Backwards-compatible assertion for exceptions raised. ''' |
|
|
|
exc = None |
|
try: |
|
fn() |
|
except Exception as e: |
|
exc = e |
|
test_case.assertTrue(exc is not None) |
|
test_case.assertTrue(isinstance(exc, exception_class)) |
|
|
|
|
|
class TestWireFormat(unittest.TestCase): |
|
def test_wire_format(self): |
|
# Verify that using the generated Python code builds a buffer without |
|
# returning errors, and is interpreted correctly: |
|
gen_buf, gen_off = make_monster_from_generated_code() |
|
CheckReadBuffer(gen_buf, gen_off) |
|
|
|
# Verify that the canonical flatbuffer file is readable by the |
|
# generated Python code. Note that context managers are not part of |
|
# Python 2.5, so we use the simpler open/close methods here: |
|
f = open('monsterdata_test.mon', 'rb') |
|
canonicalWireData = f.read() |
|
f.close() |
|
CheckReadBuffer(bytearray(canonicalWireData), 0) |
|
|
|
# Write the generated buffer out to a file: |
|
f = open('monsterdata_python_wire.mon', 'wb') |
|
f.write(gen_buf[gen_off:]) |
|
f.close() |
|
|
|
|
|
def CheckReadBuffer(buf, offset): |
|
''' CheckReadBuffer checks that the given buffer is evaluated correctly |
|
as the example Monster. ''' |
|
|
|
def asserter(stmt): |
|
''' An assertion helper that is separated from TestCase classes. ''' |
|
if not stmt: |
|
raise AssertionError('CheckReadBuffer case failed') |
|
|
|
monster = MyGame.Example.Monster.Monster.GetRootAsMonster(buf, offset) |
|
|
|
asserter(monster.Hp() == 80) |
|
asserter(monster.Mana() == 150) |
|
asserter(monster.Name() == b'MyMonster') |
|
|
|
# initialize a Vec3 from Pos() |
|
vec = monster.Pos() |
|
asserter(vec is not None) |
|
|
|
# verify the properties of the Vec3 |
|
asserter(vec.X() == 1.0) |
|
asserter(vec.Y() == 2.0) |
|
asserter(vec.Z() == 3.0) |
|
asserter(vec.Test1() == 3.0) |
|
asserter(vec.Test2() == 2) |
|
|
|
# initialize a Test from Test3(...) |
|
t = MyGame.Example.Test.Test() |
|
t = vec.Test3(t) |
|
asserter(t is not None) |
|
|
|
# verify the properties of the Test |
|
asserter(t.A() == 5) |
|
asserter(t.B() == 6) |
|
|
|
# verify that the enum code matches the enum declaration: |
|
union_type = MyGame.Example.Any.Any |
|
asserter(monster.TestType() == union_type.Monster) |
|
|
|
# initialize a Table from a union field Test(...) |
|
table2 = monster.Test() |
|
asserter(type(table2) is flatbuffers.table.Table) |
|
|
|
# initialize a Monster from the Table from the union |
|
monster2 = MyGame.Example.Monster.Monster() |
|
monster2.Init(table2.Bytes, table2.Pos) |
|
|
|
asserter(monster2.Name() == b"Fred") |
|
|
|
# iterate through the first monster's inventory: |
|
asserter(monster.InventoryLength() == 5) |
|
|
|
invsum = 0 |
|
for i in compat_range(monster.InventoryLength()): |
|
v = monster.Inventory(i) |
|
invsum += int(v) |
|
asserter(invsum == 10) |
|
|
|
asserter(monster.Test4Length() == 2) |
|
|
|
# create a 'Test' object and populate it: |
|
test0 = monster.Test4(0) |
|
asserter(type(test0) is MyGame.Example.Test.Test) |
|
|
|
test1 = monster.Test4(1) |
|
asserter(type(test1) is MyGame.Example.Test.Test) |
|
|
|
# the position of test0 and test1 are swapped in monsterdata_java_wire |
|
# and monsterdata_test_wire, so ignore ordering |
|
v0 = test0.A() |
|
v1 = test0.B() |
|
v2 = test1.A() |
|
v3 = test1.B() |
|
sumtest12 = int(v0) + int(v1) + int(v2) + int(v3) |
|
|
|
asserter(sumtest12 == 100) |
|
|
|
asserter(monster.TestarrayofstringLength() == 2) |
|
asserter(monster.Testarrayofstring(0) == b"test1") |
|
asserter(monster.Testarrayofstring(1) == b"test2") |
|
|
|
asserter(monster.TestarrayoftablesLength() == 0) |
|
asserter(monster.TestnestedflatbufferLength() == 0) |
|
asserter(monster.Testempty() is None) |
|
|
|
|
|
class TestFuzz(unittest.TestCase): |
|
''' Low level stress/fuzz test: serialize/deserialize a variety of |
|
different kinds of data in different combinations ''' |
|
|
|
binary_type = compat.binary_types[0] # this will always exist |
|
ofInt32Bytes = binary_type([0x83, 0x33, 0x33, 0x33]) |
|
ofInt64Bytes = binary_type([0x84, 0x44, 0x44, 0x44, |
|
0x44, 0x44, 0x44, 0x44]) |
|
overflowingInt32Val = flatbuffers.encode.Get(flatbuffers.packer.int32, |
|
ofInt32Bytes, 0) |
|
overflowingInt64Val = flatbuffers.encode.Get(flatbuffers.packer.int64, |
|
ofInt64Bytes, 0) |
|
|
|
# Values we're testing against: chosen to ensure no bits get chopped |
|
# off anywhere, and also be different from eachother. |
|
boolVal = True |
|
int8Val = N.Int8Flags.py_type(-127) # 0x81 |
|
uint8Val = N.Uint8Flags.py_type(0xFF) |
|
int16Val = N.Int16Flags.py_type(-32222) # 0x8222 |
|
uint16Val = N.Uint16Flags.py_type(0xFEEE) |
|
int32Val = N.Int32Flags.py_type(overflowingInt32Val) |
|
uint32Val = N.Uint32Flags.py_type(0xFDDDDDDD) |
|
int64Val = N.Int64Flags.py_type(overflowingInt64Val) |
|
uint64Val = N.Uint64Flags.py_type(0xFCCCCCCCCCCCCCCC) |
|
# Python uses doubles, so force it here |
|
float32Val = N.Float32Flags.py_type(ctypes.c_float(3.14159).value) |
|
float64Val = N.Float64Flags.py_type(3.14159265359) |
|
|
|
def test_fuzz(self): |
|
return self.check_once(11, 100) |
|
|
|
def check_once(self, fuzzFields, fuzzObjects): |
|
testValuesMax = 11 # hardcoded to the number of scalar types |
|
|
|
builder = flatbuffers.Builder(0) |
|
l = LCG() |
|
|
|
objects = [0 for _ in compat_range(fuzzObjects)] |
|
|
|
# Generate fuzzObjects random objects each consisting of |
|
# fuzzFields fields, each of a random type. |
|
for i in compat_range(fuzzObjects): |
|
builder.StartObject(fuzzFields) |
|
|
|
for j in compat_range(fuzzFields): |
|
choice = int(l.Next()) % testValuesMax |
|
if choice == 0: |
|
builder.PrependBoolSlot(int(j), self.boolVal, False) |
|
elif choice == 1: |
|
builder.PrependInt8Slot(int(j), self.int8Val, 0) |
|
elif choice == 2: |
|
builder.PrependUint8Slot(int(j), self.uint8Val, 0) |
|
elif choice == 3: |
|
builder.PrependInt16Slot(int(j), self.int16Val, 0) |
|
elif choice == 4: |
|
builder.PrependUint16Slot(int(j), self.uint16Val, 0) |
|
elif choice == 5: |
|
builder.PrependInt32Slot(int(j), self.int32Val, 0) |
|
elif choice == 6: |
|
builder.PrependUint32Slot(int(j), self.uint32Val, 0) |
|
elif choice == 7: |
|
builder.PrependInt64Slot(int(j), self.int64Val, 0) |
|
elif choice == 8: |
|
builder.PrependUint64Slot(int(j), self.uint64Val, 0) |
|
elif choice == 9: |
|
builder.PrependFloat32Slot(int(j), self.float32Val, 0) |
|
elif choice == 10: |
|
builder.PrependFloat64Slot(int(j), self.float64Val, 0) |
|
else: |
|
raise RuntimeError('unreachable') |
|
|
|
off = builder.EndObject() |
|
|
|
# store the offset from the end of the builder buffer, |
|
# since it will keep growing: |
|
objects[i] = off |
|
|
|
# Do some bookkeeping to generate stats on fuzzes: |
|
stats = defaultdict(int) |
|
def check(table, desc, want, got): |
|
stats[desc] += 1 |
|
self.assertEqual(want, got, "%s != %s, %s" % (want, got, desc)) |
|
|
|
l = LCG() # Reset. |
|
|
|
# Test that all objects we generated are readable and return the |
|
# expected values. We generate random objects in the same order |
|
# so this is deterministic. |
|
for i in compat_range(fuzzObjects): |
|
|
|
table = flatbuffers.table.Table(builder.Bytes, |
|
len(builder.Bytes) - objects[i]) |
|
|
|
for j in compat_range(fuzzFields): |
|
field_count = flatbuffers.builder.VtableMetadataFields + j |
|
f = N.VOffsetTFlags.py_type(field_count * |
|
N.VOffsetTFlags.bytewidth) |
|
choice = int(l.Next()) % testValuesMax |
|
|
|
if choice == 0: |
|
check(table, "bool", self.boolVal, |
|
table.GetSlot(f, False, N.BoolFlags)) |
|
elif choice == 1: |
|
check(table, "int8", self.int8Val, |
|
table.GetSlot(f, 0, N.Int8Flags)) |
|
elif choice == 2: |
|
check(table, "uint8", self.uint8Val, |
|
table.GetSlot(f, 0, N.Uint8Flags)) |
|
elif choice == 3: |
|
check(table, "int16", self.int16Val, |
|
table.GetSlot(f, 0, N.Int16Flags)) |
|
elif choice == 4: |
|
check(table, "uint16", self.uint16Val, |
|
table.GetSlot(f, 0, N.Uint16Flags)) |
|
elif choice == 5: |
|
check(table, "int32", self.int32Val, |
|
table.GetSlot(f, 0, N.Int32Flags)) |
|
elif choice == 6: |
|
check(table, "uint32", self.uint32Val, |
|
table.GetSlot(f, 0, N.Uint32Flags)) |
|
elif choice == 7: |
|
check(table, "int64", self.int64Val, |
|
table.GetSlot(f, 0, N.Int64Flags)) |
|
elif choice == 8: |
|
check(table, "uint64", self.uint64Val, |
|
table.GetSlot(f, 0, N.Uint64Flags)) |
|
elif choice == 9: |
|
check(table, "float32", self.float32Val, |
|
table.GetSlot(f, 0, N.Float32Flags)) |
|
elif choice == 10: |
|
check(table, "float64", self.float64Val, |
|
table.GetSlot(f, 0, N.Float64Flags)) |
|
else: |
|
raise RuntimeError('unreachable') |
|
|
|
# If enough checks were made, verify that all scalar types were used: |
|
self.assertEqual(testValuesMax, len(stats), |
|
"fuzzing failed to test all scalar types: %s" % stats) |
|
|
|
|
|
class TestByteLayout(unittest.TestCase): |
|
''' TestByteLayout checks the bytes of a Builder in various scenarios. ''' |
|
|
|
def assertBuilderEquals(self, builder, want_chars_or_ints): |
|
def integerize(x): |
|
if isinstance(x, compat.string_types): |
|
return ord(x) |
|
return x |
|
|
|
want_ints = list(map(integerize, want_chars_or_ints)) |
|
want = bytearray(want_ints) |
|
got = builder.Bytes[builder.Head():] # use the buffer directly |
|
self.assertEqual(want, got) |
|
|
|
def test_numbers(self): |
|
b = flatbuffers.Builder(0) |
|
self.assertBuilderEquals(b, []) |
|
b.PrependBool(True) |
|
self.assertBuilderEquals(b, [1]) |
|
b.PrependInt8(-127) |
|
self.assertBuilderEquals(b, [129, 1]) |
|
b.PrependUint8(255) |
|
self.assertBuilderEquals(b, [255, 129, 1]) |
|
b.PrependInt16(-32222) |
|
self.assertBuilderEquals(b, [0x22, 0x82, 0, 255, 129, 1]) # first pad |
|
b.PrependUint16(0xFEEE) |
|
# no pad this time: |
|
self.assertBuilderEquals(b, [0xEE, 0xFE, 0x22, 0x82, 0, 255, 129, 1]) |
|
b.PrependInt32(-53687092) |
|
self.assertBuilderEquals(b, [204, 204, 204, 252, 0xEE, 0xFE, |
|
0x22, 0x82, 0, 255, 129, 1]) |
|
b.PrependUint32(0x98765432) |
|
self.assertBuilderEquals(b, [0x32, 0x54, 0x76, 0x98, |
|
204, 204, 204, 252, |
|
0xEE, 0xFE, 0x22, 0x82, |
|
0, 255, 129, 1]) |
|
|
|
def test_numbers64(self): |
|
b = flatbuffers.Builder(0) |
|
b.PrependUint64(0x1122334455667788) |
|
self.assertBuilderEquals(b, [0x88, 0x77, 0x66, 0x55, |
|
0x44, 0x33, 0x22, 0x11]) |
|
|
|
b = flatbuffers.Builder(0) |
|
b.PrependInt64(0x1122334455667788) |
|
self.assertBuilderEquals(b, [0x88, 0x77, 0x66, 0x55, |
|
0x44, 0x33, 0x22, 0x11]) |
|
|
|
def test_1xbyte_vector(self): |
|
b = flatbuffers.Builder(0) |
|
self.assertBuilderEquals(b, []) |
|
b.StartVector(flatbuffers.number_types.Uint8Flags.bytewidth, 1, 1) |
|
self.assertBuilderEquals(b, [0, 0, 0]) # align to 4bytes |
|
b.PrependByte(1) |
|
self.assertBuilderEquals(b, [1, 0, 0, 0]) |
|
b.EndVector(1) |
|
self.assertBuilderEquals(b, [1, 0, 0, 0, 1, 0, 0, 0]) # padding |
|
|
|
def test_2xbyte_vector(self): |
|
b = flatbuffers.Builder(0) |
|
b.StartVector(flatbuffers.number_types.Uint8Flags.bytewidth, 2, 1) |
|
self.assertBuilderEquals(b, [0, 0]) # align to 4bytes |
|
b.PrependByte(1) |
|
self.assertBuilderEquals(b, [1, 0, 0]) |
|
b.PrependByte(2) |
|
self.assertBuilderEquals(b, [2, 1, 0, 0]) |
|
b.EndVector(2) |
|
self.assertBuilderEquals(b, [2, 0, 0, 0, 2, 1, 0, 0]) # padding |
|
|
|
def test_1xuint16_vector(self): |
|
b = flatbuffers.Builder(0) |
|
b.StartVector(flatbuffers.number_types.Uint16Flags.bytewidth, 1, 1) |
|
self.assertBuilderEquals(b, [0, 0]) # align to 4bytes |
|
b.PrependUint16(1) |
|
self.assertBuilderEquals(b, [1, 0, 0, 0]) |
|
b.EndVector(1) |
|
self.assertBuilderEquals(b, [1, 0, 0, 0, 1, 0, 0, 0]) # padding |
|
|
|
def test_2xuint16_vector(self): |
|
b = flatbuffers.Builder(0) |
|
b.StartVector(flatbuffers.number_types.Uint16Flags.bytewidth, 2, 1) |
|
self.assertBuilderEquals(b, []) # align to 4bytes |
|
b.PrependUint16(0xABCD) |
|
self.assertBuilderEquals(b, [0xCD, 0xAB]) |
|
b.PrependUint16(0xDCBA) |
|
self.assertBuilderEquals(b, [0xBA, 0xDC, 0xCD, 0xAB]) |
|
b.EndVector(2) |
|
self.assertBuilderEquals(b, [2, 0, 0, 0, 0xBA, 0xDC, 0xCD, 0xAB]) |
|
|
|
def test_create_ascii_string(self): |
|
b = flatbuffers.Builder(0) |
|
b.CreateString(u"foo", encoding='ascii') |
|
|
|
# 0-terminated, no pad: |
|
self.assertBuilderEquals(b, [3, 0, 0, 0, 'f', 'o', 'o', 0]) |
|
b.CreateString(u"moop", encoding='ascii') |
|
# 0-terminated, 3-byte pad: |
|
self.assertBuilderEquals(b, [4, 0, 0, 0, 'm', 'o', 'o', 'p', |
|
0, 0, 0, 0, |
|
3, 0, 0, 0, 'f', 'o', 'o', 0]) |
|
|
|
def test_create_utf8_string(self): |
|
b = flatbuffers.Builder(0) |
|
b.CreateString(u"Цлїςσδε") |
|
self.assertBuilderEquals(b, "\x0e\x00\x00\x00\xd0\xa6\xd0\xbb\xd1\x97" \ |
|
"\xcf\x82\xcf\x83\xce\xb4\xce\xb5\x00\x00") |
|
|
|
b.CreateString(u"ﾌﾑｱﾑｶﾓｹﾓ") |
|
self.assertBuilderEquals(b, "\x18\x00\x00\x00\xef\xbe\x8c\xef\xbe\x91" \ |
|
"\xef\xbd\xb1\xef\xbe\x91\xef\xbd\xb6\xef\xbe\x93\xef\xbd\xb9\xef" \ |
|
"\xbe\x93\x00\x00\x00\x00\x0e\x00\x00\x00\xd0\xa6\xd0\xbb\xd1\x97" \ |
|
"\xcf\x82\xcf\x83\xce\xb4\xce\xb5\x00\x00") |
|
|
|
def test_create_arbitrary_string(self): |
|
b = flatbuffers.Builder(0) |
|
s = "\x01\x02\x03" |
|
b.CreateString(s) # Default encoding is utf-8. |
|
# 0-terminated, no pad: |
|
self.assertBuilderEquals(b, [3, 0, 0, 0, 1, 2, 3, 0]) |
|
s2 = "\x04\x05\x06\x07" |
|
b.CreateString(s2) # Default encoding is utf-8. |
|
# 0-terminated, 3-byte pad: |
|
self.assertBuilderEquals(b, [4, 0, 0, 0, 4, 5, 6, 7, 0, 0, 0, 0, |
|
3, 0, 0, 0, 1, 2, 3, 0]) |
|
|
|
def test_empty_vtable(self): |
|
b = flatbuffers.Builder(0) |
|
b.StartObject(0) |
|
self.assertBuilderEquals(b, []) |
|
b.EndObject() |
|
self.assertBuilderEquals(b, [4, 0, 4, 0, 4, 0, 0, 0]) |
|
|
|
def test_vtable_with_one_true_bool(self): |
|
b = flatbuffers.Builder(0) |
|
self.assertBuilderEquals(b, []) |
|
b.StartObject(1) |
|
self.assertBuilderEquals(b, []) |
|
b.PrependBoolSlot(0, True, False) |
|
b.EndObject() |
|
self.assertBuilderEquals(b, [ |
|
6, 0, # vtable bytes |
|
8, 0, # length of object including vtable offset |
|
7, 0, # start of bool value |
|
6, 0, 0, 0, # offset for start of vtable (int32) |
|
0, 0, 0, # padded to 4 bytes |
|
1, # bool value |
|
]) |
|
|
|
def test_vtable_with_one_default_bool(self): |
|
b = flatbuffers.Builder(0) |
|
self.assertBuilderEquals(b, []) |
|
b.StartObject(1) |
|
self.assertBuilderEquals(b, []) |
|
b.PrependBoolSlot(0, False, False) |
|
b.EndObject() |
|
self.assertBuilderEquals(b, [ |
|
6, 0, # vtable bytes |
|
4, 0, # end of object from here |
|
0, 0, # entry 1 is zero |
|
6, 0, 0, 0, # offset for start of vtable (int32) |
|
]) |
|
|
|
def test_vtable_with_one_int16(self): |
|
b = flatbuffers.Builder(0) |
|
b.StartObject(1) |
|
b.PrependInt16Slot(0, 0x789A, 0) |
|
b.EndObject() |
|
self.assertBuilderEquals(b, [ |
|
6, 0, # vtable bytes |
|
8, 0, # end of object from here |
|
6, 0, # offset to value |
|
6, 0, 0, 0, # offset for start of vtable (int32) |
|
0, 0, # padding to 4 bytes |
|
0x9A, 0x78, |
|
]) |
|
|
|
def test_vtable_with_two_int16(self): |
|
b = flatbuffers.Builder(0) |
|
b.StartObject(2) |
|
b.PrependInt16Slot(0, 0x3456, 0) |
|
b.PrependInt16Slot(1, 0x789A, 0) |
|
b.EndObject() |
|
self.assertBuilderEquals(b, [ |
|
8, 0, # vtable bytes |
|
8, 0, # end of object from here |
|
6, 0, # offset to value 0 |
|
4, 0, # offset to value 1 |
|
8, 0, 0, 0, # offset for start of vtable (int32) |
|
0x9A, 0x78, # value 1 |
|
0x56, 0x34, # value 0 |
|
]) |
|
|
|
def test_vtable_with_int16_and_bool(self): |
|
b = flatbuffers.Builder(0) |
|
b.StartObject(2) |
|
b.PrependInt16Slot(0, 0x3456, 0) |
|
b.PrependBoolSlot(1, True, False) |
|
b.EndObject() |
|
self.assertBuilderEquals(b, [ |
|
8, 0, # vtable bytes |
|
8, 0, # end of object from here |
|
6, 0, # offset to value 0 |
|
5, 0, # offset to value 1 |
|
8, 0, 0, 0, # offset for start of vtable (int32) |
|
0, # padding |
|
1, # value 1 |
|
0x56, 0x34, # value 0 |
|
]) |
|
|
|
def test_vtable_with_empty_vector(self): |
|
b = flatbuffers.Builder(0) |
|
b.StartVector(flatbuffers.number_types.Uint8Flags.bytewidth, 0, 1) |
|
vecend = b.EndVector(0) |
|
b.StartObject(1) |
|
b.PrependUOffsetTRelativeSlot(0, vecend, 0) |
|
b.EndObject() |
|
self.assertBuilderEquals(b, [ |
|
6, 0, # vtable bytes |
|
8, 0, |
|
4, 0, # offset to vector offset |
|
6, 0, 0, 0, # offset for start of vtable (int32) |
|
4, 0, 0, 0, |
|
0, 0, 0, 0, # length of vector (not in struct) |
|
]) |
|
|
|
def test_vtable_with_empty_vector_of_byte_and_some_scalars(self): |
|
b = flatbuffers.Builder(0) |
|
b.StartVector(flatbuffers.number_types.Uint8Flags.bytewidth, 0, 1) |
|
vecend = b.EndVector(0) |
|
b.StartObject(2) |
|
b.PrependInt16Slot(0, 55, 0) |
|
b.PrependUOffsetTRelativeSlot(1, vecend, 0) |
|
b.EndObject() |
|
self.assertBuilderEquals(b, [ |
|
8, 0, # vtable bytes |
|
12, 0, |
|
10, 0, # offset to value 0 |
|
4, 0, # offset to vector offset |
|
8, 0, 0, 0, # vtable loc |
|
8, 0, 0, 0, # value 1 |
|
0, 0, 55, 0, # value 0 |
|
|
|
0, 0, 0, 0, # length of vector (not in struct) |
|
]) |
|
|
|
def test_vtable_with_1_int16_and_2vector_of_int16(self): |
|
b = flatbuffers.Builder(0) |
|
b.StartVector(flatbuffers.number_types.Int16Flags.bytewidth, 2, 1) |
|
b.PrependInt16(0x1234) |
|
b.PrependInt16(0x5678) |
|
vecend = b.EndVector(2) |
|
b.StartObject(2) |
|
b.PrependUOffsetTRelativeSlot(1, vecend, 0) |
|
b.PrependInt16Slot(0, 55, 0) |
|
b.EndObject() |
|
self.assertBuilderEquals(b, [ |
|
8, 0, # vtable bytes |
|
12, 0, # length of object |
|
6, 0, # start of value 0 from end of vtable |
|
8, 0, # start of value 1 from end of buffer |
|
8, 0, 0, 0, # offset for start of vtable (int32) |
|
0, 0, # padding |
|
55, 0, # value 0 |
|
4, 0, 0, 0, # vector position from here |
|
2, 0, 0, 0, # length of vector (uint32) |
|
0x78, 0x56, # vector value 1 |
|
0x34, 0x12, # vector value 0 |
|
]) |
|
|
|
def test_vtable_with_1_struct_of_1_int8__1_int16__1_int32(self): |
|
b = flatbuffers.Builder(0) |
|
b.StartObject(1) |
|
b.Prep(4+4+4, 0) |
|
b.PrependInt8(55) |
|
b.Pad(3) |
|
b.PrependInt16(0x1234) |
|
b.Pad(2) |
|
b.PrependInt32(0x12345678) |
|
structStart = b.Offset() |
|
b.PrependStructSlot(0, structStart, 0) |
|
b.EndObject() |
|
self.assertBuilderEquals(b, [ |
|
6, 0, # vtable bytes |
|
16, 0, # end of object from here |
|
4, 0, # start of struct from here |
|
6, 0, 0, 0, # offset for start of vtable (int32) |
|
0x78, 0x56, 0x34, 0x12, # value 2 |
|
0, 0, # padding |
|
0x34, 0x12, # value 1 |
|
0, 0, 0, # padding |
|
55, # value 0 |
|
]) |
|
|
|
def test_vtable_with_1_vector_of_2_struct_of_2_int8(self): |
|
b = flatbuffers.Builder(0) |
|
b.StartVector(flatbuffers.number_types.Int8Flags.bytewidth*2, 2, 1) |
|
b.PrependInt8(33) |
|
b.PrependInt8(44) |
|
b.PrependInt8(55) |
|
b.PrependInt8(66) |
|
vecend = b.EndVector(2) |
|
b.StartObject(1) |
|
b.PrependUOffsetTRelativeSlot(0, vecend, 0) |
|
b.EndObject() |
|
self.assertBuilderEquals(b, [ |
|
6, 0, # vtable bytes |
|
8, 0, |
|
4, 0, # offset of vector offset |
|
6, 0, 0, 0, # offset for start of vtable (int32) |
|
4, 0, 0, 0, # vector start offset |
|
|
|
2, 0, 0, 0, # vector length |
|
66, # vector value 1,1 |
|
55, # vector value 1,0 |
|
44, # vector value 0,1 |
|
33, # vector value 0,0 |
|
]) |
|
|
|
def test_table_with_some_elements(self): |
|
b = flatbuffers.Builder(0) |
|
b.StartObject(2) |
|
b.PrependInt8Slot(0, 33, 0) |
|
b.PrependInt16Slot(1, 66, 0) |
|
off = b.EndObject() |
|
b.Finish(off) |
|
|
|
self.assertBuilderEquals(b, [ |
|
12, 0, 0, 0, # root of table: points to vtable offset |
|
|
|
8, 0, # vtable bytes |
|
8, 0, # end of object from here |
|
7, 0, # start of value 0 |
|
4, 0, # start of value 1 |
|
|
|
8, 0, 0, 0, # offset for start of vtable (int32) |
|
|
|
66, 0, # value 1 |
|
0, # padding |
|
33, # value 0 |
|
]) |
|
|
|
def test__one_unfinished_table_and_one_finished_table(self): |
|
b = flatbuffers.Builder(0) |
|
b.StartObject(2) |
|
b.PrependInt8Slot(0, 33, 0) |
|
b.PrependInt8Slot(1, 44, 0) |
|
off = b.EndObject() |
|
b.Finish(off) |
|
|
|
b.StartObject(3) |
|
b.PrependInt8Slot(0, 55, 0) |
|
b.PrependInt8Slot(1, 66, 0) |
|
b.PrependInt8Slot(2, 77, 0) |
|
off = b.EndObject() |
|
b.Finish(off) |
|
|
|
self.assertBuilderEquals(b, [ |
|
16, 0, 0, 0, # root of table: points to object |
|
0, 0, # padding |
|
|
|
10, 0, # vtable bytes |
|
8, 0, # size of object |
|
7, 0, # start of value 0 |
|
6, 0, # start of value 1 |
|
5, 0, # start of value 2 |
|
10, 0, 0, 0, # offset for start of vtable (int32) |
|
0, # padding |
|
77, # value 2 |
|
66, # value 1 |
|
55, # value 0 |
|
|
|
12, 0, 0, 0, # root of table: points to object |
|
|
|
8, 0, # vtable bytes |
|
8, 0, # size of object |
|
7, 0, # start of value 0 |
|
6, 0, # start of value 1 |
|
8, 0, 0, 0, # offset for start of vtable (int32) |
|
0, 0, # padding |
|
44, # value 1 |
|
33, # value 0 |
|
]) |
|
|
|
def test_a_bunch_of_bools(self): |
|
b = flatbuffers.Builder(0) |
|
b.StartObject(8) |
|
b.PrependBoolSlot(0, True, False) |
|
b.PrependBoolSlot(1, True, False) |
|
b.PrependBoolSlot(2, True, False) |
|
b.PrependBoolSlot(3, True, False) |
|
b.PrependBoolSlot(4, True, False) |
|
b.PrependBoolSlot(5, True, False) |
|
b.PrependBoolSlot(6, True, False) |
|
b.PrependBoolSlot(7, True, False) |
|
off = b.EndObject() |
|
b.Finish(off) |
|
|
|
self.assertBuilderEquals(b, [ |
|
24, 0, 0, 0, # root of table: points to vtable offset |
|
|
|
20, 0, # vtable bytes |
|
12, 0, # size of object |
|
11, 0, # start of value 0 |
|
10, 0, # start of value 1 |
|
9, 0, # start of value 2 |
|
8, 0, # start of value 3 |
|
7, 0, # start of value 4 |
|
6, 0, # start of value 5 |
|
5, 0, # start of value 6 |
|
4, 0, # start of value 7 |
|
20, 0, 0, 0, # vtable offset |
|
|
|
1, # value 7 |
|
1, # value 6 |
|
1, # value 5 |
|
1, # value 4 |
|
1, # value 3 |
|
1, # value 2 |
|
1, # value 1 |
|
1, # value 0 |
|
]) |
|
|
|
def test_three_bools(self): |
|
b = flatbuffers.Builder(0) |
|
b.StartObject(3) |
|
b.PrependBoolSlot(0, True, False) |
|
b.PrependBoolSlot(1, True, False) |
|
b.PrependBoolSlot(2, True, False) |
|
off = b.EndObject() |
|
b.Finish(off) |
|
|
|
self.assertBuilderEquals(b, [ |
|
16, 0, 0, 0, # root of table: points to vtable offset |
|
|
|
0, 0, # padding |
|
|
|
10, 0, # vtable bytes |
|
8, 0, # size of object |
|
7, 0, # start of value 0 |
|
6, 0, # start of value 1 |
|
5, 0, # start of value 2 |
|
10, 0, 0, 0, # vtable offset from here |
|
|
|
0, # padding |
|
1, # value 2 |
|
1, # value 1 |
|
1, # value 0 |
|
]) |
|
|
|
def test_some_floats(self): |
|
b = flatbuffers.Builder(0) |
|
b.StartObject(1) |
|
b.PrependFloat32Slot(0, 1.0, 0.0) |
|
off = b.EndObject() |
|
|
|
self.assertBuilderEquals(b, [ |
|
6, 0, # vtable bytes |
|
8, 0, # size of object |
|
4, 0, # start of value 0 |
|
6, 0, 0, 0, # vtable offset |
|
|
|
0, 0, 128, 63, # value 0 |
|
]) |
|
|
|
|
|
def make_monster_from_generated_code(): |
|
''' Use generated code to build the example Monster. ''' |
|
|
|
b = flatbuffers.Builder(0) |
|
string = b.CreateString("MyMonster") |
|
test1 = b.CreateString("test1") |
|
test2 = b.CreateString("test2") |
|
fred = b.CreateString("Fred") |
|
|
|
MyGame.Example.Monster.MonsterStartInventoryVector(b, 5) |
|
b.PrependByte(4) |
|
b.PrependByte(3) |
|
b.PrependByte(2) |
|
b.PrependByte(1) |
|
b.PrependByte(0) |
|
inv = b.EndVector(5) |
|
|
|
MyGame.Example.Monster.MonsterStart(b) |
|
MyGame.Example.Monster.MonsterAddName(b, fred) |
|
mon2 = MyGame.Example.Monster.MonsterEnd(b) |
|
|
|
MyGame.Example.Monster.MonsterStartTest4Vector(b, 2) |
|
MyGame.Example.Test.CreateTest(b, 10, 20) |
|
MyGame.Example.Test.CreateTest(b, 30, 40) |
|
test4 = b.EndVector(2) |
|
|
|
MyGame.Example.Monster.MonsterStartTestarrayofstringVector(b, 2) |
|
b.PrependUOffsetTRelative(test2) |
|
b.PrependUOffsetTRelative(test1) |
|
testArrayOfString = b.EndVector(2) |
|
|
|
MyGame.Example.Monster.MonsterStart(b) |
|
|
|
pos = MyGame.Example.Vec3.CreateVec3(b, 1.0, 2.0, 3.0, 3.0, 2, 5, 6) |
|
MyGame.Example.Monster.MonsterAddPos(b, pos) |
|
|
|
MyGame.Example.Monster.MonsterAddHp(b, 80) |
|
MyGame.Example.Monster.MonsterAddName(b, string) |
|
MyGame.Example.Monster.MonsterAddInventory(b, inv) |
|
MyGame.Example.Monster.MonsterAddTestType(b, 1) |
|
MyGame.Example.Monster.MonsterAddTest(b, mon2) |
|
MyGame.Example.Monster.MonsterAddTest4(b, test4) |
|
MyGame.Example.Monster.MonsterAddTestarrayofstring(b, testArrayOfString) |
|
mon = MyGame.Example.Monster.MonsterEnd(b) |
|
|
|
b.Finish(mon) |
|
|
|
return b.Bytes, b.Head() |
|
|
|
|
|
class TestAllCodePathsOfExampleSchema(unittest.TestCase): |
|
def setUp(self, *args, **kwargs): |
|
super(TestAllCodePathsOfExampleSchema, self).setUp(*args, **kwargs) |
|
|
|
b = flatbuffers.Builder(0) |
|
MyGame.Example.Monster.MonsterStart(b) |
|
gen_mon = MyGame.Example.Monster.MonsterEnd(b) |
|
b.Finish(gen_mon) |
|
|
|
self.mon = MyGame.Example.Monster.Monster.GetRootAsMonster(b.Bytes, |
|
b.Head()) |
|
|
|
def test_default_monster_pos(self): |
|
self.assertTrue(self.mon.Pos() is None) |
|
|
|
def test_nondefault_monster_mana(self): |
|
b = flatbuffers.Builder(0) |
|
MyGame.Example.Monster.MonsterStart(b) |
|
MyGame.Example.Monster.MonsterAddMana(b, 50) |
|
mon = MyGame.Example.Monster.MonsterEnd(b) |
|
b.Finish(mon) |
|
|
|
got_mon = MyGame.Example.Monster.Monster.GetRootAsMonster(b.Bytes, |
|
b.Head()) |
|
self.assertEqual(50, got_mon.Mana()) |
|
|
|
def test_default_monster_hp(self): |
|
self.assertEqual(100, self.mon.Hp()) |
|
|
|
def test_default_monster_name(self): |
|
self.assertEqual('', self.mon.Name()) |
|
|
|
def test_default_monster_inventory_item(self): |
|
self.assertEqual(0, self.mon.Inventory(0)) |
|
|
|
def test_default_monster_inventory_length(self): |
|
self.assertEqual(0, self.mon.InventoryLength()) |
|
|
|
def test_default_monster_color(self): |
|
self.assertEqual(MyGame.Example.Color.Color.Blue, self.mon.Color()) |
|
|
|
def test_nondefault_monster_color(self): |
|
b = flatbuffers.Builder(0) |
|
color = MyGame.Example.Color.Color.Red |
|
MyGame.Example.Monster.MonsterStart(b) |
|
MyGame.Example.Monster.MonsterAddColor(b, color) |
|
mon = MyGame.Example.Monster.MonsterEnd(b) |
|
b.Finish(mon) |
|
|
|
mon2 = MyGame.Example.Monster.Monster.GetRootAsMonster(b.Bytes, |
|
b.Head()) |
|
self.assertEqual(MyGame.Example.Color.Color.Red, mon2.Color()) |
|
|
|
def test_default_monster_testtype(self): |
|
self.assertEqual(0, self.mon.TestType()) |
|
|
|
def test_default_monster_test_field(self): |
|
self.assertEqual(None, self.mon.Test()) |
|
|
|
def test_default_monster_test4_item(self): |
|
self.assertEqual(None, self.mon.Test4(0)) |
|
|
|
def test_default_monster_test4_length(self): |
|
self.assertEqual(0, self.mon.Test4Length()) |
|
|
|
def test_default_monster_testarrayofstring(self): |
|
self.assertEqual("", self.mon.Testarrayofstring(0)) |
|
|
|
def test_default_monster_testarrayofstring_length(self): |
|
self.assertEqual(0, self.mon.TestarrayofstringLength()) |
|
|
|
def test_default_monster_testarrayoftables(self): |
|
self.assertEqual(None, self.mon.Testarrayoftables(0)) |
|
|
|
def test_nondefault_monster_testarrayoftables(self): |
|
b = flatbuffers.Builder(0) |
|
|
|
# make a child Monster within a vector of Monsters: |
|
MyGame.Example.Monster.MonsterStart(b) |
|
MyGame.Example.Monster.MonsterAddHp(b, 99) |
|
sub_monster = MyGame.Example.Monster.MonsterEnd(b) |
|
|
|
# build the vector: |
|
MyGame.Example.Monster.MonsterStartTestarrayoftablesVector(b, 1) |
|
b.PrependUOffsetTRelative(sub_monster) |
|
vec = b.EndVector(1) |
|
|
|
# make the parent monster and include the vector of Monster: |
|
MyGame.Example.Monster.MonsterStart(b) |
|
MyGame.Example.Monster.MonsterAddTestarrayoftables(b, vec) |
|
mon = MyGame.Example.Monster.MonsterEnd(b) |
|
b.Finish(mon) |
|
|
|
# inspect the resulting data: |
|
mon2 = MyGame.Example.Monster.Monster.GetRootAsMonster(b.Output(), 0) |
|
self.assertEqual(99, mon2.Testarrayoftables(0).Hp()) |
|
self.assertEqual(1, mon2.TestarrayoftablesLength()) |
|
|
|
def test_default_monster_testarrayoftables_length(self): |
|
self.assertEqual(0, self.mon.TestarrayoftablesLength()) |
|
|
|
def test_nondefault_monster_enemy(self): |
|
b = flatbuffers.Builder(0) |
|
|
|
# make an Enemy object: |
|
MyGame.Example.Monster.MonsterStart(b) |
|
MyGame.Example.Monster.MonsterAddHp(b, 88) |
|
enemy = MyGame.Example.Monster.MonsterEnd(b) |
|
b.Finish(enemy) |
|
|
|
# make the parent monster and include the vector of Monster: |
|
MyGame.Example.Monster.MonsterStart(b) |
|
MyGame.Example.Monster.MonsterAddEnemy(b, enemy) |
|
mon = MyGame.Example.Monster.MonsterEnd(b) |
|
b.Finish(mon) |
|
|
|
# inspect the resulting data: |
|
mon2 = MyGame.Example.Monster.Monster.GetRootAsMonster(b.Bytes, |
|
b.Head()) |
|
self.assertEqual(88, mon2.Enemy().Hp()) |
|
|
|
def test_default_monster_testnestedflatbuffer(self): |
|
self.assertEqual(0, self.mon.Testnestedflatbuffer(0)) |
|
|
|
def test_default_monster_testnestedflatbuffer_length(self): |
|
self.assertEqual(0, self.mon.TestnestedflatbufferLength()) |
|
|
|
def test_nondefault_monster_testnestedflatbuffer(self): |
|
b = flatbuffers.Builder(0) |
|
|
|
MyGame.Example.Monster.MonsterStartTestnestedflatbufferVector(b, 3) |
|
b.PrependByte(4) |
|
b.PrependByte(2) |
|
b.PrependByte(0) |
|
sub_buf = b.EndVector(3) |
|
|
|
# make the parent monster and include the vector of Monster: |
|
MyGame.Example.Monster.MonsterStart(b) |
|
MyGame.Example.Monster.MonsterAddTestnestedflatbuffer(b, sub_buf) |
|
mon = MyGame.Example.Monster.MonsterEnd(b) |
|
b.Finish(mon) |
|
|
|
# inspect the resulting data: |
|
mon2 = MyGame.Example.Monster.Monster.GetRootAsMonster(b.Bytes, |
|
b.Head()) |
|
self.assertEqual(3, mon2.TestnestedflatbufferLength()) |
|
self.assertEqual(0, mon2.Testnestedflatbuffer(0)) |
|
self.assertEqual(2, mon2.Testnestedflatbuffer(1)) |
|
self.assertEqual(4, mon2.Testnestedflatbuffer(2)) |
|
|
|
def test_nondefault_monster_testempty(self): |
|
b = flatbuffers.Builder(0) |
|
|
|
# make a Stat object: |
|
MyGame.Example.Stat.StatStart(b) |
|
MyGame.Example.Stat.StatAddVal(b, 123) |
|
my_stat = MyGame.Example.Stat.StatEnd(b) |
|
b.Finish(my_stat) |
|
|
|
# include the stat object in a monster: |
|
MyGame.Example.Monster.MonsterStart(b) |
|
MyGame.Example.Monster.MonsterAddTestempty(b, my_stat) |
|
mon = MyGame.Example.Monster.MonsterEnd(b) |
|
b.Finish(mon) |
|
|
|
# inspect the resulting data: |
|
mon2 = MyGame.Example.Monster.Monster.GetRootAsMonster(b.Bytes, |
|
b.Head()) |
|
self.assertEqual(123, mon2.Testempty().Val()) |
|
|
|
def test_default_monster_testbool(self): |
|
self.assertFalse(self.mon.Testbool()) |
|
|
|
def test_nondefault_monster_testbool(self): |
|
b = flatbuffers.Builder(0) |
|
MyGame.Example.Monster.MonsterStart(b) |
|
MyGame.Example.Monster.MonsterAddTestbool(b, True) |
|
mon = MyGame.Example.Monster.MonsterEnd(b) |
|
b.Finish(mon) |
|
|
|
# inspect the resulting data: |
|
mon2 = MyGame.Example.Monster.Monster.GetRootAsMonster(b.Bytes, |
|
b.Head()) |
|
self.assertTrue(mon2.Testbool()) |
|
|
|
def test_default_monster_testhashes(self): |
|
self.assertEqual(0, self.mon.Testhashs32Fnv1()) |
|
self.assertEqual(0, self.mon.Testhashu32Fnv1()) |
|
self.assertEqual(0, self.mon.Testhashs64Fnv1()) |
|
self.assertEqual(0, self.mon.Testhashu64Fnv1()) |
|
self.assertEqual(0, self.mon.Testhashs32Fnv1a()) |
|
self.assertEqual(0, self.mon.Testhashu32Fnv1a()) |
|
self.assertEqual(0, self.mon.Testhashs64Fnv1a()) |
|
self.assertEqual(0, self.mon.Testhashu64Fnv1a()) |
|
|
|
def test_nondefault_monster_testhashes(self): |
|
b = flatbuffers.Builder(0) |
|
MyGame.Example.Monster.MonsterStart(b) |
|
MyGame.Example.Monster.MonsterAddTesthashs32Fnv1(b, 1) |
|
MyGame.Example.Monster.MonsterAddTesthashu32Fnv1(b, 2) |
|
MyGame.Example.Monster.MonsterAddTesthashs64Fnv1(b, 3) |
|
MyGame.Example.Monster.MonsterAddTesthashu64Fnv1(b, 4) |
|
MyGame.Example.Monster.MonsterAddTesthashs32Fnv1a(b, 5) |
|
MyGame.Example.Monster.MonsterAddTesthashu32Fnv1a(b, 6) |
|
MyGame.Example.Monster.MonsterAddTesthashs64Fnv1a(b, 7) |
|
MyGame.Example.Monster.MonsterAddTesthashu64Fnv1a(b, 8) |
|
mon = MyGame.Example.Monster.MonsterEnd(b) |
|
b.Finish(mon) |
|
|
|
# inspect the resulting data: |
|
mon2 = MyGame.Example.Monster.Monster.GetRootAsMonster(b.Bytes, |
|
b.Head()) |
|
self.assertEqual(1, mon2.Testhashs32Fnv1()) |
|
self.assertEqual(2, mon2.Testhashu32Fnv1()) |
|
self.assertEqual(3, mon2.Testhashs64Fnv1()) |
|
self.assertEqual(4, mon2.Testhashu64Fnv1()) |
|
self.assertEqual(5, mon2.Testhashs32Fnv1a()) |
|
self.assertEqual(6, mon2.Testhashu32Fnv1a()) |
|
self.assertEqual(7, mon2.Testhashs64Fnv1a()) |
|
self.assertEqual(8, mon2.Testhashu64Fnv1a()) |
|
|
|
def test_getrootas_for_nonroot_table(self): |
|
b = flatbuffers.Builder(0) |
|
string = b.CreateString("MyStat") |
|
|
|
MyGame.Example.Stat.StatStart(b) |
|
MyGame.Example.Stat.StatAddId(b, string) |
|
MyGame.Example.Stat.StatAddVal(b, 12345678) |
|
MyGame.Example.Stat.StatAddCount(b, 12345) |
|
stat = MyGame.Example.Stat.StatEnd(b) |
|
b.Finish(stat) |
|
|
|
stat2 = MyGame.Example.Stat.Stat.GetRootAsStat(b.Bytes, b.Head()) |
|
|
|
self.assertEqual(b"MyStat", stat2.Id()) |
|
self.assertEqual(12345678, stat2.Val()) |
|
self.assertEqual(12345, stat2.Count()) |
|
|
|
|
|
class TestVtableDeduplication(unittest.TestCase): |
|
''' TestVtableDeduplication verifies that vtables are deduplicated. ''' |
|
|
|
def test_vtable_deduplication(self): |
|
b = flatbuffers.Builder(0) |
|
|
|
b.StartObject(4) |
|
b.PrependByteSlot(0, 0, 0) |
|
b.PrependByteSlot(1, 11, 0) |
|
b.PrependByteSlot(2, 22, 0) |
|
b.PrependInt16Slot(3, 33, 0) |
|
obj0 = b.EndObject() |
|
|
|
b.StartObject(4) |
|
b.PrependByteSlot(0, 0, 0) |
|
b.PrependByteSlot(1, 44, 0) |
|
b.PrependByteSlot(2, 55, 0) |
|
b.PrependInt16Slot(3, 66, 0) |
|
obj1 = b.EndObject() |
|
|
|
b.StartObject(4) |
|
b.PrependByteSlot(0, 0, 0) |
|
b.PrependByteSlot(1, 77, 0) |
|
b.PrependByteSlot(2, 88, 0) |
|
b.PrependInt16Slot(3, 99, 0) |
|
obj2 = b.EndObject() |
|
|
|
got = b.Bytes[b.Head():] |
|
|
|
want = bytearray([ |
|
240, 255, 255, 255, # == -12. offset to dedupped vtable. |
|
99, 0, |
|
88, |
|
77, |
|
248, 255, 255, 255, # == -8. offset to dedupped vtable. |
|
66, 0, |
|
55, |
|
44, |
|
12, 0, |
|
8, 0, |
|
0, 0, |
|
7, 0, |
|
6, 0, |
|
4, 0, |
|
12, 0, 0, 0, |
|
33, 0, |
|
22, |
|
11, |
|
]) |
|
|
|
self.assertEqual((len(want), want), (len(got), got)) |
|
|
|
table0 = flatbuffers.table.Table(b.Bytes, len(b.Bytes) - obj0) |
|
table1 = flatbuffers.table.Table(b.Bytes, len(b.Bytes) - obj1) |
|
table2 = flatbuffers.table.Table(b.Bytes, len(b.Bytes) - obj2) |
|
|
|
def _checkTable(tab, voffsett_value, b, c, d): |
|
# vtable size |
|
got = tab.GetVOffsetTSlot(0, 0) |
|
self.assertEqual(12, got, 'case 0, 0') |
|
|
|
# object size |
|
got = tab.GetVOffsetTSlot(2, 0) |
|
self.assertEqual(8, got, 'case 2, 0') |
|
|
|
# default value |
|
got = tab.GetVOffsetTSlot(4, 0) |
|
self.assertEqual(voffsett_value, got, 'case 4, 0') |
|
|
|
got = tab.GetSlot(6, 0, N.Uint8Flags) |
|
self.assertEqual(b, got, 'case 6, 0') |
|
|
|
val = tab.GetSlot(8, 0, N.Uint8Flags) |
|
self.assertEqual(c, val, 'failed 8, 0') |
|
|
|
got = tab.GetSlot(10, 0, N.Uint8Flags) |
|
self.assertEqual(d, got, 'failed 10, 0') |
|
|
|
_checkTable(table0, 0, 11, 22, 33) |
|
_checkTable(table1, 0, 44, 55, 66) |
|
_checkTable(table2, 0, 77, 88, 99) |
|
|
|
|
|
class TestExceptions(unittest.TestCase): |
|
def test_object_is_nested_error(self): |
|
b = flatbuffers.Builder(0) |
|
b.StartObject(0) |
|
assertRaises(self, lambda: b.StartObject(0), |
|
flatbuffers.builder.IsNestedError) |
|
|
|
def test_object_is_not_nested_error(self): |
|
b = flatbuffers.Builder(0) |
|
assertRaises(self, lambda: b.EndObject(), |
|
flatbuffers.builder.IsNotNestedError) |
|
|
|
def test_struct_is_not_inline_error(self): |
|
b = flatbuffers.Builder(0) |
|
b.StartObject(0) |
|
assertRaises(self, lambda: b.PrependStructSlot(0, 1, 0), |
|
flatbuffers.builder.StructIsNotInlineError) |
|
|
|
def test_unreachable_error(self): |
|
b = flatbuffers.Builder(0) |
|
assertRaises(self, lambda: b.PrependUOffsetTRelative(1), |
|
flatbuffers.builder.OffsetArithmeticError) |
|
|
|
def test_create_string_is_nested_error(self): |
|
b = flatbuffers.Builder(0) |
|
b.StartObject(0) |
|
s = 'test1' |
|
assertRaises(self, lambda: b.CreateString(s), |
|
flatbuffers.builder.IsNestedError) |
|
|
|
def test_finished_bytes_error(self): |
|
b = flatbuffers.Builder(0) |
|
assertRaises(self, lambda: b.Output(), |
|
flatbuffers.builder.BuilderNotFinishedError) |
|
|
|
|
|
def CheckAgainstGoldDataGo(): |
|
try: |
|
gen_buf, gen_off = make_monster_from_generated_code() |
|
fn = 'monsterdata_go_wire.mon' |
|
if not os.path.exists(fn): |
|
print('Go-generated data does not exist, failed.') |
|
return False |
|
|
|
# would like to use a context manager here, but it's less |
|
# backwards-compatible: |
|
f = open(fn, 'rb') |
|
go_wire_data = f.read() |
|
f.close() |
|
|
|
CheckReadBuffer(bytearray(go_wire_data), 0) |
|
if not bytearray(gen_buf[gen_off:]) == bytearray(go_wire_data): |
|
raise AssertionError('CheckAgainstGoldDataGo failed') |
|
except: |
|
print('Failed to test against Go-generated test data.') |
|
return False |
|
|
|
print('Can read Go-generated test data, and Python generates bytewise identical data.') |
|
return True |
|
|
|
|
|
def CheckAgainstGoldDataJava(): |
|
try: |
|
gen_buf, gen_off = make_monster_from_generated_code() |
|
fn = 'monsterdata_java_wire.mon' |
|
if not os.path.exists(fn): |
|
print('Java-generated data does not exist, failed.') |
|
return False |
|
f = open(fn, 'rb') |
|
java_wire_data = f.read() |
|
f.close() |
|
|
|
CheckReadBuffer(bytearray(java_wire_data), 0) |
|
except: |
|
print('Failed to read Java-generated test data.') |
|
return False |
|
|
|
print('Can read Java-generated test data.') |
|
return True |
|
|
|
|
|
class LCG(object): |
|
''' Include simple random number generator to ensure results will be the |
|
same cross platform. |
|
http://en.wikipedia.org/wiki/Park%E2%80%93Miller_random_number_generator ''' |
|
|
|
__slots__ = ['n'] |
|
|
|
InitialLCGSeed = 48271 |
|
|
|
def __init__(self): |
|
self.n = self.InitialLCGSeed |
|
|
|
def Reset(self): |
|
self.n = self.InitialLCGSeed |
|
|
|
def Next(self): |
|
self.n = ((self.n * 279470273) % 4294967291) & 0xFFFFFFFF |
|
return self.n |
|
|
|
|
|
def BenchmarkVtableDeduplication(count): |
|
''' |
|
BenchmarkVtableDeduplication measures the speed of vtable deduplication |
|
by creating `prePop` vtables, then populating `count` objects with a |
|
different single vtable. |
|
|
|
When count is large (as in long benchmarks), memory usage may be high. |
|
''' |
|
|
|
prePop = 10 |
|
builder = flatbuffers.Builder(0) |
|
|
|
# pre-populate some vtables: |
|
for i in compat_range(prePop): |
|
builder.StartObject(i) |
|
for j in compat_range(i): |
|
builder.PrependInt16Slot(j, j, 0) |
|
builder.EndObject() |
|
|
|
# benchmark deduplication of a new vtable: |
|
def f(): |
|
builder.StartObject(prePop) |
|
for j in compat_range(prePop): |
|
builder.PrependInt16Slot(j, j, 0) |
|
builder.EndObject() |
|
|
|
duration = timeit.timeit(stmt=f, number=count) |
|
rate = float(count) / duration |
|
print(('vtable deduplication rate: %.2f/sec' % rate)) |
|
|
|
|
|
def BenchmarkCheckReadBuffer(count, buf, off): |
|
''' |
|
BenchmarkCheckReadBuffer measures the speed of flatbuffer reading |
|
by re-using the CheckReadBuffer function with the gold data. |
|
''' |
|
|
|
def f(): |
|
CheckReadBuffer(buf, off) |
|
|
|
duration = timeit.timeit(stmt=f, number=count) |
|
rate = float(count) / duration |
|
data = float(len(buf) * count) / float(1024 * 1024) |
|
data_rate = data / float(duration) |
|
|
|
print(('traversed %d %d-byte flatbuffers in %.2fsec: %.2f/sec, %.2fMB/sec') |
|
% (count, len(buf), duration, rate, data_rate)) |
|
|
|
|
|
def BenchmarkMakeMonsterFromGeneratedCode(count, length): |
|
''' |
|
BenchmarkMakeMonsterFromGeneratedCode measures the speed of flatbuffer |
|
creation by re-using the make_monster_from_generated_code function for |
|
generating gold data examples. |
|
''' |
|
|
|
duration = timeit.timeit(stmt=make_monster_from_generated_code, |
|
number=count) |
|
rate = float(count) / duration |
|
data = float(length * count) / float(1024 * 1024) |
|
data_rate = data / float(duration) |
|
|
|
print(('built %d %d-byte flatbuffers in %.2fsec: %.2f/sec, %.2fMB/sec' % \ |
|
(count, length, duration, rate, data_rate))) |
|
|
|
|
|
def backward_compatible_run_tests(**kwargs): |
|
if PY_VERSION < (2, 6): |
|
sys.stderr.write("Python version less than 2.6 are not supported") |
|
sys.stderr.flush() |
|
return False |
|
|
|
# python2.6 has a reduced-functionality unittest.main function: |
|
if PY_VERSION == (2, 6): |
|
try: |
|
unittest.main(**kwargs) |
|
except SystemExit as e: |
|
if not e.code == 0: |
|
return False |
|
return True |
|
|
|
# python2.7 and above let us not exit once unittest.main is run: |
|
kwargs['exit'] = False |
|
kwargs['verbosity'] = 0 |
|
ret = unittest.main(**kwargs) |
|
if ret.result.errors or ret.result.failures: |
|
return False |
|
|
|
return True |
|
|
|
def main(): |
|
import os |
|
import sys |
|
if not len(sys.argv) == 4: |
|
sys.stderr.write('Usage: %s <benchmark vtable count>' |
|
'<benchmark read count> <benchmark build count>\n' |
|
% sys.argv[0]) |
|
sys.stderr.write(' Provide COMPARE_GENERATED_TO_GO=1 to check' |
|
'for bytewise comparison to Go data.\n') |
|
sys.stderr.write(' Provide COMPARE_GENERATED_TO_JAVA=1 to check' |
|
'for bytewise comparison to Java data.\n') |
|
sys.stderr.flush() |
|
sys.exit(1) |
|
|
|
kwargs = dict(argv=sys.argv[:-3]) |
|
|
|
# run tests, and run some language comparison checks if needed: |
|
success = backward_compatible_run_tests(**kwargs) |
|
if success and os.environ.get('COMPARE_GENERATED_TO_GO', 0) == "1": |
|
success = success and CheckAgainstGoldDataGo() |
|
if success and os.environ.get('COMPARE_GENERATED_TO_JAVA', 0) == "1": |
|
success = success and CheckAgainstGoldDataJava() |
|
|
|
if not success: |
|
sys.stderr.write('Tests failed, skipping benchmarks.\n') |
|
sys.stderr.flush() |
|
sys.exit(1) |
|
|
|
# run benchmarks (if 0, they will be a noop): |
|
bench_vtable = int(sys.argv[1]) |
|
bench_traverse = int(sys.argv[2]) |
|
bench_build = int(sys.argv[3]) |
|
if bench_vtable: |
|
BenchmarkVtableDeduplication(bench_vtable) |
|
if bench_traverse: |
|
buf, off = make_monster_from_generated_code() |
|
BenchmarkCheckReadBuffer(bench_traverse, buf, off) |
|
if bench_build: |
|
buf, off = make_monster_from_generated_code() |
|
BenchmarkMakeMonsterFromGeneratedCode(bench_build, len(buf)) |
|
|
|
if __name__ == '__main__': |
|
main()
|
|
|