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

Applying black formatting

Browse Source 
- Fixing flake8 configuration to agree with black
- Adding black validation check to github actions
This commit is contained in:
Jacob Alexander
2021-10-01 11:00:22 -07:00
parent 5dade41aeb
commit 6e7fffd7de
51 changed files with 2536 additions and 1837 deletions
Download Patch File Download Diff File Expand all files Collapse all files

79
test/test_capability.py
View File

@@ -16,17 +16,17 @@ class Server(capability.TestInterface.Server):
return str(i * 5 + extra + self.val)
def buz(self, i, **kwargs):
return i.host + '_test'
return i.host + "_test"
def bam(self, i, **kwargs):
return str(i) + '_test', i
return str(i) + "_test", i
class PipelineServer(capability.TestPipeline.Server):
def getCap(self, n, inCap, _context, **kwargs):
def _then(response):
_results = _context.results
_results.s = response.x + '_foo'
_results.s = response.x + "_foo"
_results.outBox.cap = Server(100)
return inCap.foo(i=n).then(_then)
@@ -35,13 +35,13 @@ class PipelineServer(capability.TestPipeline.Server):
def test_client():
client = capability.TestInterface._new_client(Server())
req = client._request('foo')
req = client._request("foo")
req.i = 5
remote = req.send()
response = remote.wait()
assert response.x == '26'
assert response.x == "26"
req = client.foo_request()
req.i = 5
@@ -49,7 +49,7 @@ def test_client():
remote = req.send()
response = remote.wait()
assert response.x == '26'
assert response.x == "26"
with pytest.raises(AttributeError):
client.foo2_request()
@@ -57,7 +57,7 @@ def test_client():
req = client.foo_request()
with pytest.raises(Exception):
req.i = 'foo'
req.i = "foo"
req = client.foo_request()
@@ -68,45 +68,45 @@ def test_client():
def test_simple_client():
client = capability.TestInterface._new_client(Server())
remote = client._send('foo', i=5)
remote = client._send("foo", i=5)
response = remote.wait()
assert response.x == '26'
assert response.x == "26"
remote = client.foo(i=5)
response = remote.wait()
assert response.x == '26'
assert response.x == "26"
remote = client.foo(i=5, j=True)
response = remote.wait()
assert response.x == '27'
assert response.x == "27"
remote = client.foo(5)
response = remote.wait()
assert response.x == '26'
assert response.x == "26"
remote = client.foo(5, True)
response = remote.wait()
assert response.x == '27'
assert response.x == "27"
remote = client.foo(5, j=True)
response = remote.wait()
assert response.x == '27'
assert response.x == "27"
remote = client.buz(capability.TestSturdyRefHostId.new_message(host='localhost'))
remote = client.buz(capability.TestSturdyRefHostId.new_message(host="localhost"))
response = remote.wait()
assert response.x == 'localhost_test'
assert response.x == "localhost_test"
remote = client.bam(i=5)
response = remote.wait()
assert response.x == '5_test'
assert response.x == "5_test"
assert response.i == 5
with pytest.raises(Exception):
@@ -116,7 +116,7 @@ def test_simple_client():
remote = client.foo(5, True, 100)
with pytest.raises(Exception):
remote = client.foo(i='foo')
remote = client.foo(i="foo")
with pytest.raises(AttributeError):
remote = client.foo2(i=5)
@@ -135,10 +135,10 @@ def test_pipeline():
pipelinePromise = outCap.foo(i=10)
response = pipelinePromise.wait()
assert response.x == '150'
assert response.x == "150"
response = remote.wait()
assert response.s == '26_foo'
assert response.s == "26_foo"
class BadServer(capability.TestInterface.Server):
@@ -155,7 +155,7 @@ class BadServer(capability.TestInterface.Server):
def test_exception_client():
client = capability.TestInterface._new_client(BadServer())
remote = client._send('foo', i=5)
remote = client._send("foo", i=5)
with pytest.raises(capnp.KjException):
remote.wait()
@@ -164,11 +164,11 @@ class BadPipelineServer(capability.TestPipeline.Server):
def getCap(self, n, inCap, _context, **kwargs):
def _then(response):
_results = _context.results
_results.s = response.x + '_foo'
_results.s = response.x + "_foo"
_results.outBox.cap = Server(100)
def _error(error):
raise Exception('test was a success')
raise Exception("test was a success")
return inCap.foo(i=n).then(_then, _error)
@@ -182,7 +182,7 @@ def test_exception_chain():
try:
remote.wait()
except Exception as e:
assert 'test was a success' in str(e)
assert "test was a success" in str(e)
def test_pipeline_exception():
@@ -226,7 +226,7 @@ class TailCaller(capability.TestTailCaller.Server):
def foo(self, i, callee, _context, **kwargs):
self.count += 1
tail = callee.foo_request(i=i, t='from TailCaller')
tail = callee.foo_request(i=i, t="from TailCaller")
return _context.tail_call(tail)
@@ -275,7 +275,7 @@ def test_tail_call():
def test_cancel():
client = capability.TestInterface._new_client(Server())
req = client._request('foo')
req = client._request("foo")
req.i = 5
remote = req.send()
@@ -292,17 +292,17 @@ def test_timer():
def set_timer_var():
global test_timer_var
test_timer_var = True
capnp.getTimer().after_delay(1).then(set_timer_var).wait()
assert test_timer_var is True
test_timer_var = False
promise = capnp.Promise(0).then(
lambda x: time.sleep(.1)
).then(
lambda x: time.sleep(.1)
).then(
lambda x: set_timer_var()
promise = (
capnp.Promise(0)
.then(lambda x: time.sleep(0.1))
.then(lambda x: time.sleep(0.1))
.then(lambda x: set_timer_var())
)
canceller = capnp.getTimer().after_delay(1).then(lambda: promise.cancel())
@@ -317,7 +317,7 @@ def test_timer():
def test_double_send():
client = capability.TestInterface._new_client(Server())
req = client._request('foo')
req = client._request("foo")
req.i = 5
req.send()
@@ -362,19 +362,20 @@ def test_inheritance():
remote = client.foo(i=5)
response = remote.wait()
assert response.x == '26'
assert response.x == "26"
class PassedCapTest(capability.TestPassedCap.Server):
def foo(self, cap, _context, **kwargs):
def set_result(res):
_context.results.x = res.x
return cap.foo(5).then(set_result)
def test_null_cap():
client = capability.TestPassedCap._new_client(PassedCapTest())
assert client.foo(Server()).wait().x == '26'
assert client.foo(Server()).wait().x == "26"
with pytest.raises(capnp.KjException):
client.foo().wait()
@@ -387,14 +388,14 @@ class StructArgTest(capability.TestStructArg.Server):
def test_struct_args():
client = capability.TestStructArg._new_client(StructArgTest())
assert client.bar(a='test', b=1).wait().c == 'test1'
assert client.bar(a="test", b=1).wait().c == "test1"
with pytest.raises(capnp.KjException):
assert client.bar('test', 1).wait().c == 'test1'
assert client.bar("test", 1).wait().c == "test1"
class GenericTest(capability.TestGeneric.Server):
def foo(self, a, **kwargs):
return a.as_text() + 'test'
return a.as_text() + "test"
def test_generic():
@@ -402,4 +403,4 @@ def test_generic():
obj = capnp._MallocMessageBuilder().get_root_as_any()
obj.set_as_text("anypointer_")
assert client.foo(obj).wait().b == 'anypointer_test'
assert client.foo(obj).wait().b == "anypointer_test"

85
test/test_capability_context.py
View File

@@ -7,10 +7,12 @@ this_dir = os.path.dirname(__file__)
# flake8: noqa: E501
@pytest.fixture
def capability():
capnp.cleanup_global_schema_parser()
return capnp.load(os.path.join(this_dir, 'test_capability.capnp'))
return capnp.load(os.path.join(this_dir, "test_capability.capnp"))
class Server:
def __init__(self, val=1):
@@ -23,26 +25,30 @@ class Server:
context.results.x = str(context.params.i * 5 + extra + self.val)
def buz_context(self, context):
context.results.x = context.params.i.host + '_test'
context.results.x = context.params.i.host + "_test"
class PipelineServer:
def getCap_context(self, context):
def _then(response):
context.results.s = response.x + '_foo'
context.results.outBox.cap = capability().TestInterface._new_server(Server(100))
context.results.s = response.x + "_foo"
context.results.outBox.cap = capability().TestInterface._new_server(
Server(100)
)
return context.params.inCap.foo(i=context.params.n).then(_then)
def test_client_context(capability):
client = capability.TestInterface._new_client(Server())
req = client._request('foo')
req = client._request("foo")
req.i = 5
remote = req.send()
response = remote.wait()
assert response.x == '26'
assert response.x == "26"
req = client.foo_request()
req.i = 5
@@ -50,7 +56,7 @@ def test_client_context(capability):
remote = req.send()
response = remote.wait()
assert response.x == '26'
assert response.x == "26"
with pytest.raises(AttributeError):
client.foo2_request()
@@ -58,51 +64,51 @@ def test_client_context(capability):
req = client.foo_request()
with pytest.raises(Exception):
req.i = 'foo'
req.i = "foo"
req = client.foo_request()
with pytest.raises(AttributeError):
req.baz = 1
def test_simple_client_context(capability):
client = capability.TestInterface._new_client(Server())
remote = client._send('foo', i=5)
remote = client._send("foo", i=5)
response = remote.wait()
assert response.x == '26'
assert response.x == "26"
remote = client.foo(i=5)
response = remote.wait()
assert response.x == '26'
assert response.x == "26"
remote = client.foo(i=5, j=True)
response = remote.wait()
assert response.x == '27'
assert response.x == "27"
remote = client.foo(5)
response = remote.wait()
assert response.x == '26'
assert response.x == "26"
remote = client.foo(5, True)
response = remote.wait()
assert response.x == '27'
assert response.x == "27"
remote = client.foo(5, j=True)
response = remote.wait()
assert response.x == '27'
assert response.x == "27"
remote = client.buz(capability.TestSturdyRefHostId.new_message(host='localhost'))
remote = client.buz(capability.TestSturdyRefHostId.new_message(host="localhost"))
response = remote.wait()
assert response.x == 'localhost_test'
assert response.x == "localhost_test"
with pytest.raises(Exception):
remote = client.foo(5, 10)
@@ -111,7 +117,7 @@ def test_simple_client_context(capability):
remote = client.foo(5, True, 100)
with pytest.raises(Exception):
remote = client.foo(i='foo')
remote = client.foo(i="foo")
with pytest.raises(AttributeError):
remote = client.foo2(i=5)
@@ -119,15 +125,16 @@ def test_simple_client_context(capability):
with pytest.raises(Exception):
remote = client.foo(baz=5)
@pytest.mark.xfail
def test_pipeline_context(capability):
'''
"""
E capnp.lib.capnp.KjException: capnp/lib/capnp.pyx:61: failed: <class 'Failed'>:Fixture "capability" called directly. Fixtures are not meant to be called directly,
E but are created automatically when test functions request them as parameters.
E See https://docs.pytest.org/en/latest/fixture.html for more information about fixtures, and
E https://docs.pytest.org/en/latest/deprecations.html#calling-fixtures-directly about how to update your code.
E stack: 7f87c1ac6e40 7f87c17c3250 7f87c17be260 7f87c17c49f0 7f87c17c0f50 7f87c17c5540 7f87c17d7bf0 7f87c1acb768 7f87c1aaf185 7f87c1aaf2dc 7f87c1a6da1d 7f87c3895459 7f87c3895713 7f87c38c72eb 7f87c3901409 7f87c38b5767 7f87c38b6e7e 7f87c38fe48d 7f87c38b5767 7f87c38b6e7e 7f87c38fe48d 7f87c38b5767 7f87c38b67d2 7f87c38c71cf 7f87c38fdb77 7f87c38b5767 7f87c38b67d2 7f87c38c71cf 7f87c3901409 7f87c38b6632 7f87c38c71cf 7f87c3901409
'''
"""
client = capability.TestPipeline._new_client(PipelineServer())
foo_client = capability.TestInterface._new_client(Server())
@@ -137,10 +144,11 @@ def test_pipeline_context(capability):
pipelinePromise = outCap.foo(i=10)
response = pipelinePromise.wait()
assert response.x == '150'
assert response.x == "150"
response = remote.wait()
assert response.s == '26_foo'
assert response.s == "26_foo"
class BadServer:
def __init__(self, val=1):
@@ -148,26 +156,31 @@ class BadServer:
def foo_context(self, context):
context.results.x = str(context.params.i * 5 + self.val)
context.results.x2 = 5 # raises exception
context.results.x2 = 5 # raises exception
def test_exception_client_context(capability):
client = capability.TestInterface._new_client(BadServer())
remote = client._send('foo', i=5)
remote = client._send("foo", i=5)
with pytest.raises(capnp.KjException):
remote.wait()
class BadPipelineServer:
def getCap_context(self, context):
def _then(response):
context.results.s = response.x + '_foo'
context.results.outBox.cap = capability().TestInterface._new_server(Server(100))
context.results.s = response.x + "_foo"
context.results.outBox.cap = capability().TestInterface._new_server(
Server(100)
)
def _error(error):
raise Exception('test was a success')
raise Exception("test was a success")
return context.params.inCap.foo(i=context.params.n).then(_then, _error)
def test_exception_chain_context(capability):
client = capability.TestPipeline._new_client(BadPipelineServer())
foo_client = capability.TestInterface._new_client(BadServer())
@@ -177,7 +190,8 @@ def test_exception_chain_context(capability):
try:
remote.wait()
except Exception as e:
assert 'test was a success' in str(e)
assert "test was a success" in str(e)
def test_pipeline_exception_context(capability):
client = capability.TestPipeline._new_client(BadPipelineServer())
@@ -194,6 +208,7 @@ def test_pipeline_exception_context(capability):
with pytest.raises(Exception):
remote.wait()
def test_casting_context(capability):
client = capability.TestExtends._new_client(Server())
client2 = client.upcast(capability.TestInterface)
@@ -202,6 +217,7 @@ def test_casting_context(capability):
with pytest.raises(Exception):
client.upcast(capability.TestPipeline)
class TailCallOrder:
def __init__(self):
self.count = -1
@@ -210,6 +226,7 @@ class TailCallOrder:
self.count += 1
context.results.n = self.count
class TailCaller:
def __init__(self):
self.count = 0
@@ -217,9 +234,12 @@ class TailCaller:
def foo_context(self, context):
self.count += 1
tail = context.params.callee.foo_request(i=context.params.i, t='from TailCaller')
tail = context.params.callee.foo_request(
i=context.params.i, t="from TailCaller"
)
return context.tail_call(tail)
class TailCallee:
def __init__(self):
self.count = 0
@@ -232,15 +252,16 @@ class TailCallee:
results.t = context.params.t
results.c = capability().TestCallOrder._new_server(TailCallOrder())
@pytest.mark.xfail
def test_tail_call(capability):
'''
"""
E capnp.lib.capnp.KjException: capnp/lib/capnp.pyx:75: failed: <class 'Failed'>:Fixture "capability" called directly. Fixtures are not meant to be called directly,
E but are created automatically when test functions request them as parameters.
E See https://docs.pytest.org/en/latest/fixture.html for more information about fixtures, and
E https://docs.pytest.org/en/latest/deprecations.html#calling-fixtures-directly about how to update your code.
E stack: 7f87c17c5540 7f87c17c51b0 7f87c17c5540 7f87c17d7bf0 7f87c1acb768 7f87c1aaf185 7f87c1aaf2dc 7f87c1a6da1d 7f87c3895459 7f87c3895713 7f87c38c72eb 7f87c3901409 7f87c38b5767 7f87c38b6e7e 7f87c38fe48d 7f87c38b5767 7f87c38b6e7e 7f87c38fe48d 7f87c38b5767 7f87c38b67d2 7f87c38c71cf 7f87c38fdb77 7f87c38b5767 7f87c38b67d2 7f87c38c71cf 7f87c3901409 7f87c38b6632 7f87c38c71cf 7f87c3901409 7f87c38b5767 7f87c38b6e7e 7f87c388ace7
'''
"""
callee_server = TailCallee()
caller_server = TailCaller()

75
test/test_capability_old.py
View File

@@ -7,9 +7,11 @@ this_dir = os.path.dirname(__file__)
# flake8: noqa: E501
@pytest.fixture
def capability():
return capnp.load(os.path.join(this_dir, 'test_capability.capnp'))
return capnp.load(os.path.join(this_dir, "test_capability.capnp"))
class Server:
def __init__(self, val=1):
@@ -22,27 +24,29 @@ class Server:
return str(i * 5 + extra + self.val)
def buz(self, i, **kwargs):
return i.host + '_test'
return i.host + "_test"
class PipelineServer:
def getCap(self, n, inCap, _context, **kwargs):
def _then(response):
_results = _context.results
_results.s = response.x + '_foo'
_results.s = response.x + "_foo"
_results.outBox.cap = capability().TestInterface._new_server(Server(100))
return inCap.foo(i=n).then(_then)
def test_client(capability):
client = capability.TestInterface._new_client(Server())
req = client._request('foo')
req = client._request("foo")
req.i = 5
remote = req.send()
response = remote.wait()
assert response.x == '26'
assert response.x == "26"
req = client.foo_request()
req.i = 5
@@ -50,7 +54,7 @@ def test_client(capability):
remote = req.send()
response = remote.wait()
assert response.x == '26'
assert response.x == "26"
with pytest.raises(AttributeError):
client.foo2_request()
@@ -58,51 +62,51 @@ def test_client(capability):
req = client.foo_request()
with pytest.raises(Exception):
req.i = 'foo'
req.i = "foo"
req = client.foo_request()
with pytest.raises(AttributeError):
req.baz = 1
def test_simple_client(capability):
client = capability.TestInterface._new_client(Server())
remote = client._send('foo', i=5)
remote = client._send("foo", i=5)
response = remote.wait()
assert response.x == '26'
assert response.x == "26"
remote = client.foo(i=5)
response = remote.wait()
assert response.x == '26'
assert response.x == "26"
remote = client.foo(i=5, j=True)
response = remote.wait()
assert response.x == '27'
assert response.x == "27"
remote = client.foo(5)
response = remote.wait()
assert response.x == '26'
assert response.x == "26"
remote = client.foo(5, True)
response = remote.wait()
assert response.x == '27'
assert response.x == "27"
remote = client.foo(5, j=True)
response = remote.wait()
assert response.x == '27'
assert response.x == "27"
remote = client.buz(capability.TestSturdyRefHostId.new_message(host='localhost'))
remote = client.buz(capability.TestSturdyRefHostId.new_message(host="localhost"))
response = remote.wait()
assert response.x == 'localhost_test'
assert response.x == "localhost_test"
with pytest.raises(Exception):
remote = client.foo(5, 10)
@@ -111,7 +115,7 @@ def test_simple_client(capability):
remote = client.foo(5, True, 100)
with pytest.raises(Exception):
remote = client.foo(i='foo')
remote = client.foo(i="foo")
with pytest.raises(AttributeError):
remote = client.foo2(i=5)
@@ -119,15 +123,16 @@ def test_simple_client(capability):
with pytest.raises(Exception):
remote = client.foo(baz=5)
@pytest.mark.xfail
def test_pipeline(capability):
'''
"""
E capnp.lib.capnp.KjException: capnp/lib/capnp.pyx:61: failed: <class 'Failed'>:Fixture "capability" called directly. Fixtures are not meant to be called directly,
E but are created automatically when test functions request them as parameters.
E See https://docs.pytest.org/en/latest/fixture.html for more information about fixtures, and
E https://docs.pytest.org/en/latest/deprecations.html#calling-fixtures-directly about how to update your code.
E stack: 7f680f7fce40 7f680f4f9250 7f680f4f4260 7f680f4fa9f0 7f680f4f6f50 7f680f4fb540 7f680f50dbf0 7f680f801768 7f680f7e5185 7f680f7e52dc 7f680f7a3a1d 7f68115cb459 7f68115cb713 7f68115fd2eb 7f6811637409 7f68115eb767 7f68115ece7e 7f681163448d 7f68115eb767 7f68115ece7e 7f681163448d 7f68115eb767 7f68115ec7d2 7f68115fd1cf 7f6811633b77 7f68115eb767 7f68115ec7d2 7f68115fd1cf 7f6811637409 7f68115ec632 7f68115fd1cf 7f6811637409
'''
"""
client = capability.TestPipeline._new_client(PipelineServer())
foo_client = capability.TestInterface._new_client(Server())
@@ -137,10 +142,11 @@ def test_pipeline(capability):
pipelinePromise = outCap.foo(i=10)
response = pipelinePromise.wait()
assert response.x == '150'
assert response.x == "150"
response = remote.wait()
assert response.s == '26_foo'
assert response.s == "26_foo"
class BadServer:
def __init__(self, val=1):
@@ -150,27 +156,30 @@ class BadServer:
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(capability):
client = capability.TestInterface._new_client(BadServer())
remote = client._send('foo', i=5)
remote = client._send("foo", i=5)
with pytest.raises(capnp.KjException):
remote.wait()
class BadPipelineServer:
def getCap(self, n, inCap, _context, **kwargs):
def _then(response):
_results = _context.results
_results.s = response.x + '_foo'
_results.s = response.x + "_foo"
_results.outBox.cap = capability().TestInterface._new_server(Server(100))
def _error(error):
raise Exception('test was a success')
raise Exception("test was a success")
return inCap.foo(i=n).then(_then, _error)
def test_exception_chain(capability):
client = capability.TestPipeline._new_client(BadPipelineServer())
foo_client = capability.TestInterface._new_client(BadServer())
@@ -180,7 +189,8 @@ def test_exception_chain(capability):
try:
remote.wait()
except Exception as e:
assert 'test was a success' in str(e)
assert "test was a success" in str(e)
def test_pipeline_exception(capability):
client = capability.TestPipeline._new_client(BadPipelineServer())
@@ -197,6 +207,7 @@ def test_pipeline_exception(capability):
with pytest.raises(Exception):
remote.wait()
def test_casting(capability):
client = capability.TestExtends._new_client(Server())
client2 = client.upcast(capability.TestInterface)
@@ -205,6 +216,7 @@ def test_casting(capability):
with pytest.raises(Exception):
client.upcast(capability.TestPipeline)
class TailCallOrder:
def __init__(self):
self.count = -1
@@ -213,6 +225,7 @@ class TailCallOrder:
self.count += 1
return self.count
class TailCaller:
def __init__(self):
self.count = 0
@@ -220,9 +233,10 @@ class TailCaller:
def foo(self, i, callee, _context, **kwargs):
self.count += 1
tail = callee.foo_request(i=i, t='from TailCaller')
tail = callee.foo_request(i=i, t="from TailCaller")
return _context.tail_call(tail)
class TailCallee:
def __init__(self):
self.count = 0
@@ -235,15 +249,16 @@ class TailCallee:
results.t = t
results.c = capability().TestCallOrder._new_server(TailCallOrder())
@pytest.mark.xfail
def test_tail_call(capability):
'''
"""
E capnp.lib.capnp.KjException: capnp/lib/capnp.pyx:104: failed: <class 'Failed'>:Fixture "capability" called directly. Fixtures are not meant to be called directly,
E but are created automatically when test functions request them as parameters.
E See https://docs.pytest.org/en/latest/fixture.html for more information about fixtures, and
E https://docs.pytest.org/en/latest/deprecations.html#calling-fixtures-directly about how to update your code.
E stack: 7f680f4fb540 7f680f4fb1b0 7f680f4fb540 7f680f50dbf0 7f680f801768 7f680f7e5185 7f680f7e52dc 7f680f7a3a1d 7f68115cb459 7f68115cb713 7f68115fd2eb 7f6811637409 7f68115eb767 7f68115ece7e 7f681163448d 7f68115eb767 7f68115ece7e 7f681163448d 7f68115eb767 7f68115ec7d2 7f68115fd1cf 7f6811633b77 7f68115eb767 7f68115ec7d2 7f68115fd1cf 7f6811637409 7f68115ec632 7f68115fd1cf 7f6811637409 7f68115eb767 7f68115ece7e 7f68115c0ce7
'''
"""
callee_server = TailCallee()
caller_server = TailCaller()

72
test/test_examples.py
View File

@@ -5,8 +5,8 @@ import subprocess
import sys
import time
examples_dir = os.path.join(os.path.dirname(__file__), '..', 'examples')
hostname = 'localhost'
examples_dir = os.path.join(os.path.dirname(__file__), "..", "examples")
hostname = "localhost"
processes = []
@@ -19,24 +19,28 @@ def cleanup():
p.kill()
def run_subprocesses(address, server, client, wildcard_server=False, ipv4_force=True): # noqa
def run_subprocesses(
address, server, client, wildcard_server=False, ipv4_force=True
): # noqa
server_attempt = 0
server_attempts = 2
done = False
addr, port = address.split(':')
addr, port = address.split(":")
c_address = address
s_address = address
while not done:
assert server_attempt < server_attempts, "Failed {} server attempts".format(server_attempts)
assert server_attempt < server_attempts, "Failed {} server attempts".format(
server_attempts
)
server_attempt += 1
# Force ipv4 for tests (known issues on GitHub Actions with IPv6 for some targets)
if 'unix' not in addr and ipv4_force:
if "unix" not in addr and ipv4_force:
addr = socket.gethostbyname(addr)
c_address = '{}:{}'.format(addr, port)
c_address = "{}:{}".format(addr, port)
s_address = c_address
if wildcard_server:
s_address = '*:{}'.format(port) # Use wildcard address for server
s_address = "*:{}".format(port) # Use wildcard address for server
print("Forcing ipv4 -> {} => {} {}".format(address, c_address, s_address))
# Start server
@@ -48,7 +52,7 @@ def run_subprocesses(address, server, client, wildcard_server=False, ipv4_force=
# Loop until we have a socket connection to the server (with timeout)
while True:
try:
if 'unix' in address:
if "unix" in address:
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
result = sock.connect_ex(port)
if result == 0:
@@ -114,22 +118,26 @@ def run_subprocesses(address, server, client, wildcard_server=False, ipv4_force=
def test_async_calculator_example(cleanup):
address = '{}:36432'.format(hostname)
server = 'async_calculator_server.py'
client = 'async_calculator_client.py'
address = "{}:36432".format(hostname)
server = "async_calculator_server.py"
client = "async_calculator_client.py"
run_subprocesses(address, server, client)
@pytest.mark.xfail(reason="Some versions of python don't like to share ports, don't worry if this fails")
@pytest.mark.xfail(
reason="Some versions of python don't like to share ports, don't worry if this fails"
)
def test_thread_example(cleanup):
address = '{}:36433'.format(hostname)
server = 'thread_server.py'
client = 'thread_client.py'
address = "{}:36433".format(hostname)
server = "thread_server.py"
client = "thread_client.py"
run_subprocesses(address, server, client, wildcard_server=True)
def test_addressbook_example(cleanup):
proc = subprocess.Popen([sys.executable, os.path.join(examples_dir, 'addressbook.py')])
proc = subprocess.Popen(
[sys.executable, os.path.join(examples_dir, "addressbook.py")]
)
ret = proc.wait()
assert ret == 0
@@ -139,12 +147,12 @@ def test_addressbook_example(cleanup):
reason="""
Asyncio bug with libcapnp timer, likely due to asyncio starving some event loop.
See https://github.com/capnproto/pycapnp/issues/196
"""
""",
)
def test_async_example(cleanup):
address = '{}:36434'.format(hostname)
server = 'async_server.py'
client = 'async_client.py'
address = "{}:36434".format(hostname)
server = "async_server.py"
client = "async_client.py"
run_subprocesses(address, server, client)
@@ -153,12 +161,12 @@ def test_async_example(cleanup):
reason="""
Asyncio bug with libcapnp timer, likely due to asyncio starving some event loop.
See https://github.com/capnproto/pycapnp/issues/196
"""
""",
)
def test_ssl_async_example(cleanup):
address = '{}:36435'.format(hostname)
server = 'async_ssl_server.py'
client = 'async_ssl_client.py'
address = "{}:36435".format(hostname)
server = "async_ssl_server.py"
client = "async_ssl_client.py"
run_subprocesses(address, server, client, ipv4_force=False)
@@ -167,17 +175,17 @@ def test_ssl_async_example(cleanup):
reason="""
Asyncio bug with libcapnp timer, likely due to asyncio starving some event loop.
See https://github.com/capnproto/pycapnp/issues/196
"""
""",
)
def test_ssl_reconnecting_async_example(cleanup):
address = '{}:36436'.format(hostname)
server = 'async_ssl_server.py'
client = 'async_reconnecting_ssl_client.py'
address = "{}:36436".format(hostname)
server = "async_ssl_server.py"
client = "async_reconnecting_ssl_client.py"
run_subprocesses(address, server, client, ipv4_force=False)
def test_async_ssl_calculator_example(cleanup):
address = '{}:36437'.format(hostname)
server = 'async_ssl_calculator_server.py'
client = 'async_ssl_calculator_client.py'
address = "{}:36437".format(hostname)
server = "async_ssl_calculator_server.py"
client = "async_ssl_calculator_client.py"
run_subprocesses(address, server, client, ipv4_force=False)

