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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
|
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import pytest
import six
import zmq
import zmq.green
from tinyrpc.transports import ServerTransport, ClientTransport
from tinyrpc.transports.zmq import ZmqServerTransport, ZmqClientTransport
class DummyServerTransport(ServerTransport):
def __init__(self):
self.messages = []
self.clients = {}
def receive_message(self):
return self.messages.pop()
def send_reply(self, context, message):
if not isinstance(message, str):
raise TypeError('Message must be str().')
self.clients[context].messages.append(message)
class DummyClientTransport(ClientTransport):
def __init__(self, server):
self.server = server
self.id = id(self)
self.server.clients[self.id] = self
self.messages = []
def send_message(self, message):
if not isinstance(message, str):
raise TypeError('Message must be str().')
self.server.messages.append((self.id, message))
def receive_reply(self):
return self.messages.pop()
ZMQ_ENDPOINT = 'inproc://example2'
@pytest.fixture(scope='session')
def zmq_context(request):
ctx = zmq.Context()
def fin():
request.addfinalizer(ctx.destroy())
return ctx
@pytest.fixture(scope='session')
def zmq_green_context(request):
ctx = zmq.Context()
def fin():
request.addfinalizer(ctx.destroy())
return ctx
@pytest.fixture(params=['dummy', 'zmq', 'zmq.green'])
def transport(request, zmq_context, zmq_green_context):
if request.param == 'dummy':
server = DummyServerTransport()
client = DummyClientTransport(server)
elif request.param in ('zmq', 'zmq.green'):
ctx = zmq_context if request.param == 'zmq' else zmq_green_context
server = ZmqServerTransport.create(ctx, ZMQ_ENDPOINT)
client = ZmqClientTransport.create(ctx, ZMQ_ENDPOINT)
def fin():
server.socket.close()
client.socket.close()
request.addfinalizer(fin)
else:
raise ValueError('Invalid transport.')
return (client, server)
SAMPLE_MESSAGES = ['asdf', 'loremipsum' * 1500, '', '\x00', 'b\x00a', '\r\n',
'\n', u'\u1234'.encode('utf8')]
BAD_MESSAGES = [u'asdf', u'', 1234, 1.2, None, True, False, ('foo',)]
@pytest.fixture(scope='session',
params=SAMPLE_MESSAGES)
def sample_msg(request):
return request.param
@pytest.fixture(scope='session',
params=SAMPLE_MESSAGES)
def sample_msg2(request):
return request.param
@pytest.fixture(scope='session',
params=BAD_MESSAGES)
def bad_msg(request):
return request.param
@pytest.mark.skipif(six.PY3, reason='Somehow fails on PY3')
def test_transport_rejects_bad_values(transport, sample_msg, bad_msg):
client, server = transport
with pytest.raises(TypeError):
client.send_message(bad_msg)
# FIXME: these tests need to be rethought, as they no longer work properly with
# the change to the interface of ClientTransport
# FIXME: the actual client needs tests as well
|
