IQ.Lvbs/pycapnp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fixing remaining flake8 lint warnings

Browse Source
This commit is contained in:
Jacob Alexander
2019-12-11 22:44:44 -08:00
parent f9647aa9a7
commit fc75e4083d
21 changed files with 128 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
test/test_capability.py
View File

@@ -4,6 +4,7 @@ import time
import capnp
import test_capability_capnp as capability
class Server(capability.TestInterface.Server):
def __init__(self, val=1):
self.val = val
@@ -20,6 +21,7 @@ class Server(capability.TestInterface.Server):
def bam(self, i, **kwargs):
return str(i) + '_test', i
class PipelineServer(capability.TestPipeline.Server):
def getCap(self, n, inCap, _context, **kwargs):
def _then(response):
@@ -29,6 +31,7 @@ class PipelineServer(capability.TestPipeline.Server):
return inCap.foo(i=n).then(_then)
def test_client():
client = capability.TestInterface._new_client(Server())
@@ -61,6 +64,7 @@ def test_client():
with pytest.raises(AttributeError):
req.baz = 1
def test_simple_client():
client = capability.TestInterface._new_client(Server())
@@ -69,7 +73,6 @@ def test_simple_client():
assert response.x == '26'
remote = client.foo(i=5)
response = remote.wait()
@@ -121,6 +124,7 @@ def test_simple_client():
with pytest.raises(Exception):
remote = client.foo(baz=5)
def test_pipeline():
client = capability.TestPipeline._new_client(PipelineServer())
foo_client = capability.TestInterface._new_client(Server())
@@ -136,6 +140,7 @@ def test_pipeline():
response = remote.wait()
assert response.s == '26_foo'
class BadServer(capability.TestInterface.Server):
def __init__(self, val=1):
self.val = val
@@ -144,7 +149,8 @@ class BadServer(capability.TestInterface.Server):
extra = 0
if j:
extra = 1
return str(i * 5 + extra + self.val), 10 # returning too many args
return str(i * 5 + extra + self.val), 10 # returning too many args
def test_exception_client():
client = capability.TestInterface._new_client(BadServer())
@@ -153,6 +159,7 @@ def test_exception_client():
with pytest.raises(capnp.KjException):
remote.wait()
class BadPipelineServer(capability.TestPipeline.Server):
def getCap(self, n, inCap, _context, **kwargs):
def _then(response):
@@ -165,6 +172,7 @@ class BadPipelineServer(capability.TestPipeline.Server):
return inCap.foo(i=n).then(_then, _error)
def test_exception_chain():
client = capability.TestPipeline._new_client(BadPipelineServer())
foo_client = capability.TestInterface._new_client(BadServer())
@@ -176,6 +184,7 @@ def test_exception_chain():
except Exception as e:
assert 'test was a success' in str(e)
def test_pipeline_exception():
client = capability.TestPipeline._new_client(BadPipelineServer())
foo_client = capability.TestInterface._new_client(BadServer())
@@ -191,6 +200,7 @@ def test_pipeline_exception():
with pytest.raises(Exception):
remote.wait()
def test_casting():
client = capability.TestExtends._new_client(Server())
client2 = client.upcast(capability.TestInterface)
@@ -199,6 +209,7 @@ def test_casting():
with pytest.raises(Exception):
client.upcast(capability.TestPipeline)
class TailCallOrder(capability.TestCallOrder.Server):
def __init__(self):
self.count = -1
@@ -207,6 +218,7 @@ class TailCallOrder(capability.TestCallOrder.Server):
self.count += 1
return self.count
class TailCaller(capability.TestTailCaller.Server):
def __init__(self):
self.count = 0
@@ -217,6 +229,7 @@ class TailCaller(capability.TestTailCaller.Server):
tail = callee.foo_request(i=i, t='from TailCaller')
return _context.tail_call(tail)
class TailCallee(capability.TestTailCallee.Server):
def __init__(self):
self.count = 0
@@ -229,6 +242,7 @@ class TailCallee(capability.TestTailCallee.Server):
results.t = t
results.c = TailCallOrder()
def test_tail_call():
callee_server = TailCallee()
caller_server = TailCaller()

4
test/test_large_read.py
View File

@@ -31,6 +31,7 @@ def test_large_read(test_capnp):
del f
assert array.rows[0].values[9000] == 9000
def test_large_read_multiple(test_capnp):
f = tempfile.TemporaryFile()
msg1 = test_capnp.Msg.new_message()
@@ -43,6 +44,7 @@ def test_large_read_multiple(test_capnp):
for m in test_capnp.Msg.read_multiple(f):
pass
def get_two_adjacent_messages(test_capnp):
msg1 = test_capnp.Msg.new_message()
msg1.data = [0x41] * 8192
@@ -52,6 +54,7 @@ def get_two_adjacent_messages(test_capnp):
return m1 + m2
def test_large_read_multiple_bytes(test_capnp):
data = get_two_adjacent_messages(test_capnp)
for m in test_capnp.Msg.read_multiple_bytes(data):
@@ -67,6 +70,7 @@ def test_large_read_multiple_bytes(test_capnp):
for m in test_capnp.Msg.read_multiple_bytes(data):
pass
@pytest.mark.skipif(platform.python_implementation() == 'PyPy', reason="PyPy memoryview support is limited")
def test_large_read_mutltiple_bytes_memoryview(test_capnp):
data = get_two_adjacent_messages(test_capnp)

