def test_readonly(self):
client_conn = Connection(self.client_stream, client_side=True, readonly=True)
client_conn.initiate_connection()
client_conn.send_request(
client_conn.get_next_available_stream_id(),
HttpRequest(headers=[
(":method", "GET"),
(":path", "/"),
("aaa", "bbb")]))
with self.assertRaises(gen.TimeoutError):
yield gen.with_timeout(
timedelta(milliseconds=100),
self.server_stream.read_bytes(1))
评论列表
文章目录