346 lines
12 KiB
Python
346 lines
12 KiB
Python
#!/usr/bin/env python
|
|
|
|
import unittest
|
|
import io
|
|
import collections
|
|
|
|
import spdylay
|
|
|
|
class BufferList:
|
|
def __init__(self):
|
|
self.buffers = collections.deque()
|
|
|
|
def add_buffer(self, bytebuf):
|
|
self.buffers.append(bytebuf)
|
|
|
|
def get_bytes(self, length):
|
|
while self.buffers:
|
|
first = self.buffers[0]
|
|
data = first.read(length)
|
|
if data:
|
|
return data
|
|
else:
|
|
self.buffers.popleft()
|
|
return None
|
|
|
|
class IOBridge:
|
|
def __init__(self, inputs, outputs):
|
|
self.inputs = inputs
|
|
self.outputs = outputs
|
|
|
|
class Streams:
|
|
def __init__(self, iob):
|
|
self.iob = iob
|
|
self.streams = {}
|
|
self.recv_frames = []
|
|
self.recv_data = io.BytesIO()
|
|
|
|
def recv_cb(session, length):
|
|
iob = session.user_data.iob
|
|
return iob.inputs.get_bytes(length)
|
|
|
|
def send_cb(session, data):
|
|
iob = session.user_data.iob
|
|
iob.outputs.add_buffer(io.BytesIO(data))
|
|
return len(data)
|
|
|
|
def read_cb(session, stream_id, length, read_ctrl, source):
|
|
data = source.read(length)
|
|
if not data:
|
|
read_ctrl.flags = spdylay.READ_EOF
|
|
return data
|
|
|
|
def on_data_chunk_recv_cb(session, flags, stream_id, data):
|
|
session.user_data.recv_data.write(data)
|
|
|
|
def on_ctrl_recv_cb(session, frame):
|
|
session.user_data.recv_frames.append(frame)
|
|
|
|
class SpdylayTests(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
client_output = BufferList()
|
|
server_output = BufferList()
|
|
|
|
client_iob = IOBridge(server_output, client_output)
|
|
server_iob = IOBridge(client_output, server_output)
|
|
|
|
self.client_streams = Streams(client_iob)
|
|
self.server_streams = Streams(server_iob)
|
|
|
|
self.client_session = spdylay.Session(\
|
|
spdylay.CLIENT,
|
|
spdylay.PROTO_SPDY3,
|
|
user_data=self.client_streams,
|
|
recv_cb=recv_cb,
|
|
send_cb=send_cb,
|
|
on_ctrl_recv_cb=on_ctrl_recv_cb,
|
|
on_data_chunk_recv_cb=on_data_chunk_recv_cb)
|
|
|
|
self.server_session = spdylay.Session(\
|
|
spdylay.SERVER,
|
|
spdylay.PROTO_SPDY3,
|
|
user_data=self.server_streams,
|
|
recv_cb=recv_cb,
|
|
send_cb=send_cb,
|
|
on_ctrl_recv_cb=on_ctrl_recv_cb,
|
|
on_data_chunk_recv_cb=on_data_chunk_recv_cb)
|
|
|
|
def test_submit_request_and_response(self):
|
|
data_prd = spdylay.DataProvider(io.BytesIO(b'Hello World'), read_cb)
|
|
self.client_session.submit_request(0, [(u':method', u'POST')],
|
|
data_prd=data_prd,
|
|
stream_user_data=data_prd)
|
|
self.client_session.send()
|
|
self.server_session.recv()
|
|
|
|
self.assertEqual(1, len(self.server_streams.recv_frames))
|
|
frame = self.server_streams.recv_frames[0]
|
|
self.assertEqual(spdylay.SYN_STREAM, frame.frame_type)
|
|
self.assertEqual(1, frame.stream_id)
|
|
self.assertEqual(0, frame.assoc_stream_id)
|
|
self.assertEqual(0, frame.pri)
|
|
self.assertEqual((u':method', u'POST'), frame.nv[0])
|
|
|
|
self.assertEqual(b'Hello World',
|
|
self.server_streams.recv_data.getvalue())
|
|
|
|
self.assertEqual(data_prd, self.client_session.get_stream_user_data(1))
|
|
|
|
data_prd = spdylay.DataProvider(io.BytesIO(b'Foo the bar'), read_cb)
|
|
self.server_session.submit_response(1, [(u':status', u'200 OK')],
|
|
data_prd=data_prd)
|
|
self.server_session.send()
|
|
self.client_session.recv()
|
|
|
|
self.assertEqual(1, len(self.client_streams.recv_frames))
|
|
frame = self.client_streams.recv_frames[0]
|
|
self.assertEqual(spdylay.SYN_REPLY, frame.frame_type)
|
|
self.assertEqual(1, frame.stream_id)
|
|
self.assertEqual((u':status', u'200 OK'), frame.nv[0])
|
|
|
|
self.assertEqual(b'Foo the bar',
|
|
self.client_streams.recv_data.getvalue())
|
|
|
|
def test_submit_syn_stream_and_syn_stream(self):
|
|
self.client_session.submit_syn_stream(spdylay.CTRL_FLAG_FIN, 2,
|
|
[(u':path', u'/')])
|
|
self.client_session.send()
|
|
self.server_session.recv()
|
|
|
|
self.assertEqual(1, len(self.server_streams.recv_frames))
|
|
frame = self.server_streams.recv_frames[0]
|
|
self.assertEqual(spdylay.SYN_STREAM, frame.frame_type)
|
|
self.assertEqual(1, frame.stream_id)
|
|
self.assertEqual(0, frame.assoc_stream_id)
|
|
self.assertEqual(2, frame.pri)
|
|
self.assertEqual((u':path', u'/'), frame.nv[0])
|
|
|
|
self.server_session.submit_syn_reply(spdylay.CTRL_FLAG_FIN, 1,
|
|
[(u':version', u'HTTP/1.1')])
|
|
self.server_session.send()
|
|
self.client_session.recv()
|
|
|
|
self.assertEqual(1, len(self.client_streams.recv_frames))
|
|
frame = self.client_streams.recv_frames[0]
|
|
self.assertEqual(spdylay.SYN_REPLY, frame.frame_type)
|
|
self.assertEqual(1, frame.stream_id)
|
|
self.assertEqual((u':version', u'HTTP/1.1'), frame.nv[0])
|
|
|
|
def test_submit_rst_stream(self):
|
|
self.client_session.submit_syn_stream(spdylay.CTRL_FLAG_FIN, 2,
|
|
[(u':path', u'/')])
|
|
self.client_session.send()
|
|
self.server_session.recv()
|
|
|
|
self.server_session.submit_rst_stream(1, spdylay.PROTOCOL_ERROR)
|
|
self.server_session.send()
|
|
self.client_session.recv()
|
|
|
|
self.assertEqual(1, len(self.client_streams.recv_frames))
|
|
frame = self.client_streams.recv_frames[0]
|
|
self.assertEqual(spdylay.RST_STREAM, frame.frame_type)
|
|
self.assertEqual(1, frame.stream_id)
|
|
self.assertEqual(spdylay.PROTOCOL_ERROR, frame.status_code)
|
|
|
|
def test_submit_goaway(self):
|
|
self.client_session.submit_goaway(spdylay.GOAWAY_PROTOCOL_ERROR)
|
|
self.client_session.send()
|
|
self.server_session.recv()
|
|
|
|
self.assertEqual(1, len(self.server_streams.recv_frames))
|
|
frame = self.server_streams.recv_frames[0]
|
|
self.assertEqual(spdylay.GOAWAY, frame.frame_type)
|
|
self.assertEqual(spdylay.GOAWAY_PROTOCOL_ERROR, frame.status_code)
|
|
|
|
def test_resume_data(self):
|
|
self.assertFalse(self.client_session.resume_data(1))
|
|
|
|
def test_get_pri_lowest(self):
|
|
self.assertEqual(7, self.client_session.get_pri_lowest())
|
|
|
|
def test_fail_session(self):
|
|
self.client_session.fail_session(spdylay.GOAWAY_PROTOCOL_ERROR)
|
|
self.client_session.send()
|
|
self.server_session.recv()
|
|
|
|
self.assertEqual(1, len(self.server_streams.recv_frames))
|
|
frame = self.server_streams.recv_frames[0]
|
|
self.assertEqual(spdylay.GOAWAY, frame.frame_type)
|
|
self.assertEqual(spdylay.GOAWAY_PROTOCOL_ERROR, frame.status_code)
|
|
|
|
self.assertFalse(self.client_session.want_read())
|
|
self.assertFalse(self.client_session.want_write())
|
|
|
|
def test_deferred_data(self):
|
|
def deferred_read_cb(session, stream_id, length, read_ctrl, source):
|
|
return spdylay.ERR_DEFERRED
|
|
|
|
data_prd = spdylay.DataProvider(io.BytesIO(b'Hello World'),
|
|
deferred_read_cb)
|
|
self.client_session.submit_request(0, [(u':method', u'POST')],
|
|
data_prd=data_prd,
|
|
stream_user_data=data_prd)
|
|
self.client_session.send()
|
|
self.server_session.recv()
|
|
|
|
self.assertEqual(1, len(self.server_streams.recv_frames))
|
|
frame = self.server_streams.recv_frames[0]
|
|
self.assertEqual(spdylay.SYN_STREAM, frame.frame_type)
|
|
self.assertEqual(1, frame.stream_id)
|
|
self.assertEqual(0, frame.assoc_stream_id)
|
|
self.assertEqual(0, frame.pri)
|
|
self.assertEqual((u':method', u'POST'), frame.nv[0])
|
|
|
|
self.assertEqual(b'', self.server_streams.recv_data.getvalue())
|
|
|
|
data_prd.read_cb = read_cb
|
|
|
|
self.client_session.resume_data(1)
|
|
|
|
self.client_session.send()
|
|
self.server_session.recv()
|
|
|
|
self.assertEqual(b'Hello World',
|
|
self.server_streams.recv_data.getvalue())
|
|
|
|
def test_recv_cb_eof(self):
|
|
def eof_recv_cb(session, length):
|
|
raise spdylay.EOFError()
|
|
|
|
self.client_session = spdylay.Session(\
|
|
spdylay.CLIENT,
|
|
spdylay.PROTO_SPDY3,
|
|
user_data=self.client_streams,
|
|
recv_cb=eof_recv_cb)
|
|
|
|
with self.assertRaises(spdylay.EOFError):
|
|
self.client_session.recv()
|
|
|
|
def test_recv_cb_callback_failure(self):
|
|
def cbfail_recv_cb(session, length):
|
|
raise spdylay.CallbackFailureError()
|
|
|
|
self.client_session = spdylay.Session(\
|
|
spdylay.CLIENT,
|
|
spdylay.PROTO_SPDY3,
|
|
user_data=self.client_streams,
|
|
recv_cb=cbfail_recv_cb)
|
|
|
|
with self.assertRaises(spdylay.CallbackFailureError):
|
|
self.client_session.recv()
|
|
|
|
def test_send_cb_callback_failure(self):
|
|
def cbfail_send_cb(session, data):
|
|
raise spdylay.CallbackFailureError()
|
|
|
|
self.client_session = spdylay.Session(\
|
|
spdylay.CLIENT,
|
|
spdylay.PROTO_SPDY3,
|
|
user_data=self.client_streams,
|
|
send_cb=cbfail_send_cb)
|
|
|
|
self.client_session.submit_goaway(spdylay.GOAWAY_OK)
|
|
|
|
with self.assertRaises(spdylay.CallbackFailureError):
|
|
self.client_session.send()
|
|
|
|
def test_submit_data(self):
|
|
self.client_session.submit_syn_stream(spdylay.CTRL_FLAG_NONE, 2,
|
|
[(u':path', u'/')])
|
|
self.client_session.send()
|
|
self.server_session.recv()
|
|
|
|
self.assertEqual(1, len(self.server_streams.recv_frames))
|
|
frame = self.server_streams.recv_frames[0]
|
|
self.assertEqual(spdylay.SYN_STREAM, frame.frame_type)
|
|
self.assertEqual(1, frame.stream_id)
|
|
|
|
data_prd = spdylay.DataProvider(io.BytesIO(b'Hello World'), read_cb)
|
|
self.client_session.submit_data(1, spdylay.DATA_FLAG_FIN, data_prd)
|
|
self.client_session.send()
|
|
self.server_session.recv()
|
|
|
|
self.assertEqual(b'Hello World',
|
|
self.server_streams.recv_data.getvalue())
|
|
|
|
def test_submit_headers(self):
|
|
self.client_session.submit_syn_stream(spdylay.CTRL_FLAG_NONE, 2,
|
|
[(u':path', u'/')])
|
|
self.client_session.send()
|
|
self.server_session.recv()
|
|
|
|
self.assertEqual(1, len(self.server_streams.recv_frames))
|
|
frame = self.server_streams.recv_frames[0]
|
|
self.assertEqual(spdylay.SYN_STREAM, frame.frame_type)
|
|
self.assertEqual(1, frame.stream_id)
|
|
|
|
self.client_session.submit_headers(spdylay.CTRL_FLAG_FIN, 1,
|
|
[(u':host', u'localhost')])
|
|
self.client_session.send()
|
|
self.server_session.recv()
|
|
|
|
self.assertEqual(2, len(self.server_streams.recv_frames))
|
|
frame = self.server_streams.recv_frames[1]
|
|
self.assertEqual(spdylay.HEADERS, frame.frame_type)
|
|
self.assertEqual(1, frame.stream_id)
|
|
self.assertEqual((u':host', u'localhost'), frame.nv[0])
|
|
|
|
def test_submit_ping(self):
|
|
self.client_session.submit_ping()
|
|
self.client_session.send()
|
|
self.server_session.recv()
|
|
|
|
self.assertEqual(1, len(self.server_streams.recv_frames))
|
|
frame = self.server_streams.recv_frames[0]
|
|
self.assertEqual(spdylay.PING, frame.frame_type)
|
|
self.assertEqual(1, frame.unique_id)
|
|
|
|
def test_submit_window_update(self):
|
|
self.client_session.submit_syn_stream(spdylay.CTRL_FLAG_NONE, 2,
|
|
[(u':path', u'/')])
|
|
self.client_session.send()
|
|
self.server_session.recv()
|
|
|
|
self.assertEqual(1, len(self.server_streams.recv_frames))
|
|
frame = self.server_streams.recv_frames[0]
|
|
self.assertEqual(spdylay.SYN_STREAM, frame.frame_type)
|
|
self.assertEqual(1, frame.stream_id)
|
|
|
|
self.server_session.submit_window_update(1, 4096)
|
|
self.server_session.send()
|
|
self.client_session.recv()
|
|
|
|
self.assertEqual(1, len(self.client_streams.recv_frames))
|
|
frame = self.client_streams.recv_frames[0]
|
|
self.assertEqual(spdylay.WINDOW_UPDATE, frame.frame_type)
|
|
self.assertEqual(1, frame.stream_id)
|
|
self.assertEqual(4096, frame.delta_window_size)
|
|
|
|
def test_get_npn_protocols(self):
|
|
protos = spdylay.get_npn_protocols()
|
|
self.assertEqual(['spdy/3', 'spdy/2'], protos)
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|