25
test/test_load.py
View File

@@ -5,28 +5,35 @@ import sys
this_dir = os.path.dirname(__file__)
@pytest.fixture
def addressbook():
return capnp.load(os.path.join(this_dir, 'addressbook.capnp'))
@pytest.fixture
def foo():
return capnp.load(os.path.join(this_dir, 'foo.capnp'))
@pytest.fixture
def bar():
return capnp.load(os.path.join(this_dir, 'bar.capnp'))
def test_basic_load():
capnp.load(os.path.join(this_dir, 'addressbook.capnp'))
def test_constants(addressbook):
assert addressbook.qux == 123
def test_classes(addressbook):
assert addressbook.AddressBook
assert addressbook.Person
def test_import(foo, bar):
m = capnp._MallocMessageBuilder()
foo = m.init_root(foo.Foo)
@@ -38,6 +45,7 @@ def test_import(foo, bar):
assert bar.foo.name == 'foo'
def test_failed_import():
s = capnp.SchemaParser()
s2 = capnp.SchemaParser()
@@ -55,17 +63,21 @@ def test_failed_import():
with pytest.raises(Exception):
bar.foo = foo
def test_defualt_import_hook():
# Make sure any previous imports of addressbook_capnp are gone
capnp.cleanup_global_schema_parser()
import addressbook_capnp # noqa: F401
import addressbook_capnp # noqa: F401
def test_dash_import():
import addressbook_with_dashes_capnp # noqa: F401
import addressbook_with_dashes_capnp # noqa: F401
def test_spaces_import():
import addressbook_with_spaces_capnp # noqa: F401
import addressbook_with_spaces_capnp # noqa: F401
def test_add_import_hook():
capnp.add_import_hook([this_dir])
@@ -76,6 +88,7 @@ def test_add_import_hook():
import addressbook_capnp
addressbook_capnp.AddressBook.new_message()
def test_multiple_add_import_hook():
capnp.add_import_hook()
capnp.add_import_hook()
@@ -87,12 +100,14 @@ def test_multiple_add_import_hook():
import addressbook_capnp
addressbook_capnp.AddressBook.new_message()
def test_remove_import_hook():
capnp.add_import_hook([this_dir])
capnp.remove_import_hook()
if 'addressbook_capnp' in sys.modules:
del sys.modules['addressbook_capnp'] # hack to deal with it being imported already
# hack to deal with it being imported already
del sys.modules['addressbook_capnp']
with pytest.raises(ImportError):
import addressbook_capnp # noqa: F401
import addressbook_capnp # noqa: F401

22
test/test_regression.py
View File

@@ -18,6 +18,7 @@ else:
def addressbook():
return capnp.load(os.path.join(this_dir, 'addressbook.capnp'))
def test_addressbook_message_classes(addressbook):
def writeAddressBook(fd):
message = capnp._MallocMessageBuilder()
@@ -46,7 +47,6 @@ def test_addressbook_message_classes(addressbook):
capnp._write_packed_message_to_fd(fd, message)
def printAddressBook(fd):
message = capnp._PackedFdMessageReader(f)
addressBook = message.get_root(addressbook.AddressBook)
@@ -79,6 +79,7 @@ def test_addressbook_message_classes(addressbook):
f = open('example', 'r')
printAddressBook(f.fileno())
def test_addressbook(addressbook):
def writeAddressBook(file):
addresses = addressbook.AddressBook.new_message()
@@ -106,7 +107,6 @@ def test_addressbook(addressbook):
addresses.write(file)
def printAddressBook(file):
addresses = addressbook.AddressBook.read(file)
@@ -132,13 +132,13 @@ def test_addressbook(addressbook):
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()
@@ -168,7 +168,6 @@ def test_addressbook_resizable(addressbook):
addresses.write(file)
def printAddressBook(file):
addresses = addressbook.AddressBook.read(file)
@@ -194,13 +193,13 @@ def test_addressbook_resizable(addressbook):
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()
@@ -233,7 +232,6 @@ def test_addressbook_explicit_fields(addressbook):
addresses.write(file)
def printAddressBook(file):
addresses = addressbook.AddressBook.read(file)
address_fields = addressbook.AddressBook.schema.fields
@@ -264,13 +262,13 @@ def test_addressbook_explicit_fields(addressbook):
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'))
@@ -280,6 +278,7 @@ def all_types():
# - Build an identical message using Python code and compare it to the golden.
#
def init_all_types(builder):
builder.voidField = None
builder.boolField = True
@@ -358,10 +357,12 @@ def init_all_types(builder):
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):
@@ -370,6 +371,7 @@ def check_list(reader, expected):
else:
assert reader[i] == v
def check_all_types(reader):
assert reader.voidField is None
assert reader.boolField
@@ -483,12 +485,14 @@ def check_all_types(reader):
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)
@@ -500,6 +504,7 @@ def test_build_first_segment_size(all_types):
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)
@@ -517,6 +522,7 @@ def test_binary_read(all_types):
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)
@@ -525,6 +531,7 @@ def test_packed_read(all_types):
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)
@@ -532,6 +539,7 @@ def test_binary_write(all_types):
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)