15
test/test_large_read.py
View File

@@ -11,7 +11,7 @@ this_dir = os.path.dirname(__file__)
@pytest.fixture
def test_capnp():
return capnp.load(os.path.join(this_dir, 'test_large_read.capnp'))
return capnp.load(os.path.join(this_dir, "test_large_read.capnp"))
def test_large_read(test_capnp):
@@ -19,8 +19,8 @@ def test_large_read(test_capnp):
array = test_capnp.MultiArray.new_message()
row = array.init('rows', 1)[0]
values = row.init('values', 10000)
row = array.init("rows", 1)[0]
values = row.init("values", 10000)
for i in range(len(values)):
values[i] = i
@@ -66,12 +66,15 @@ def test_large_read_multiple_bytes(test_capnp):
pass
with pytest.raises(capnp.KjException):
data = get_two_adjacent_messages(test_capnp) + b' '
data = get_two_adjacent_messages(test_capnp) + b" "
for m in test_capnp.Msg.read_multiple_bytes(data):
pass
@pytest.mark.skipif(platform.python_implementation() == 'PyPy', reason="PyPy memoryview support is limited")
@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)
for m in test_capnp.Msg.read_multiple_bytes(memoryview(data)):
@@ -83,6 +86,6 @@ def test_large_read_mutltiple_bytes_memoryview(test_capnp):
pass
with pytest.raises(capnp.KjException):
data = get_two_adjacent_messages(test_capnp) + b' '
data = get_two_adjacent_messages(test_capnp) + b" "
for m in test_capnp.Msg.read_multiple_bytes(memoryview(data)):
pass

