pycapnp/test/test_regression.py
2023-03-12 11:33:41 -07:00

579 lines
20 KiB
Python

# -*- coding: utf-8 -*-
import pytest
import capnp
import os
import math
import sys
this_dir = os.path.dirname(__file__)
if sys.version_info[0] < 3:
EXPECT_BYTES = True
else:
EXPECT_BYTES = False
@pytest.fixture
def addressbook():
return capnp.load(os.path.join(this_dir, "addressbook.capnp"))
def test_addressbook_message_classes(addressbook):
def writeAddressBook(fd):
message = capnp._MallocMessageBuilder()
addressBook = message.init_root(addressbook.AddressBook)
people = addressBook.init("people", 2)
alice = people[0]
alice.id = 123
alice.name = "Alice"
alice.email = "alice@example.com"
alicePhones = alice.init("phones", 1)
alicePhones[0].number = "555-1212"
alicePhones[0].type = "mobile"
alice.employment.school = "MIT"
bob = people[1]
bob.id = 456
bob.name = "Bob"
bob.email = "bob@example.com"
bobPhones = bob.init("phones", 2)
bobPhones[0].number = "555-4567"
bobPhones[0].type = "home"
bobPhones[1].number = "555-7654"
bobPhones[1].type = "work"
bob.employment.unemployed = None
capnp._write_packed_message_to_fd(fd, message)
def printAddressBook(fd):
message = capnp._PackedFdMessageReader(f)
addressBook = message.get_root(addressbook.AddressBook)
people = addressBook.people
alice = people[0]
assert alice.id == 123
assert alice.name == "Alice"
assert alice.email == "alice@example.com"
alicePhones = alice.phones
assert alicePhones[0].number == "555-1212"
assert alicePhones[0].type == "mobile"
assert alice.employment.school == "MIT"
bob = people[1]
assert bob.id == 456
assert bob.name == "Bob"
assert bob.email == "bob@example.com"
bobPhones = bob.phones
assert bobPhones[0].number == "555-4567"
assert bobPhones[0].type == "home"
assert bobPhones[1].number == "555-7654"
assert bobPhones[1].type == "work"
assert bob.employment.unemployed is None
f = open("example", "w")
writeAddressBook(f.fileno())
f = open("example", "r")
printAddressBook(f.fileno())
def test_addressbook(addressbook):
def writeAddressBook(file):
addresses = addressbook.AddressBook.new_message()
people = addresses.init("people", 2)
alice = people[0]
alice.id = 123
alice.name = "Alice"
alice.email = "alice@example.com"
alicePhones = alice.init("phones", 1)
alicePhones[0].number = "555-1212"
alicePhones[0].type = "mobile"
alice.employment.school = "MIT"
bob = people[1]
bob.id = 456
bob.name = "Bob"
bob.email = "bob@example.com"
bobPhones = bob.init("phones", 2)
bobPhones[0].number = "555-4567"
bobPhones[0].type = "home"
bobPhones[1].number = "555-7654"
bobPhones[1].type = "work"
bob.employment.unemployed = None
addresses.write(file)
def printAddressBook(file):
addresses = addressbook.AddressBook.read(file)
people = addresses.people
alice = people[0]
assert alice.id == 123
assert alice.name == "Alice"
assert alice.email == "alice@example.com"
alicePhones = alice.phones
assert alicePhones[0].number == "555-1212"
assert alicePhones[0].type == "mobile"
assert alice.employment.school == "MIT"
bob = people[1]
assert bob.id == 456
assert bob.name == "Bob"
assert bob.email == "bob@example.com"
bobPhones = bob.phones
assert bobPhones[0].number == "555-4567"
assert bobPhones[0].type == "home"
assert bobPhones[1].number == "555-7654"
assert bobPhones[1].type == "work"
assert bob.employment.unemployed is None
f = open("example", "w")
writeAddressBook(f)
f = open("example", "r")
printAddressBook(f)
def test_addressbook_resizable(addressbook):
def writeAddressBook(file):
addresses = addressbook.AddressBook.new_message()
people = addresses.init_resizable_list("people")
alice = people.add()
alice.id = 123
alice.name = "Alice"
alice.email = "alice@example.com"
alicePhones = alice.init("phones", 1)
alicePhones[0].number = "555-1212"
alicePhones[0].type = "mobile"
alice.employment.school = "MIT"
bob = people.add()
bob.id = 456
bob.name = "Bob"
bob.email = "bob@example.com"
bobPhones = bob.init("phones", 2)
bobPhones[0].number = "555-4567"
bobPhones[0].type = "home"
bobPhones[1].number = "555-7654"
bobPhones[1].type = "work"
bob.employment.unemployed = None
people.finish()
addresses.write(file)
def printAddressBook(file):
addresses = addressbook.AddressBook.read(file)
people = addresses.people
alice = people[0]
assert alice.id == 123
assert alice.name == "Alice"
assert alice.email == "alice@example.com"
alicePhones = alice.phones
assert alicePhones[0].number == "555-1212"
assert alicePhones[0].type == "mobile"
assert alice.employment.school == "MIT"
bob = people[1]
assert bob.id == 456
assert bob.name == "Bob"
assert bob.email == "bob@example.com"
bobPhones = bob.phones
assert bobPhones[0].number == "555-4567"
assert bobPhones[0].type == "home"
assert bobPhones[1].number == "555-7654"
assert bobPhones[1].type == "work"
assert bob.employment.unemployed is None
f = open("example", "w")
writeAddressBook(f)
f = open("example", "r")
printAddressBook(f)
def test_addressbook_explicit_fields(addressbook):
def writeAddressBook(file):
addresses = addressbook.AddressBook.new_message()
address_fields = addressbook.AddressBook.schema.fields
person_fields = addressbook.Person.schema.fields
phone_fields = addressbook.Person.PhoneNumber.schema.fields
people = addresses._init_by_field(address_fields["people"], 2)
alice = people[0]
alice._set_by_field(person_fields["id"], 123)
alice._set_by_field(person_fields["name"], "Alice")
alice._set_by_field(person_fields["email"], "alice@example.com")
alicePhones = alice._init_by_field(person_fields["phones"], 1)
alicePhones[0]._set_by_field(phone_fields["number"], "555-1212")
alicePhones[0]._set_by_field(phone_fields["type"], "mobile")
employment = alice._get_by_field(person_fields["employment"])
employment._set_by_field(
addressbook.Person.Employment.schema.fields["school"], "MIT"
)
bob = people[1]
bob._set_by_field(person_fields["id"], 456)
bob._set_by_field(person_fields["name"], "Bob")
bob._set_by_field(person_fields["email"], "bob@example.com")
bobPhones = bob._init_by_field(person_fields["phones"], 2)
bobPhones[0]._set_by_field(phone_fields["number"], "555-4567")
bobPhones[0]._set_by_field(phone_fields["type"], "home")
bobPhones[1]._set_by_field(phone_fields["number"], "555-7654")
bobPhones[1]._set_by_field(phone_fields["type"], "work")
employment = bob._get_by_field(person_fields["employment"])
employment._set_by_field(
addressbook.Person.Employment.schema.fields["unemployed"], None
)
addresses.write(file)
def printAddressBook(file):
addresses = addressbook.AddressBook.read(file)
address_fields = addressbook.AddressBook.schema.fields
person_fields = addressbook.Person.schema.fields
phone_fields = addressbook.Person.PhoneNumber.schema.fields
people = addresses._get_by_field(address_fields["people"])
alice = people[0]
assert alice._get_by_field(person_fields["id"]) == 123
assert alice._get_by_field(person_fields["name"]) == "Alice"
assert alice._get_by_field(person_fields["email"]) == "alice@example.com"
alicePhones = alice._get_by_field(person_fields["phones"])
assert alicePhones[0]._get_by_field(phone_fields["number"]) == "555-1212"
assert alicePhones[0]._get_by_field(phone_fields["type"]) == "mobile"
employment = alice._get_by_field(person_fields["employment"])
employment._get_by_field(
addressbook.Person.Employment.schema.fields["school"]
) == "MIT"
bob = people[1]
assert bob._get_by_field(person_fields["id"]) == 456
assert bob._get_by_field(person_fields["name"]) == "Bob"
assert bob._get_by_field(person_fields["email"]) == "bob@example.com"
bobPhones = bob._get_by_field(person_fields["phones"])
assert bobPhones[0]._get_by_field(phone_fields["number"]) == "555-4567"
assert bobPhones[0]._get_by_field(phone_fields["type"]) == "home"
assert bobPhones[1]._get_by_field(phone_fields["number"]) == "555-7654"
assert bobPhones[1]._get_by_field(phone_fields["type"]) == "work"
employment = bob._get_by_field(person_fields["employment"])
employment._get_by_field(
addressbook.Person.Employment.schema.fields["unemployed"]
) is None
f = open("example", "w")
writeAddressBook(f)
f = open("example", "r")
printAddressBook(f)
@pytest.fixture
def all_types():
return capnp.load(os.path.join(this_dir, "all_types.capnp"))
# TODO: These tests should be extended to:
# - Read each field in Python and assert that it is equal to the expected value.
# - Build an identical message using Python code and compare it to the golden.
#
def init_all_types(builder):
builder.voidField = None
builder.boolField = True
builder.int8Field = -123
builder.int16Field = -12345
builder.int32Field = -12345678
builder.int64Field = -123456789012345
builder.uInt8Field = 234
builder.uInt16Field = 45678
builder.uInt32Field = 3456789012
builder.uInt64Field = 12345678901234567890
builder.float32Field = 1234.5
builder.float64Field = -123e45
builder.textField = "foo"
builder.dataField = b"bar"
subBuilder = builder.structField
subBuilder.voidField = None
subBuilder.boolField = True
subBuilder.int8Field = -12
subBuilder.int16Field = 3456
subBuilder.int32Field = -78901234
subBuilder.int64Field = 56789012345678
subBuilder.uInt8Field = 90
subBuilder.uInt16Field = 1234
subBuilder.uInt32Field = 56789012
subBuilder.uInt64Field = 345678901234567890
subBuilder.float32Field = -1.25e-10
subBuilder.float64Field = 345
subBuilder.textField = ""
subBuilder.dataField = b"qux"
subSubBuilder = subBuilder.structField
subSubBuilder.textField = "nested"
subSubBuilder.structField.textField = "really nested"
subBuilder.enumField = "baz"
subBuilder.voidList = [None, None, None]
subBuilder.boolList = [False, True, False, True, True]
subBuilder.int8List = [12, -34, -0x80, 0x7F]
subBuilder.int16List = [1234, -5678, -0x8000, 0x7FFF]
subBuilder.int32List = [12345678, -90123456, -0x80000000, 0x7FFFFFFF]
subBuilder.int64List = [
123456789012345,
-678901234567890,
-0x8000000000000000,
0x7FFFFFFFFFFFFFFF,
]
subBuilder.uInt8List = [12, 34, 0, 0xFF]
subBuilder.uInt16List = [1234, 5678, 0, 0xFFFF]
subBuilder.uInt32List = [12345678, 90123456, 0, 0xFFFFFFFF]
subBuilder.uInt64List = [123456789012345, 678901234567890, 0, 0xFFFFFFFFFFFFFFFF]
subBuilder.float32List = [0, 1234567, 1e37, -1e37, 1e-37, -1e-37]
subBuilder.float64List = [0, 123456789012345, 1e306, -1e306, 1e-306, -1e-306]
subBuilder.textList = ["quux", "corge", "grault"]
subBuilder.dataList = [b"garply", b"waldo", b"fred"]
listBuilder = subBuilder.init("structList", 3)
listBuilder[0].textField = "x structlist 1"
listBuilder[1].textField = "x structlist 2"
listBuilder[2].textField = "x structlist 3"
subBuilder.enumList = ["qux", "bar", "grault"]
builder.enumField = "corge"
builder.init("voidList", 6)
builder.boolList = [True, False, False, True]
builder.int8List = [111, -111]
builder.int16List = [11111, -11111]
builder.int32List = [111111111, -111111111]
builder.int64List = [1111111111111111111, -1111111111111111111]
builder.uInt8List = [111, 222]
builder.uInt16List = [33333, 44444]
builder.uInt32List = [3333333333]
builder.uInt64List = [11111111111111111111]
builder.float32List = [5555.5, float("inf"), float("-inf"), float("nan")]
builder.float64List = [7777.75, float("inf"), float("-inf"), float("nan")]
builder.textList = ["plugh", "xyzzy", "thud"]
builder.dataList = [b"oops", b"exhausted", b"rfc3092"]
listBuilder = builder.init("structList", 3)
listBuilder[0].textField = "structlist 1"
listBuilder[1].textField = "structlist 2"
listBuilder[2].textField = "structlist 3"
builder.enumList = ["foo", "garply"]
def assert_almost(float1, float2):
if float1 != float2:
assert abs((float1 - float2) / float1) < 0.00001
def check_list(reader, expected):
assert len(reader) == len(expected)
for i, v in enumerate(expected):
if type(v) is float:
assert_almost(reader[i], v)
else:
assert reader[i] == v
def check_all_types(reader):
assert reader.voidField is None
assert reader.boolField
assert reader.int8Field == -123
assert reader.int16Field == -12345
assert reader.int32Field == -12345678
assert reader.int64Field == -123456789012345
assert reader.uInt8Field == 234
assert reader.uInt16Field == 45678
assert reader.uInt32Field == 3456789012
assert reader.uInt64Field == 12345678901234567890
assert reader.float32Field == 1234.5
assert_almost(reader.float64Field, -123e45)
assert reader.textField == "foo"
assert reader.dataField == b"bar"
subReader = reader.structField
assert subReader.voidField is None
assert subReader.boolField
assert subReader.int8Field == -12
assert subReader.int16Field == 3456
assert subReader.int32Field == -78901234
assert subReader.int64Field == 56789012345678
assert subReader.uInt8Field == 90
assert subReader.uInt16Field == 1234
assert subReader.uInt32Field == 56789012
assert subReader.uInt64Field == 345678901234567890
assert_almost(subReader.float32Field, -1.25e-10)
assert subReader.float64Field == 345
assert subReader.textField == ""
# This assertion highlights the encoding we expect to see here, since
# otherwise this appears a bit magical...
if EXPECT_BYTES:
assert len(subReader.textField) == 3
else:
assert len(subReader.textField) == 1
assert subReader.dataField == b"qux"
subSubReader = subReader.structField
assert subSubReader.textField == "nested"
assert subSubReader.structField.textField == "really nested"
assert subReader.enumField == "baz"
# Check that enums are hashable and can be used as keys in dicts
# interchangably with their string version.
assert hash(subReader.enumField) == hash("baz")
assert {subReader.enumField: 17}.get(subReader.enumField) == 17
assert {subReader.enumField: 17}.get("baz") == 17
assert {"baz": 17}.get(subReader.enumField) == 17
check_list(subReader.voidList, [None, None, None])
check_list(subReader.boolList, [False, True, False, True, True])
check_list(subReader.int8List, [12, -34, -0x80, 0x7F])
check_list(subReader.int16List, [1234, -5678, -0x8000, 0x7FFF])
check_list(subReader.int32List, [12345678, -90123456, -0x80000000, 0x7FFFFFFF])
check_list(
subReader.int64List,
[123456789012345, -678901234567890, -0x8000000000000000, 0x7FFFFFFFFFFFFFFF],
)
check_list(subReader.uInt8List, [12, 34, 0, 0xFF])
check_list(subReader.uInt16List, [1234, 5678, 0, 0xFFFF])
check_list(subReader.uInt32List, [12345678, 90123456, 0, 0xFFFFFFFF])
check_list(
subReader.uInt64List, [123456789012345, 678901234567890, 0, 0xFFFFFFFFFFFFFFFF]
)
check_list(subReader.float32List, [0.0, 1234567.0, 1e37, -1e37, 1e-37, -1e-37])
check_list(
subReader.float64List, [0.0, 123456789012345.0, 1e306, -1e306, 1e-306, -1e-306]
)
check_list(subReader.textList, ["quux", "corge", "grault"])
check_list(subReader.dataList, [b"garply", b"waldo", b"fred"])
listReader = subReader.structList
assert len(listReader) == 3
assert listReader[0].textField == "x structlist 1"
assert listReader[1].textField == "x structlist 2"
assert listReader[2].textField == "x structlist 3"
check_list(subReader.enumList, ["qux", "bar", "grault"])
assert reader.enumField == "corge"
assert len(reader.voidList) == 6
check_list(reader.boolList, [True, False, False, True])
check_list(reader.int8List, [111, -111])
check_list(reader.int16List, [11111, -11111])
check_list(reader.int32List, [111111111, -111111111])
check_list(reader.int64List, [1111111111111111111, -1111111111111111111])
check_list(reader.uInt8List, [111, 222])
check_list(reader.uInt16List, [33333, 44444])
check_list(reader.uInt32List, [3333333333])
check_list(reader.uInt64List, [11111111111111111111])
listReader = reader.float32List
assert len(listReader) == 4
assert listReader[0] == 5555.5
assert listReader[1] == float("inf")
assert listReader[2] == -float("inf")
assert math.isnan(listReader[3])
listReader = reader.float64List
len(listReader) == 4
assert listReader[0] == 7777.75
assert listReader[1] == float("inf")
assert listReader[2] == -float("inf")
assert math.isnan(listReader[3])
check_list(reader.textList, ["plugh", "xyzzy", "thud"])
check_list(reader.dataList, [b"oops", b"exhausted", b"rfc3092"])
listReader = reader.structList
len(listReader) == 3
assert listReader[0].textField == "structlist 1"
assert listReader[1].textField == "structlist 2"
assert listReader[2].textField == "structlist 3"
check_list(reader.enumList, ["foo", "garply"])
def test_build(all_types):
root = all_types.TestAllTypes.new_message()
init_all_types(root)
expectedText = open(
os.path.join(this_dir, "all-types.txt"), "r", encoding="utf8"
).read()
assert str(root) + "\n" == expectedText
def test_build_first_segment_size(all_types):
root = all_types.TestAllTypes.new_message(1)
init_all_types(root)
expectedText = open(
os.path.join(this_dir, "all-types.txt"), "r", encoding="utf8"
).read()
assert str(root) + "\n" == expectedText
root = all_types.TestAllTypes.new_message(1024 * 1024)
init_all_types(root)
expectedText = open(
os.path.join(this_dir, "all-types.txt"), "r", encoding="utf8"
).read()
assert str(root) + "\n" == expectedText
def test_binary_read(all_types):
f = open(os.path.join(this_dir, "all-types.binary"), "r", encoding="utf8")
root = all_types.TestAllTypes.read(f)
check_all_types(root)
expectedText = open(
os.path.join(this_dir, "all-types.txt"), "r", encoding="utf8"
).read()
assert str(root) + "\n" == expectedText
# Test set_root().
builder = capnp._MallocMessageBuilder()
builder.set_root(root)
check_all_types(builder.get_root(all_types.TestAllTypes))
builder2 = capnp._MallocMessageBuilder()
builder2.set_root(builder.get_root(all_types.TestAllTypes))
check_all_types(builder2.get_root(all_types.TestAllTypes))
def test_packed_read(all_types):
f = open(os.path.join(this_dir, "all-types.packed"), "r", encoding="utf8")
root = all_types.TestAllTypes.read_packed(f)
check_all_types(root)
expectedText = open(
os.path.join(this_dir, "all-types.txt"), "r", encoding="utf8"
).read()
assert str(root) + "\n" == expectedText
def test_binary_write(all_types):
root = all_types.TestAllTypes.new_message()
init_all_types(root)
root.write(open("example", "w"))
check_all_types(all_types.TestAllTypes.read(open("example", "r")))
def test_packed_write(all_types):
root = all_types.TestAllTypes.new_message()
init_all_types(root)
root.write_packed(open("example", "w"))
check_all_types(all_types.TestAllTypes.read_packed(open("example", "r")))