forked from 0rpc/zerorpc-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_python_usage.py
More file actions
40 lines (29 loc) · 923 Bytes
/
test_python_usage.py
File metadata and controls
40 lines (29 loc) · 923 Bytes
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
from datetime import datetime
from zerorpc import Context
__author__ = 'nemo'
from nose.tools import assert_raises
import gevent
import zerorpc
from testutils import random_ipc_endpoint
def test_kwargs():
class Srv(object):
def echo(self, *args, **kwargs):
return args, kwargs
endpoint = random_ipc_endpoint()
context = Context()
context.register_serializer("pickle")
module = Srv()
server = zerorpc.Server(module, context=context)
server.bind(endpoint)
gevent.spawn(server.run)
client = zerorpc.Client(context=context)
client.connect(endpoint)
args = 1,2,3
kwargs = {'a':7, 'b':8, 'now': datetime.now()}
res = client.echo(*args, **kwargs)
assert len(res) == 2
assert res[0] == args
assert len(res[1]) == 3
assert 'a' in res[1] and 'b' in res[1] and isinstance(res[1]['now'], datetime)
client.close()
server.close()