24
test/test_load.py
View File

@@ -8,21 +8,21 @@ this_dir = os.path.dirname(__file__)
@pytest.fixture
def addressbook():
return capnp.load(os.path.join(this_dir, 'addressbook.capnp'))
return capnp.load(os.path.join(this_dir, "addressbook.capnp"))
@pytest.fixture
def foo():
return capnp.load(os.path.join(this_dir, 'foo.capnp'))
return capnp.load(os.path.join(this_dir, "foo.capnp"))
@pytest.fixture
def bar():
return capnp.load(os.path.join(this_dir, 'bar.capnp'))
return capnp.load(os.path.join(this_dir, "bar.capnp"))
def test_basic_load():
capnp.load(os.path.join(this_dir, 'addressbook.capnp'))
capnp.load(os.path.join(this_dir, "addressbook.capnp"))
def test_constants(addressbook):
@@ -40,25 +40,25 @@ def test_import(foo, bar):
m2 = capnp._MallocMessageBuilder()
bar = m2.init_root(bar.Bar)
foo.name = 'foo'
foo.name = "foo"
bar.foo = foo
assert bar.foo.name == 'foo'
assert bar.foo.name == "foo"
def test_failed_import():
s = capnp.SchemaParser()
s2 = capnp.SchemaParser()
foo = s.load(os.path.join(this_dir, 'foo.capnp'))
bar = s2.load(os.path.join(this_dir, 'bar.capnp'))
foo = s.load(os.path.join(this_dir, "foo.capnp"))
bar = s2.load(os.path.join(this_dir, "bar.capnp"))
m = capnp._MallocMessageBuilder()
foo = m.init_root(foo.Foo)
m2 = capnp._MallocMessageBuilder()
bar = m2.init_root(bar.Bar)
foo.name = 'foo'
foo.name = "foo"
with pytest.raises(Exception):
bar.foo = foo
@@ -86,6 +86,7 @@ def test_add_import_hook():
capnp.cleanup_global_schema_parser()
import addressbook_capnp
addressbook_capnp.AddressBook.new_message()
@@ -98,6 +99,7 @@ def test_multiple_add_import_hook():
capnp.cleanup_global_schema_parser()
import addressbook_capnp
addressbook_capnp.AddressBook.new_message()
@@ -105,9 +107,9 @@ def test_remove_import_hook():
capnp.add_import_hook([this_dir])
capnp.remove_import_hook()
if 'addressbook_capnp' in sys.modules:
if "addressbook_capnp" in sys.modules:
# hack to deal with it being imported already
del sys.modules['addressbook_capnp']
del sys.modules["addressbook_capnp"]
with pytest.raises(ImportError):
import addressbook_capnp # noqa: F401