4
test/test_response.py
View File

@@ -1,5 +1,6 @@
import test_response_capnp
class FooServer(test_response_capnp.Foo.Server):
def __init__(self, val=1):
self.val = val
@@ -7,6 +8,7 @@ class FooServer(test_response_capnp.Foo.Server):
def foo(self, **kwargs):
return 1
class BazServer(test_response_capnp.Baz.Server):
def __init__(self, val=1):
self.val = val
@@ -14,6 +16,7 @@ class BazServer(test_response_capnp.Baz.Server):
def grault(self, **kwargs):
return {"foo": FooServer()}
def test_response_reference():
baz = test_response_capnp.Baz._new_client(BazServer())
@@ -23,6 +26,7 @@ def test_response_reference():
# This used to cause an exception about invalid pointers because the response got garbage collected
assert foo.foo().wait().val == 1
def test_response_reference2():
baz = test_response_capnp.Baz._new_client(BazServer())

5
test/test_rpc_calculator.py
View File

@@ -11,8 +11,8 @@ import capnp
examples_dir = os.path.join(os.path.dirname(__file__), '..', 'examples')
sys.path.append(examples_dir)
import calculator_client # noqa: E402
import calculator_server # noqa: E402
import calculator_client # noqa: E402
import calculator_server # noqa: E402
def test_calculator():
@@ -78,6 +78,7 @@ def test_calculator_unix():
address = 'unix:' + path
run_subprocesses(address)
def test_calculator_gc():
def new_evaluate_impl(old_evaluate_impl):
def call(*args, **kwargs):

30
test/test_serialization.py
View File