20
test/test_object.py
View File

@@ -7,22 +7,22 @@ this_dir = os.path.dirname(__file__)
@pytest.fixture
def addressbook():
return capnp.load(os.path.join(this_dir, 'addressbook.capnp'))
return capnp.load(os.path.join(this_dir, "addressbook.capnp"))
def test_object_basic(addressbook):
obj = capnp._MallocMessageBuilder().get_root_as_any()
person = obj.as_struct(addressbook.Person)
person.name = 'test'
person.name = "test"
person.id = 1000
same_person = obj.as_struct(addressbook.Person)
assert same_person.name == 'test'
assert same_person.name == "test"
assert same_person.id == 1000
obj_r = obj.as_reader()
same_person = obj_r.as_struct(addressbook.Person)
assert same_person.name == 'test'
assert same_person.name == "test"
assert same_person.id == 1000
@@ -31,21 +31,21 @@ def test_object_list(addressbook):
listSchema = capnp._ListSchema(addressbook.Person)
people = obj.init_as_list(listSchema, 2)
person = people[0]
person.name = 'test'
person.name = "test"
person.id = 1000
person = people[1]
person.name = 'test2'
person.name = "test2"
person.id = 1001
same_person = obj.as_list(listSchema)
assert same_person[0].name == 'test'
assert same_person[0].name == "test"
assert same_person[0].id == 1000
assert same_person[1].name == 'test2'
assert same_person[1].name == "test2"
assert same_person[1].id == 1001
obj_r = obj.as_reader()
same_person = obj_r.as_list(listSchema)
assert same_person[0].name == 'test'
assert same_person[0].name == "test"
assert same_person[0].id == 1000
assert same_person[1].name == 'test2'
assert same_person[1].name == "test2"
assert same_person[1].id == 1001

305
test/test_regression.py
View File

@@ -16,33 +16,33 @@ else:
@pytest.fixture
def addressbook():
return capnp.load(os.path.join(this_dir, 'addressbook.capnp'))
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)
people = addressBook.init("people", 2)
alice = people[0]
alice.id = 123
alice.name = 'Alice'
alice.email = 'alice@example.com'
alicePhones = alice.init('phones', 1)
alice.name = "Alice"
alice.email = "alice@example.com"
alicePhones = alice.init("phones", 1)
alicePhones[0].number = "555-1212"
alicePhones[0].type = 'mobile'
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)
bob.name = "Bob"
bob.email = "bob@example.com"
bobPhones = bob.init("phones", 2)
bobPhones[0].number = "555-4567"
bobPhones[0].type = 'home'
bobPhones[0].type = "home"
bobPhones[1].number = "555-7654"
bobPhones[1].type = 'work'
bobPhones[1].type = "work"
bob.employment.unemployed = None
capnp._write_packed_message_to_fd(fd, message)
@@ -55,54 +55,54 @@ def test_addressbook_message_classes(addressbook):
alice = people[0]
assert alice.id == 123
assert alice.name == 'Alice'
assert alice.email == 'alice@example.com'
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 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'
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[0].type == "home"
assert bobPhones[1].number == "555-7654"
assert bobPhones[1].type == 'work'
assert bobPhones[1].type == "work"
assert bob.employment.unemployed is None
f = open('example', 'w')
f = open("example", "w")
writeAddressBook(f.fileno())
f = open('example', 'r')
f = open("example", "r")
printAddressBook(f.fileno())
def test_addressbook(addressbook):
def writeAddressBook(file):
addresses = addressbook.AddressBook.new_message()
people = addresses.init('people', 2)
people = addresses.init("people", 2)
alice = people[0]
alice.id = 123
alice.name = 'Alice'
alice.email = 'alice@example.com'
alicePhones = alice.init('phones', 1)
alice.name = "Alice"
alice.email = "alice@example.com"
alicePhones = alice.init("phones", 1)
alicePhones[0].number = "555-1212"
alicePhones[0].type = 'mobile'
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)
bob.name = "Bob"
bob.email = "bob@example.com"
bobPhones = bob.init("phones", 2)
bobPhones[0].number = "555-4567"
bobPhones[0].type = 'home'
bobPhones[0].type = "home"
bobPhones[1].number = "555-7654"
bobPhones[1].type = 'work'
bobPhones[1].type = "work"
bob.employment.unemployed = None
addresses.write(file)
@@ -114,54 +114,54 @@ def test_addressbook(addressbook):
alice = people[0]
assert alice.id == 123
assert alice.name == 'Alice'
assert alice.email == 'alice@example.com'
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 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'
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[0].type == "home"
assert bobPhones[1].number == "555-7654"
assert bobPhones[1].type == 'work'
assert bobPhones[1].type == "work"
assert bob.employment.unemployed is None
f = open('example', 'w')
f = open("example", "w")
writeAddressBook(f)
f = open('example', 'r')
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')
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)
alice.name = "Alice"
alice.email = "alice@example.com"
alicePhones = alice.init("phones", 1)
alicePhones[0].number = "555-1212"
alicePhones[0].type = 'mobile'
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)
bob.name = "Bob"
bob.email = "bob@example.com"
bobPhones = bob.init("phones", 2)
bobPhones[0].number = "555-4567"
bobPhones[0].type = 'home'
bobPhones[0].type = "home"
bobPhones[1].number = "555-7654"
bobPhones[1].type = 'work'
bobPhones[1].type = "work"
bob.employment.unemployed = None
people.finish()
@@ -175,28 +175,28 @@ def test_addressbook_resizable(addressbook):
alice = people[0]
assert alice.id == 123
assert alice.name == 'Alice'
assert alice.email == 'alice@example.com'
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 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'
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[0].type == "home"
assert bobPhones[1].number == "555-7654"
assert bobPhones[1].type == 'work'
assert bobPhones[1].type == "work"
assert bob.employment.unemployed is None
f = open('example', 'w')
f = open("example", "w")
writeAddressBook(f)
f = open('example', 'r')
f = open("example", "r")
printAddressBook(f)
@@ -206,29 +206,33 @@ def test_addressbook_explicit_fields(addressbook):
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)
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")
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)
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)
@@ -238,40 +242,45 @@ def test_addressbook_explicit_fields(addressbook):
person_fields = addressbook.Person.schema.fields
phone_fields = addressbook.Person.PhoneNumber.schema.fields
people = addresses._get_by_field(address_fields['people'])
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"
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
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')
f = open("example", "w")
writeAddressBook(f)
f = open('example', 'r')
f = open("example", "r")
printAddressBook(f)
@pytest.fixture
def all_types():
return capnp.load(os.path.join(this_dir, 'all_types.capnp'))
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.
@@ -317,19 +326,24 @@ def init_all_types(builder):
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.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 = subBuilder.init("structList", 3)
listBuilder[0].textField = "x structlist 1"
listBuilder[1].textField = "x structlist 2"
listBuilder[2].textField = "x structlist 3"
@@ -351,7 +365,7 @@ def init_all_types(builder):
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 = builder.init("structList", 3)
listBuilder[0].textField = "structlist 1"
listBuilder[1].textField = "structlist 2"
listBuilder[2].textField = "structlist 3"
@@ -419,23 +433,30 @@ def check_all_types(reader):
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 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
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.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.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"])
@@ -489,29 +510,37 @@ def check_all_types(reader):
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
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
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
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')
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
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()
@@ -524,25 +553,27 @@ def test_binary_read(all_types):
def test_packed_read(all_types):
f = open(os.path.join(this_dir, 'all-types.packed'), 'r', encoding='utf8')
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
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'))
root.write(open("example", "w"))
check_all_types(all_types.TestAllTypes.read(open('example', 'r')))
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'))
root.write_packed(open("example", "w"))
check_all_types(all_types.TestAllTypes.read_packed(open('example', 'r')))
check_all_types(all_types.TestAllTypes.read_packed(open("example", "r")))