@@ -10,10 +10,12 @@ import sys
this_dir = os.path.dirname(__file__)
@pytest.fixture
def all_types():
return capnp.load(os.path.join(this_dir, 'all_types.capnp'))
def test_roundtrip_file(all_types):
f = tempfile.TemporaryFile()
msg = all_types.TestAllTypes.new_message()
@@ -24,6 +26,7 @@ def test_roundtrip_file(all_types):
msg = all_types.TestAllTypes.read(f)
test_regression.check_all_types(msg)
def test_roundtrip_file_packed(all_types):
f = tempfile.TemporaryFile()
msg = all_types.TestAllTypes.new_message()
@@ -34,6 +37,7 @@ def test_roundtrip_file_packed(all_types):
msg = all_types.TestAllTypes.read_packed(f)
test_regression.check_all_types(msg)
def test_roundtrip_bytes(all_types):
msg = all_types.TestAllTypes.new_message()
test_regression.init_all_types(msg)
@@ -42,6 +46,7 @@ def test_roundtrip_bytes(all_types):
msg = all_types.TestAllTypes.from_bytes(message_bytes)
test_regression.check_all_types(msg)
@pytest.mark.skipif(
platform.python_implementation() == 'PyPy',
reason="TODO: Investigate why this works on CPython but fails on PyPy."
@@ -53,6 +58,7 @@ def test_roundtrip_segments(all_types):
msg = all_types.TestAllTypes.from_segments(segments)
test_regression.check_all_types(msg)
@pytest.mark.skipif(sys.version_info[0] < 3, reason="mmap doesn't implement the buffer interface under python 2.")
def test_roundtrip_bytes_mmap(all_types):
msg = all_types.TestAllTypes.new_message()
@@ -68,6 +74,7 @@ def test_roundtrip_bytes_mmap(all_types):
msg = all_types.TestAllTypes.from_bytes(memory)
test_regression.check_all_types(msg)
@pytest.mark.skipif(sys.version_info[0] < 3, reason="memoryview is a builtin on Python 3")
def test_roundtrip_bytes_buffer(all_types):
msg = all_types.TestAllTypes.new_message()
@@ -78,10 +85,12 @@ def test_roundtrip_bytes_buffer(all_types):
msg = all_types.TestAllTypes.from_bytes(v)
test_regression.check_all_types(msg)
def test_roundtrip_bytes_fail(all_types):
with pytest.raises(TypeError):
all_types.TestAllTypes.from_bytes(42)
@pytest.mark.skipif(
platform.python_implementation() == 'PyPy',
reason="This works in PyPy 4.0.1 but travisci's version of PyPy has some bug that fails this test."
@@ -94,6 +103,7 @@ def test_roundtrip_bytes_packed(all_types):
msg = all_types.TestAllTypes.from_bytes_packed(message_bytes)
test_regression.check_all_types(msg)
def test_roundtrip_file_multiple(all_types):
f = tempfile.TemporaryFile()
msg = all_types.TestAllTypes.new_message()
@@ -109,6 +119,7 @@ def test_roundtrip_file_multiple(all_types):
i += 1
assert i == 3
def test_roundtrip_bytes_multiple(all_types):
msg = all_types.TestAllTypes.new_message()
test_regression.init_all_types(msg)
@@ -123,6 +134,7 @@ def test_roundtrip_bytes_multiple(all_types):
i += 1
assert i == 3
def test_roundtrip_file_multiple_packed(all_types):
f = tempfile.TemporaryFile()
msg = all_types.TestAllTypes.new_message()
@@ -138,6 +150,7 @@ def test_roundtrip_file_multiple_packed(all_types):
i += 1
assert i == 3
def test_roundtrip_bytes_multiple_packed(all_types):
msg = all_types.TestAllTypes.new_message()
test_regression.init_all_types(msg)
@@ -152,6 +165,7 @@ def test_roundtrip_bytes_multiple_packed(all_types):
i += 1
assert i == 3
@pytest.mark.skipif(
platform.python_implementation() == 'PyPy',
reason="This works on my local PyPy v2.5.0, but is for some reason broken on TravisCI. Skip for now."
@@ -164,6 +178,7 @@ def test_roundtrip_dict(all_types):
msg = all_types.TestAllTypes.from_dict(d)
test_regression.check_all_types(msg)
def test_file_and_bytes(all_types):
f = tempfile.TemporaryFile()
msg = all_types.TestAllTypes.new_message()
@@ -174,6 +189,7 @@ def test_file_and_bytes(all_types):
assert f.read() == msg.to_bytes()
def test_file_and_bytes_packed(all_types):
f = tempfile.TemporaryFile()
msg = all_types.TestAllTypes.new_message()
@@ -184,6 +200,7 @@ def test_file_and_bytes_packed(all_types):
assert f.read() == msg.to_bytes_packed()
def test_pickle(all_types):
msg = all_types.TestAllTypes.new_message()
test_regression.init_all_types(msg)
@@ -192,6 +209,7 @@ def test_pickle(all_types):
test_regression.check_all_types(msg2)
def test_from_bytes_traversal_limit(all_types):
size = 1024
bld = all_types.TestAllTypes.new_message()
@@ -203,8 +221,10 @@ def test_from_bytes_traversal_limit(all_types):
for i in range(0, size):
msg.structList[i].uInt8Field == 0
msg = all_types.TestAllTypes.from_bytes(data,
traversal_limit_in_words=2**62)
msg = all_types.TestAllTypes.from_bytes(
data,
traversal_limit_in_words=2**62
)
for i in range(0, size):
assert msg.structList[i].uInt8Field == 0
@@ -220,7 +240,9 @@ def test_from_bytes_packed_traversal_limit(all_types):
for i in range(0, size):
msg.structList[i].uInt8Field == 0
msg = all_types.TestAllTypes.from_bytes_packed(data,
traversal_limit_in_words=2**62)
msg = all_types.TestAllTypes.from_bytes_packed(
data,
traversal_limit_in_words=2**62
)
for i in range(0, size):
assert msg.structList[i].uInt8Field == 0

2
test/test_struct.py
View File

@@ -191,6 +191,7 @@ def test_set_dict_union(addressbook):
assert person.employment.employer.name == 'foo'
def isstr(s):
return isinstance(s, str)
@@ -238,6 +239,7 @@ def test_to_dict_ordered(addressbook):
with pytest.raises(Exception):
person.to_dict(ordered=True)
def test_nested_list(addressbook):
struct = addressbook.NestedList.new_message()
struct.init('list', 2)

2
test/test_threads.py
View File

@@ -11,6 +11,7 @@ import pytest
import capnp
import test_capability_capnp
@pytest.mark.skipif(
platform.python_implementation() == 'PyPy',
reason="pycapnp's GIL handling isn't working properly at the moment for PyPy"
@@ -25,6 +26,7 @@ def test_making_event_loop():
capnp.remove_event_loop()
capnp.create_event_loop()
@pytest.mark.skipif(
platform.python_implementation() == 'PyPy',
reason="pycapnp's GIL handling isn't working properly at the moment for PyPy"