7
test/test_rpc.py
View File

@@ -1,6 +1,6 @@
'''
"""
rpc test
'''
"""
import pytest
import capnp
@@ -10,7 +10,6 @@ import test_capability_capnp
class Server(test_capability_capnp.TestInterface.Server):
def __init__(self, val=100):
self.val = val
@@ -45,4 +44,4 @@ def test_simple_rpc_bootstrap():
remote = cap.foo(i=5)
response = remote.wait()
assert response.x == '125'
assert response.x == "125"

27
test/test_rpc_calculator.py
View File

@@ -6,7 +6,7 @@ import sys # add examples dir to sys.path
import capnp
examples_dir = os.path.join(os.path.dirname(__file__), '..', 'examples')
examples_dir = os.path.join(os.path.dirname(__file__), "..", "examples")
sys.path.append(examples_dir)
import calculator_client # noqa: E402
@@ -32,23 +32,31 @@ def test_calculator():
calculator_client.main(read)
@pytest.mark.xfail(reason="Some versions of python don't like to share ports, don't worry if this fails")
@pytest.mark.xfail(
reason="Some versions of python don't like to share ports, don't worry if this fails"
)
def test_calculator_tcp(cleanup):
address = 'localhost:36431'
test_examples.run_subprocesses(address, 'calculator_server.py', 'calculator_client.py', wildcard_server=True)
address = "localhost:36431"
test_examples.run_subprocesses(
address, "calculator_server.py", "calculator_client.py", wildcard_server=True
)
@pytest.mark.xfail(reason="Some versions of python don't like to share ports, don't worry if this fails")
@pytest.mark.skipif(os.name == 'nt', reason="socket.AF_UNIX not supported on Windows")
@pytest.mark.xfail(
reason="Some versions of python don't like to share ports, don't worry if this fails"
)
@pytest.mark.skipif(os.name == "nt", reason="socket.AF_UNIX not supported on Windows")
def test_calculator_unix(cleanup):
path = '/tmp/pycapnp-test'
path = "/tmp/pycapnp-test"
try:
os.unlink(path)
except OSError:
pass
address = 'unix:' + path
test_examples.run_subprocesses(address, 'calculator_server.py', 'calculator_client.py')
address = "unix:" + path
test_examples.run_subprocesses(
address, "calculator_server.py", "calculator_client.py"
)
def test_calculator_gc():
@@ -56,6 +64,7 @@ def test_calculator_gc():
def call(*args, **kwargs):
gc.collect()
return old_evaluate_impl(*args, **kwargs)
return call
read, write = socket.socketpair()

20
test/test_schema.py
View File

@@ -7,20 +7,20 @@ this_dir = os.path.dirname(__file__)
@pytest.fixture
def addressbook():
return capnp.load(os.path.join(this_dir, 'addressbook.capnp'))
return capnp.load(os.path.join(this_dir, "addressbook.capnp"))
@pytest.fixture
def annotations():
return capnp.load(os.path.join(this_dir, 'annotations.capnp'))
return capnp.load(os.path.join(this_dir, "annotations.capnp"))
def test_basic_schema(addressbook):
assert addressbook.Person.schema.fieldnames[0] == 'id'
assert addressbook.Person.schema.fieldnames[0] == "id"
def test_list_schema(addressbook):
peopleField = addressbook.AddressBook.schema.fields['people']
peopleField = addressbook.AddressBook.schema.fields["people"]
personType = peopleField.schema.elementType
assert personType.node.id == addressbook.Person.schema.node.id
@@ -31,20 +31,24 @@ def test_list_schema(addressbook):
def test_annotations(annotations):
assert annotations.schema.node.annotations[0].value.text == 'TestFile'
assert annotations.schema.node.annotations[0].value.text == "TestFile"
annotation = annotations.TestAnnotationOne.schema.node.annotations[0]
assert annotation.value.text == 'Test'
assert annotation.value.text == "Test"
annotation = annotations.TestAnnotationTwo.schema.node.annotations[0]
assert annotation.value.struct.as_struct(annotations.AnnotationStruct).test == 100
annotation = annotations.TestAnnotationThree.schema.node.annotations[0]
annotation_list = annotation.value.list.as_list(capnp._ListSchema(annotations.AnnotationStruct))
annotation_list = annotation.value.list.as_list(
capnp._ListSchema(annotations.AnnotationStruct)
)
assert annotation_list[0].test == 100
assert annotation_list[1].test == 101
annotation = annotations.TestAnnotationFour.schema.node.annotations[0]
annotation_list = annotation.value.list.as_list(capnp._ListSchema(capnp.types.UInt16))
annotation_list = annotation.value.list.as_list(
capnp._ListSchema(capnp.types.UInt16)
)
assert annotation_list[0] == 200
assert annotation_list[1] == 201

31
test/test_serialization.py
View File

@@ -16,7 +16,7 @@ this_dir = os.path.dirname(__file__)
@pytest.fixture
def all_types():
return capnp.load(os.path.join(this_dir, 'all_types.capnp'))
return capnp.load(os.path.join(this_dir, "all_types.capnp"))
def test_roundtrip_file(all_types):
@@ -51,8 +51,8 @@ def test_roundtrip_bytes(all_types):
@pytest.mark.skipif(
platform.python_implementation() == 'PyPy',
reason="TODO: Investigate why this works on CPython but fails on PyPy."
platform.python_implementation() == "PyPy",
reason="TODO: Investigate why this works on CPython but fails on PyPy.",
)
def test_roundtrip_segments(all_types):
msg = all_types.TestAllTypes.new_message()
@@ -62,7 +62,10 @@ def test_roundtrip_segments(all_types):
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.")
@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()
test_regression.init_all_types(msg)
@@ -78,7 +81,9 @@ def test_roundtrip_bytes_mmap(all_types):
test_regression.check_all_types(msg)
@pytest.mark.skipif(sys.version_info[0] < 3, reason="memoryview is a builtin on Python 3")
@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()
test_regression.init_all_types(msg)
@@ -95,8 +100,8 @@ def test_roundtrip_bytes_fail(all_types):
@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."
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.",
)
def test_roundtrip_bytes_packed(all_types):
msg = all_types.TestAllTypes.new_message()
@@ -108,7 +113,9 @@ def test_roundtrip_bytes_packed(all_types):
@contextmanager
def _warnings(expected_count=2, expected_text='This message has already been written once.'):
def _warnings(
expected_count=2, expected_text="This message has already been written once."
):
with warnings.catch_warnings(record=True) as w:
yield
@@ -227,10 +234,7 @@ 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
@@ -247,8 +251,7 @@ def test_from_bytes_packed_traversal_limit(all_types):
msg.structList[i].uInt8Field == 0
msg = all_types.TestAllTypes.from_bytes_packed(
data,
traversal_limit_in_words=2**62
data, traversal_limit_in_words=2 ** 62
)
for i in range(0, size):
assert msg.structList[i].uInt8Field == 0

111
test/test_struct.py
View File

@@ -11,17 +11,17 @@ this_dir = os.path.dirname(__file__)
@pytest.fixture
def addressbook():
return capnp.load(os.path.join(this_dir, 'addressbook.capnp'))
return capnp.load(os.path.join(this_dir, "addressbook.capnp"))
@pytest.fixture
def all_types():
return capnp.load(os.path.join(this_dir, 'all_types.capnp'))
return capnp.load(os.path.join(this_dir, "all_types.capnp"))
def test_which_builder(addressbook):
addresses = addressbook.AddressBook.new_message()
people = addresses.init('people', 2)
people = addresses.init("people", 2)
alice = people[0]
alice.employment.school = "MIT"
@@ -53,7 +53,7 @@ def test_which_reader(addressbook):
def writeAddressBook(fd):
message = capnp._MallocMessageBuilder()
addressBook = message.init_root(addressbook.AddressBook)
people = addressBook.init('people', 2)
people = addressBook.init("people", 2)
alice = people[0]
alice.employment.school = "MIT"
@@ -89,14 +89,14 @@ def test_which_reader(addressbook):
@pytest.mark.skipif(
capnp.version.LIBCAPNP_VERSION < 5000,
reason="Using ints as enums requires v0.5.0+ of the C++ capnp library"
reason="Using ints as enums requires v0.5.0+ of the C++ capnp library",
)
def test_enum(addressbook):
addresses = addressbook.AddressBook.new_message()
people = addresses.init('people', 2)
people = addresses.init("people", 2)
alice = people[0]
phones = alice.init('phones', 2)
phones = alice.init("phones", 2)
assert phones[0].type == phones[1].type
@@ -104,7 +104,7 @@ def test_enum(addressbook):
assert phones[0].type != phones[1].type
phones[1].type = 'home'
phones[1].type = "home"
assert phones[0].type == phones[1].type
@@ -112,12 +112,12 @@ def test_enum(addressbook):
def test_builder_set(addressbook):
person = addressbook.Person.new_message()
person.name = 'test'
person.name = "test"
assert person.name == 'test'
assert person.name == "test"
with pytest.raises(AttributeError):
person.foo = 'test'
person.foo = "test"
def test_builder_set_from_list(all_types):
@@ -150,9 +150,9 @@ def test_unicode_str(all_types):
msg = all_types.TestAllTypes.new_message()
if sys.version_info[0] == 2:
msg.textField = u"f\u00e6oo".encode('utf-8')
msg.textField = u"f\u00e6oo".encode("utf-8")
assert msg.textField.decode('utf-8') == u"f\u00e6oo"
assert msg.textField.decode("utf-8") == u"f\u00e6oo"
else:
msg.textField = "f\u00e6oo"
@@ -164,11 +164,13 @@ def test_new_message(all_types):
assert msg.int32Field == 100
msg = all_types.TestAllTypes.new_message(structField={'int32Field': 100})
msg = all_types.TestAllTypes.new_message(structField={"int32Field": 100})
assert msg.structField.int32Field == 100
msg = all_types.TestAllTypes.new_message(structList=[{'int32Field': 100}, {'int32Field': 101}])
msg = all_types.TestAllTypes.new_message(
structList=[{"int32Field": 100}, {"int32Field": 101}]
)
assert msg.structList[0].int32Field == 100
assert msg.structList[1].int32Field == 101
@@ -177,7 +179,7 @@ def test_new_message(all_types):
assert msg.int32Field == 100
msg = all_types.TestAllTypes.new_message(**{'int32Field': 100, 'int64Field': 101})
msg = all_types.TestAllTypes.new_message(**{"int32Field": 100, "int64Field": 101})
assert msg.int32Field == 100
assert msg.int64Field == 101
@@ -186,45 +188,55 @@ def test_new_message(all_types):
def test_set_dict(all_types):
msg = all_types.TestAllTypes.new_message()
msg.structField = {'int32Field': 100}
msg.structField = {"int32Field": 100}
assert msg.structField.int32Field == 100
msg.init('structList', 2)
msg.structList[0] = {'int32Field': 102}
msg.init("structList", 2)
msg.structList[0] = {"int32Field": 102}
assert msg.structList[0].int32Field == 102
def test_set_dict_union(addressbook):
person = addressbook.Person.new_message(**{'employment': {'employer': {'name': 'foo'}}})
person = addressbook.Person.new_message(
**{"employment": {"employer": {"name": "foo"}}}
)
assert person.employment.which == addressbook.Person.Employment.employer
assert person.employment.employer.name == 'foo'
assert person.employment.employer.name == "foo"
def test_union_enum(all_types):
assert all_types.UnionAllTypes.Union.UnionStructField1 == 0
assert all_types.UnionAllTypes.Union.UnionStructField2 == 1
msg = all_types.UnionAllTypes.new_message(**{'unionStructField1': {'textField': "foo"}})
msg = all_types.UnionAllTypes.new_message(
**{"unionStructField1": {"textField": "foo"}}
)
assert msg.which == all_types.UnionAllTypes.Union.UnionStructField1
assert msg.which == 'unionStructField1'
assert msg.which == "unionStructField1"
assert msg.which == 0
msg = all_types.UnionAllTypes.new_message(**{'unionStructField2': {'textField': "foo"}})
msg = all_types.UnionAllTypes.new_message(
**{"unionStructField2": {"textField": "foo"}}
)
assert msg.which == all_types.UnionAllTypes.Union.UnionStructField2
assert msg.which == 'unionStructField2'
assert msg.which == "unionStructField2"
assert msg.which == 1
assert all_types.GroupedUnionAllTypes.Union.G1 == 0
assert all_types.GroupedUnionAllTypes.Union.G2 == 1
msg = all_types.GroupedUnionAllTypes.new_message(**{'g1': {'unionStructField1': {'textField': "foo"}}})
msg = all_types.GroupedUnionAllTypes.new_message(
**{"g1": {"unionStructField1": {"textField": "foo"}}}
)
assert msg.which == all_types.GroupedUnionAllTypes.Union.G1
msg = all_types.GroupedUnionAllTypes.new_message(**{'g2': {'unionStructField2': {'textField': "foo"}}})
msg = all_types.GroupedUnionAllTypes.new_message(
**{"g2": {"unionStructField2": {"textField": "foo"}}}
)
assert msg.which == all_types.GroupedUnionAllTypes.Union.G2
msg = all_types.UnionAllTypes.new_message()
@@ -236,44 +248,55 @@ def isstr(s):
def test_to_dict_enum(addressbook):
person = addressbook.Person.new_message(**{'phones': [{'number': '999-9999', 'type': 'mobile'}]})
person = addressbook.Person.new_message(
**{"phones": [{"number": "999-9999", "type": "mobile"}]}
)
field = person.to_dict()['phones'][0]['type']
field = person.to_dict()["phones"][0]["type"]
assert isstr(field)
assert field == 'mobile'
assert field == "mobile"
def test_explicit_field(addressbook):
person = addressbook.Person.new_message(**{'name': 'Test'})
person = addressbook.Person.new_message(**{"name": "Test"})
name_field = addressbook.Person.schema.fields['name']
name_field = addressbook.Person.schema.fields["name"]
assert person.name == person._get_by_field(name_field)
assert person.name == person.as_reader()._get_by_field(name_field)
def test_to_dict_verbose(addressbook):
person = addressbook.Person.new_message(**{'name': 'Test'})
person = addressbook.Person.new_message(**{"name": "Test"})
assert person.to_dict(verbose=True)['phones'] == []
assert person.to_dict(verbose=True)["phones"] == []
if sys.version_info >= (2, 7):
assert person.to_dict(verbose=True, ordered=True)['phones'] == []
assert person.to_dict(verbose=True, ordered=True)["phones"] == []
with pytest.raises(KeyError):
assert person.to_dict()['phones'] == []
assert person.to_dict()["phones"] == []
def test_to_dict_ordered(addressbook):
person = addressbook.Person.new_message(**{
'name': 'Alice',
'phones': [{'type': 'mobile', 'number': '555-1212'}],
'id': 123,
'employment': {'school': 'MIT'}, 'email': 'alice@example.com'
})
person = addressbook.Person.new_message(
**{
"name": "Alice",
"phones": [{"type": "mobile", "number": "555-1212"}],
"id": 123,
"employment": {"school": "MIT"},
"email": "alice@example.com",
}
)
if sys.version_info >= (2, 7):
assert list(person.to_dict(ordered=True).keys()) == ['id', 'name', 'email', 'phones', 'employment']
assert list(person.to_dict(ordered=True).keys()) == [
"id",
"name",
"email",
"phones",
"employment",
]
else:
with pytest.raises(Exception):
person.to_dict(ordered=True)
@@ -281,7 +304,7 @@ def test_to_dict_ordered(addressbook):
def test_nested_list(addressbook):
struct = addressbook.NestedList.new_message()
struct.init('list', 2)
struct.init("list", 2)
struct.list.init(0, 1)
struct.list.init(1, 2)

39
test/test_threads.py
View File

@@ -1,6 +1,6 @@
'''
"""
thread test
'''
"""
import platform
import socket
@@ -16,13 +16,13 @@ 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"
platform.python_implementation() == "PyPy",
reason="pycapnp's GIL handling isn't working properly at the moment for PyPy",
)
def test_making_event_loop():
'''
"""
Event loop test
'''
"""
capnp.remove_event_loop(True)
capnp.create_event_loop()
@@ -31,13 +31,13 @@ def test_making_event_loop():
@pytest.mark.skipif(
platform.python_implementation() == 'PyPy',
reason="pycapnp's GIL handling isn't working properly at the moment for PyPy"
platform.python_implementation() == "PyPy",
reason="pycapnp's GIL handling isn't working properly at the moment for PyPy",
)
def test_making_threaded_event_loop():
'''
"""
Threaded event loop test
'''
"""
# The following raises a KjException, and if not caught causes an SIGABRT:
# kj/async.c++:973: failed: expected head == nullptr; EventLoop destroyed with events still in the queue.
# Memory leak?; head->trace() = kj::_::ForkHub<kj::_::Void>
@@ -54,27 +54,28 @@ def test_making_threaded_event_loop():
class Server(test_capability_capnp.TestInterface.Server):
'''
"""
Server
'''
"""
def __init__(self, val=100):
self.val = val
def foo(self, i, j, **kwargs):
'''
"""
foo
'''
"""
return str(i * 5 + self.val)
@pytest.mark.skipif(
platform.python_implementation() == 'PyPy',
reason="pycapnp's GIL handling isn't working properly at the moment for PyPy"
platform.python_implementation() == "PyPy",
reason="pycapnp's GIL handling isn't working properly at the moment for PyPy",
)
def test_using_threads():
'''
"""
Thread test
'''
"""
capnp.remove_event_loop(True)
capnp.create_event_loop(True)
@@ -94,4 +95,4 @@ def test_using_threads():
remote = cap.foo(i=5)
response = remote.wait()
assert response.x == '125'
assert response.x == "125"