codekingpro/portable-devtools
114k
1from tornado.concurrent import Future
2from tornado import gen
3from tornado import netutil
4from tornado.ioloop import IOLoop
5from tornado.iostream import (
6 IOStream,
7 SSLIOStream,
8 PipeIOStream,
9 StreamClosedError,
10 _StreamBuffer,
11)
12from tornado.httpclient import AsyncHTTPClient, HTTPResponse
13from tornado.httputil import HTTPHeaders
14from tornado.locks import Condition, Event
15from tornado.log import gen_log
16from tornado.netutil import ssl_options_to_context, ssl_wrap_socket
17from tornado.platform.asyncio import AddThreadSelectorEventLoop
18from tornado.tcpserver import TCPServer
19from tornado.testing import (
20 AsyncHTTPTestCase,
21 AsyncHTTPSTestCase,
22 AsyncTestCase,
23 bind_unused_port,
24 ExpectLog,
25 gen_test,
26)
27from tornado.test.util import (
28 skipIfNonUnix,
29 refusing_port,
30 ignore_deprecation,
31 abstract_base_test,
32)
33from tornado.web import RequestHandler, Application
34import asyncio
35import errno
36import hashlib
37import logging
38import os
39import platform
40import random
41import socket
42import ssl
43import typing
44from unittest import mock
45import unittest
46
47
48def _server_ssl_options():
49 return dict(
50 certfile=os.path.join(os.path.dirname(__file__), "test.crt"),
51 keyfile=os.path.join(os.path.dirname(__file__), "test.key"),
52 )
53
54
55class HelloHandler(RequestHandler):
56 def get(self):
57 self.write("Hello")
58
59
60@abstract_base_test
61class TestIOStreamWebMixin(AsyncTestCase):
62 # We want to run these tests with both AsyncHTTPTestCase and AsyncHTTPSTestCase,
63 # but this leads to some tricky inheritance situations. We want this class's
64 # get_app, but the test classes's get_http_port and fetch. There's no way to make
65 # the method resolution order to do what we want in all cases, so the current
66 # state is that that AsyncHTTP(S)TestCase must be the first base class of the
67 # final class, and that class must define a get_app method that calls mixin_get_app.
68 #
69 # Alternatives include defining this class in a factory that can change the base class
70 # or refactoring to use composition instead of inheritance for the http components.
71 def _make_client_iostream(self):
72 raise NotImplementedError()
73
74 def mixin_get_app(self):
75 return Application([("/", HelloHandler)])
76
77 def get_http_port(self) -> int:
78 raise NotImplementedError()
79
80 def fetch(
81 self, path: str, raise_error: bool = False, **kwargs: typing.Any
82 ) -> HTTPResponse:
83 # To be filled in by mixing in AsyncHTTPTestCase or AsyncHTTPSTestCase
84 raise NotImplementedError()
85
86 def test_connection_closed(self):
87 # When a server sends a response and then closes the connection,
88 # the client must be allowed to read the data before the IOStream
89 # closes itself. Epoll reports closed connections with a separate
90 # EPOLLRDHUP event delivered at the same time as the read event,
91 # while kqueue reports them as a second read/write event with an EOF
92 # flag.
93 if (
94 AsyncHTTPClient.configured_class().__name__.endswith("CurlAsyncHTTPClient")
95 and platform.system() == "Darwin"
96 ):
97 # It's possible that this is Tornado's fault, either in AsyncIOLoop or in
98 # CurlAsyncHTTPClient, but we've also seen this kind of issue in libcurl itself
99 # (especially a long time ago). The error is tied to the use of Apple's
100 # SecureTransport instead of OpenSSL.
101 self.skipTest("libcurl doesn't handle closed connections cleanly on macOS")
102 response = self.fetch("/", headers={"Connection": "close"})
103 response.rethrow()
104
105 @gen_test
106 def test_read_until_close(self):
107 stream = self._make_client_iostream()
108 yield stream.connect(("127.0.0.1", self.get_http_port()))
109 stream.write(b"GET / HTTP/1.0\r\n\r\n")
110
111 data = yield stream.read_until_close()
112 self.assertTrue(data.startswith(b"HTTP/1.1 200"))
113 self.assertTrue(data.endswith(b"Hello"))
114
115 @gen_test
116 def test_read_zero_bytes(self):
117 self.stream = self._make_client_iostream()
118 yield self.stream.connect(("127.0.0.1", self.get_http_port()))
119 self.stream.write(b"GET / HTTP/1.0\r\n\r\n")
120
121 # normal read
122 data = yield self.stream.read_bytes(9)
123 self.assertEqual(data, b"HTTP/1.1 ")
124
125 # zero bytes
126 data = yield self.stream.read_bytes(0)
127 self.assertEqual(data, b"")
128
129 # another normal read
130 data = yield self.stream.read_bytes(3)
131 self.assertEqual(data, b"200")
132
133 self.stream.close()
134
135 @gen_test
136 def test_write_while_connecting(self):
137 stream = self._make_client_iostream()
138 connect_fut = stream.connect(("127.0.0.1", self.get_http_port()))
139 # unlike the previous tests, try to write before the connection
140 # is complete.
141 write_fut = stream.write(b"GET / HTTP/1.0\r\nConnection: close\r\n\r\n")
142 self.assertFalse(connect_fut.done())
143
144 # connect will always complete before write.
145 it = gen.WaitIterator(connect_fut, write_fut)
146 resolved_order = []
147 while not it.done():
148 yield it.next()
149 resolved_order.append(it.current_future)
150 self.assertEqual(resolved_order, [connect_fut, write_fut])
151
152 data = yield stream.read_until_close()
153 self.assertTrue(data.endswith(b"Hello"))
154
155 stream.close()
156
157 @gen_test
158 def test_future_interface(self):
159 """Basic test of IOStream's ability to return Futures."""
160 stream = self._make_client_iostream()
161 connect_result = yield stream.connect(("127.0.0.1", self.get_http_port()))
162 self.assertIs(connect_result, stream)
163 yield stream.write(b"GET / HTTP/1.0\r\n\r\n")
164 first_line = yield stream.read_until(b"\r\n")
165 self.assertEqual(first_line, b"HTTP/1.1 200 OK\r\n")
166 # callback=None is equivalent to no callback.
167 header_data = yield stream.read_until(b"\r\n\r\n")
168 headers = HTTPHeaders.parse(header_data.decode("latin1"))
169 content_length = int(headers["Content-Length"])
170 body = yield stream.read_bytes(content_length)
171 self.assertEqual(body, b"Hello")
172 stream.close()
173
174 @gen_test
175 def test_future_close_while_reading(self):
176 stream = self._make_client_iostream()
177 yield stream.connect(("127.0.0.1", self.get_http_port()))
178 yield stream.write(b"GET / HTTP/1.0\r\n\r\n")
179 with self.assertRaises(StreamClosedError):
180 yield stream.read_bytes(1024 * 1024)
181 stream.close()
182
183 @gen_test
184 def test_future_read_until_close(self):
185 # Ensure that the data comes through before the StreamClosedError.
186 stream = self._make_client_iostream()
187 yield stream.connect(("127.0.0.1", self.get_http_port()))
188 yield stream.write(b"GET / HTTP/1.0\r\nConnection: close\r\n\r\n")
189 yield stream.read_until(b"\r\n\r\n")
190 body = yield stream.read_until_close()
191 self.assertEqual(body, b"Hello")
192
193 # Nothing else to read; the error comes immediately without waiting
194 # for yield.
195 with self.assertRaises(StreamClosedError):
196 stream.read_bytes(1)
197
198
199@abstract_base_test
200class TestReadWriteMixin(AsyncTestCase):
201 # Tests where one stream reads and the other writes.
202 # These should work for BaseIOStream implementations.
203
204 def make_iostream_pair(self, **kwargs):
205 raise NotImplementedError
206
207 def iostream_pair(self, **kwargs):
208 """Like make_iostream_pair, but called by ``async with``.
209
210 In py37 this becomes simpler with contextlib.asynccontextmanager.
211 """
212
213 class IOStreamPairContext:
214 def __init__(self, test, kwargs):
215 self.test = test
216 self.kwargs = kwargs
217
218 async def __aenter__(self):
219 self.pair = await self.test.make_iostream_pair(**self.kwargs)
220 return self.pair
221
222 async def __aexit__(self, typ, value, tb):
223 for s in self.pair:
224 s.close()
225
226 return IOStreamPairContext(self, kwargs)
227
228 @gen_test
229 def test_write_zero_bytes(self):
230 # Attempting to write zero bytes should run the callback without
231 # going into an infinite loop.
232 rs, ws = yield self.make_iostream_pair()
233 yield ws.write(b"")
234 ws.close()
235 rs.close()
236
237 @gen_test
238 def test_future_delayed_close_callback(self):
239 # Same as test_delayed_close_callback, but with the future interface.
240 rs, ws = yield self.make_iostream_pair()
241
242 try:
243 ws.write(b"12")
244 chunks = []
245 chunks.append((yield rs.read_bytes(1)))
246 ws.close()
247 chunks.append((yield rs.read_bytes(1)))
248 self.assertEqual(chunks, [b"1", b"2"])
249 finally:
250 ws.close()
251 rs.close()
252
253 @gen_test
254 def test_close_buffered_data(self):
255 # Similar to the previous test, but with data stored in the OS's
256 # socket buffers instead of the IOStream's read buffer. Out-of-band
257 # close notifications must be delayed until all data has been
258 # drained into the IOStream buffer. (epoll used to use out-of-band
259 # close events with EPOLLRDHUP, but no longer)
260 #
261 # This depends on the read_chunk_size being smaller than the
262 # OS socket buffer, so make it small.
263 rs, ws = yield self.make_iostream_pair(read_chunk_size=256)
264 try:
265 ws.write(b"A" * 512)
266 data = yield rs.read_bytes(256)
267 self.assertEqual(b"A" * 256, data)
268 ws.close()
269 # Allow the close to propagate to the `rs` side of the
270 # connection. Using add_callback instead of add_timeout
271 # doesn't seem to work, even with multiple iterations
272 yield gen.sleep(0.01)
273 data = yield rs.read_bytes(256)
274 self.assertEqual(b"A" * 256, data)
275 finally:
276 ws.close()
277 rs.close()
278
279 @gen_test
280 def test_read_until_close_after_close(self):
281 # Similar to test_delayed_close_callback, but read_until_close takes
282 # a separate code path so test it separately.
283 rs, ws = yield self.make_iostream_pair()
284 try:
285 ws.write(b"1234")
286 # Read one byte to make sure the client has received the data.
287 # It won't run the close callback as long as there is more buffered
288 # data that could satisfy a later read.
289 data = yield rs.read_bytes(1)
290 ws.close()
291 self.assertEqual(data, b"1")
292 data = yield rs.read_until_close()
293 self.assertEqual(data, b"234")
294 finally:
295 ws.close()
296 rs.close()
297
298 @gen_test
299 def test_large_read_until(self):
300 # Performance test: read_until used to have a quadratic component
301 # so a read_until of 4MB would take 8 seconds; now it takes 0.25
302 # seconds.
303 rs, ws = yield self.make_iostream_pair()
304 try:
305 # This test fails on pypy with ssl. I think it's because
306 # pypy's gc defeats moves objects, breaking the
307 # "frozen write buffer" assumption.
308 if (
309 isinstance(rs, SSLIOStream)
310 and platform.python_implementation() == "PyPy"
311 ):
312 raise unittest.SkipTest("pypy gc causes problems with openssl")
313 NUM_KB = 4096
314 for i in range(NUM_KB):
315 ws.write(b"A" * 1024)
316 ws.write(b"\r\n")
317 data = yield rs.read_until(b"\r\n")
318 self.assertEqual(len(data), NUM_KB * 1024 + 2)
319 finally:
320 ws.close()
321 rs.close()
322
323 @gen_test
324 async def test_read_until_with_close_after_second_packet(self):
325 # This is a regression test for a regression in Tornado 6.0
326 # (maybe 6.0.3?) reported in
327 # https://github.com/tornadoweb/tornado/issues/2717
328 #
329 # The data arrives in two chunks; the stream is closed at the
330 # same time that the second chunk is received. If the second
331 # chunk is larger than the first, it works, but when this bug
332 # existed it would fail if the second chunk were smaller than
333 # the first. This is due to the optimization that the
334 # read_until condition is only checked when the buffer doubles
335 # in size
336 async with self.iostream_pair() as (rs, ws):
337 rf = asyncio.ensure_future(rs.read_until(b"done"))
338 # We need to wait for the read_until to actually start. On
339 # windows that's tricky because the selector runs in
340 # another thread; sleeping is the simplest way.
341 await asyncio.sleep(0.1)
342 await ws.write(b"x" * 2048)
343 ws.write(b"done")
344 ws.close()
345 await rf
346
347 @gen_test
348 async def test_read_until_unsatisfied_after_close(self):
349 # If a stream is closed while reading, it raises
350 # StreamClosedError instead of UnsatisfiableReadError (the
351 # latter should only be raised when byte limits are reached).
352 # The particular scenario tested here comes from #2717.
353 async with self.iostream_pair() as (rs, ws):
354 rf = asyncio.ensure_future(rs.read_until(b"done"))
355 await ws.write(b"x" * 2048)
356 ws.write(b"foo")
357 ws.close()
358 with self.assertRaises(StreamClosedError):
359 await rf
360
361 @gen_test
362 def test_close_callback_with_pending_read(self):
363 # Regression test for a bug that was introduced in 2.3
364 # where the IOStream._close_callback would never be called
365 # if there were pending reads.
366 OK = b"OK\r\n"
367 rs, ws = yield self.make_iostream_pair()
368 event = Event()
369 rs.set_close_callback(event.set)
370 try:
371 ws.write(OK)
372 res = yield rs.read_until(b"\r\n")
373 self.assertEqual(res, OK)
374
375 ws.close()
376 rs.read_until(b"\r\n")
377 # If _close_callback (self.stop) is not called,
378 # an AssertionError: Async operation timed out after 5 seconds
379 # will be raised.
380 yield event.wait()
381 finally:
382 ws.close()
383 rs.close()
384
385 @gen_test
386 def test_future_close_callback(self):
387 # Regression test for interaction between the Future read interfaces
388 # and IOStream._maybe_add_error_listener.
389 rs, ws = yield self.make_iostream_pair()
390 closed = [False]
391 cond = Condition()
392
393 def close_callback():
394 closed[0] = True
395 cond.notify()
396
397 rs.set_close_callback(close_callback)
398 try:
399 ws.write(b"a")
400 res = yield rs.read_bytes(1)
401 self.assertEqual(res, b"a")
402 self.assertFalse(closed[0])
403 ws.close()
404 yield cond.wait()
405 self.assertTrue(closed[0])
406 finally:
407 rs.close()
408 ws.close()
409
410 @gen_test
411 def test_write_memoryview(self):
412 rs, ws = yield self.make_iostream_pair()
413 try:
414 fut = rs.read_bytes(4)
415 ws.write(memoryview(b"hello"))
416 data = yield fut
417 self.assertEqual(data, b"hell")
418 finally:
419 ws.close()
420 rs.close()
421
422 @gen_test
423 def test_read_bytes_partial(self):
424 rs, ws = yield self.make_iostream_pair()
425 try:
426 # Ask for more than is available with partial=True
427 fut = rs.read_bytes(50, partial=True)
428 ws.write(b"hello")
429 data = yield fut
430 self.assertEqual(data, b"hello")
431
432 # Ask for less than what is available; num_bytes is still
433 # respected.
434 fut = rs.read_bytes(3, partial=True)
435 ws.write(b"world")
436 data = yield fut
437 self.assertEqual(data, b"wor")
438
439 # Partial reads won't return an empty string, but read_bytes(0)
440 # will.
441 data = yield rs.read_bytes(0, partial=True)
442 self.assertEqual(data, b"")
443 finally:
444 ws.close()
445 rs.close()
446
447 @gen_test
448 def test_read_until_max_bytes(self):
449 rs, ws = yield self.make_iostream_pair()
450 closed = Event()
451 rs.set_close_callback(closed.set)
452 try:
453 # Extra room under the limit
454 fut = rs.read_until(b"def", max_bytes=50)
455 ws.write(b"abcdef")
456 data = yield fut
457 self.assertEqual(data, b"abcdef")
458
459 # Just enough space
460 fut = rs.read_until(b"def", max_bytes=6)
461 ws.write(b"abcdef")
462 data = yield fut
463 self.assertEqual(data, b"abcdef")
464
465 # Not enough space, but we don't know it until all we can do is
466 # log a warning and close the connection.
467 with ExpectLog(gen_log, "Unsatisfiable read", level=logging.INFO):
468 fut = rs.read_until(b"def", max_bytes=5)
469 ws.write(b"123456")
470 yield closed.wait()
471 finally:
472 ws.close()
473 rs.close()
474
475 @gen_test
476 def test_read_until_max_bytes_inline(self):
477 rs, ws = yield self.make_iostream_pair()
478 closed = Event()
479 rs.set_close_callback(closed.set)
480 try:
481 # Similar to the error case in the previous test, but the
482 # ws writes first so rs reads are satisfied
483 # inline. For consistency with the out-of-line case, we
484 # do not raise the error synchronously.
485 ws.write(b"123456")
486 with ExpectLog(gen_log, "Unsatisfiable read", level=logging.INFO):
487 with self.assertRaises(StreamClosedError):
488 yield rs.read_until(b"def", max_bytes=5)
489 yield closed.wait()
490 finally:
491 ws.close()
492 rs.close()
493
494 @gen_test
495 def test_read_until_max_bytes_ignores_extra(self):
496 rs, ws = yield self.make_iostream_pair()
497 closed = Event()
498 rs.set_close_callback(closed.set)
499 try:
500 # Even though data that matches arrives the same packet that
501 # puts us over the limit, we fail the request because it was not
502 # found within the limit.
503 ws.write(b"abcdef")
504 with ExpectLog(gen_log, "Unsatisfiable read", level=logging.INFO):
505 rs.read_until(b"def", max_bytes=5)
506 yield closed.wait()
507 finally:
508 ws.close()
509 rs.close()
510
511 @gen_test
512 def test_read_until_regex_max_bytes(self):
513 rs, ws = yield self.make_iostream_pair()
514 closed = Event()
515 rs.set_close_callback(closed.set)
516 try:
517 # Extra room under the limit
518 fut = rs.read_until_regex(b"def", max_bytes=50)
519 ws.write(b"abcdef")
520 data = yield fut
521 self.assertEqual(data, b"abcdef")
522
523 # Just enough space
524 fut = rs.read_until_regex(b"def", max_bytes=6)
525 ws.write(b"abcdef")
526 data = yield fut
527 self.assertEqual(data, b"abcdef")
528
529 # Not enough space, but we don't know it until all we can do is
530 # log a warning and close the connection.
531 with ExpectLog(gen_log, "Unsatisfiable read", level=logging.INFO):
532 rs.read_until_regex(b"def", max_bytes=5)
533 ws.write(b"123456")
534 yield closed.wait()
535 finally:
536 ws.close()
537 rs.close()
538
539 @gen_test
540 def test_read_until_regex_max_bytes_inline(self):
541 rs, ws = yield self.make_iostream_pair()
542 closed = Event()
543 rs.set_close_callback(closed.set)
544 try:
545 # Similar to the error case in the previous test, but the
546 # ws writes first so rs reads are satisfied
547 # inline. For consistency with the out-of-line case, we
548 # do not raise the error synchronously.
549 ws.write(b"123456")
550 with ExpectLog(gen_log, "Unsatisfiable read", level=logging.INFO):
551 rs.read_until_regex(b"def", max_bytes=5)
552 yield closed.wait()
553 finally:
554 ws.close()
555 rs.close()
556
557 @gen_test
558 def test_read_until_regex_max_bytes_ignores_extra(self):
559 rs, ws = yield self.make_iostream_pair()
560 closed = Event()
561 rs.set_close_callback(closed.set)
562 try:
563 # Even though data that matches arrives the same packet that
564 # puts us over the limit, we fail the request because it was not
565 # found within the limit.
566 ws.write(b"abcdef")
567 with ExpectLog(gen_log, "Unsatisfiable read", level=logging.INFO):
568 rs.read_until_regex(b"def", max_bytes=5)
569 yield closed.wait()
570 finally:
571 ws.close()
572 rs.close()
573
574 @gen_test
575 def test_small_reads_from_large_buffer(self):
576 # 10KB buffer size, 100KB available to read.
577 # Read 1KB at a time and make sure that the buffer is not eagerly
578 # filled.
579 rs, ws = yield self.make_iostream_pair(max_buffer_size=10 * 1024)
580 try:
581 ws.write(b"a" * 1024 * 100)
582 for i in range(100):
583 data = yield rs.read_bytes(1024)
584 self.assertEqual(data, b"a" * 1024)
585 finally:
586 ws.close()
587 rs.close()
588
589 @gen_test
590 def test_small_read_untils_from_large_buffer(self):
591 # 10KB buffer size, 100KB available to read.
592 # Read 1KB at a time and make sure that the buffer is not eagerly
593 # filled.
594 rs, ws = yield self.make_iostream_pair(max_buffer_size=10 * 1024)
595 try:
596 ws.write((b"a" * 1023 + b"\n") * 100)
597 for i in range(100):
598 data = yield rs.read_until(b"\n", max_bytes=4096)
599 self.assertEqual(data, b"a" * 1023 + b"\n")
600 finally:
601 ws.close()
602 rs.close()
603
604 @gen_test
605 def test_flow_control(self):
606 MB = 1024 * 1024
607 rs, ws = yield self.make_iostream_pair(max_buffer_size=5 * MB)
608 try:
609 # Client writes more than the rs will accept.
610 ws.write(b"a" * 10 * MB)
611 # The rs pauses while reading.
612 yield rs.read_bytes(MB)
613 yield gen.sleep(0.1)
614 # The ws's writes have been blocked; the rs can
615 # continue to read gradually.
616 for i in range(9):
617 yield rs.read_bytes(MB)
618 finally:
619 rs.close()
620 ws.close()
621
622 @gen_test
623 def test_read_into(self):
624 rs, ws = yield self.make_iostream_pair()
625
626 def sleep_some():
627 self.io_loop.run_sync(lambda: gen.sleep(0.05))
628
629 try:
630 buf = bytearray(10)
631 fut = rs.read_into(buf)
632 ws.write(b"hello")
633 yield gen.sleep(0.05)
634 self.assertTrue(rs.reading())
635 ws.write(b"world!!")
636 data = yield fut
637 self.assertFalse(rs.reading())
638 self.assertEqual(data, 10)
639 self.assertEqual(bytes(buf), b"helloworld")
640
641 # Existing buffer is fed into user buffer
642 fut = rs.read_into(buf)
643 yield gen.sleep(0.05)
644 self.assertTrue(rs.reading())
645 ws.write(b"1234567890")
646 data = yield fut
647 self.assertFalse(rs.reading())
648 self.assertEqual(data, 10)
649 self.assertEqual(bytes(buf), b"!!12345678")
650
651 # Existing buffer can satisfy read immediately
652 buf = bytearray(4)
653 ws.write(b"abcdefghi")
654 data = yield rs.read_into(buf)
655 self.assertEqual(data, 4)
656 self.assertEqual(bytes(buf), b"90ab")
657
658 data = yield rs.read_bytes(7)
659 self.assertEqual(data, b"cdefghi")
660 finally:
661 ws.close()
662 rs.close()
663
664 @gen_test
665 def test_read_into_partial(self):
666 rs, ws = yield self.make_iostream_pair()
667
668 try:
669 # Partial read
670 buf = bytearray(10)
671 fut = rs.read_into(buf, partial=True)
672 ws.write(b"hello")
673 data = yield fut
674 self.assertFalse(rs.reading())
675 self.assertEqual(data, 5)
676 self.assertEqual(bytes(buf), b"hello\0\0\0\0\0")
677
678 # Full read despite partial=True
679 ws.write(b"world!1234567890")
680 data = yield rs.read_into(buf, partial=True)
681 self.assertEqual(data, 10)
682 self.assertEqual(bytes(buf), b"world!1234")
683
684 # Existing buffer can satisfy read immediately
685 data = yield rs.read_into(buf, partial=True)
686 self.assertEqual(data, 6)
687 self.assertEqual(bytes(buf), b"5678901234")
688
689 finally:
690 ws.close()
691 rs.close()
692
693 @gen_test
694 def test_read_into_zero_bytes(self):
695 rs, ws = yield self.make_iostream_pair()
696 try:
697 buf = bytearray()
698 fut = rs.read_into(buf)
699 self.assertEqual(fut.result(), 0)
700 finally:
701 ws.close()
702 rs.close()
703
704 @gen_test
705 def test_many_mixed_reads(self):
706 # Stress buffer handling when going back and forth between
707 # read_bytes() (using an internal buffer) and read_into()
708 # (using a user-allocated buffer).
709 r = random.Random(42)
710 nbytes = 1000000
711 rs, ws = yield self.make_iostream_pair()
712
713 produce_hash = hashlib.sha1()
714 consume_hash = hashlib.sha1()
715
716 @gen.coroutine
717 def produce():
718 remaining = nbytes
719 while remaining > 0:
720 size = r.randint(1, min(1000, remaining))
721 data = os.urandom(size)
722 produce_hash.update(data)
723 yield ws.write(data)
724 remaining -= size
725 assert remaining == 0
726
727 @gen.coroutine
728 def consume():
729 remaining = nbytes
730 while remaining > 0:
731 if r.random() > 0.5:
732 # read_bytes()
733 size = r.randint(1, min(1000, remaining))
734 data = yield rs.read_bytes(size)
735 consume_hash.update(data)
736 remaining -= size
737 else:
738 # read_into()
739 size = r.randint(1, min(1000, remaining))
740 buf = bytearray(size)
741 n = yield rs.read_into(buf)
742 assert n == size
743 consume_hash.update(buf)
744 remaining -= size
745 assert remaining == 0
746
747 try:
748 yield [produce(), consume()]
749 assert produce_hash.hexdigest() == consume_hash.hexdigest()
750 finally:
751 ws.close()
752 rs.close()
753
754
755@abstract_base_test
756class TestIOStreamMixin(TestReadWriteMixin):
757 def _make_server_iostream(self, connection, **kwargs):
758 raise NotImplementedError()
759
760 def _make_client_iostream(self, connection, **kwargs):
761 raise NotImplementedError()
762
763 @gen.coroutine
764 def make_iostream_pair(self, **kwargs):
765 listener, port = bind_unused_port()
766 server_stream_fut = Future() # type: Future[IOStream]
767
768 def accept_callback(connection, address):
769 server_stream_fut.set_result(
770 self._make_server_iostream(connection, **kwargs)
771 )
772
773 netutil.add_accept_handler(listener, accept_callback)
774 client_stream = self._make_client_iostream(socket.socket(), **kwargs)
775 connect_fut = client_stream.connect(("127.0.0.1", port))
776 server_stream, client_stream = yield [server_stream_fut, connect_fut]
777 self.io_loop.remove_handler(listener.fileno())
778 listener.close()
779 raise gen.Return((server_stream, client_stream))
780
781 @gen_test
782 def test_connection_refused(self):
783 # When a connection is refused, the connect callback should not
784 # be run. (The kqueue IOLoop used to behave differently from the
785 # epoll IOLoop in this respect)
786 cleanup_func, port = refusing_port()
787 self.addCleanup(cleanup_func)
788 stream = IOStream(socket.socket())
789
790 stream.set_close_callback(self.stop)
791 # log messages vary by platform and ioloop implementation
792 with ExpectLog(gen_log, ".*", required=False):
793 with self.assertRaises(StreamClosedError):
794 yield stream.connect(("127.0.0.1", port))
795
796 self.assertTrue(isinstance(stream.error, ConnectionRefusedError), stream.error)
797
798 @gen_test
799 def test_gaierror(self):
800 # Test that IOStream sets its exc_info on getaddrinfo error.
801 # It's difficult to reliably trigger a getaddrinfo error;
802 # some resolvers own't even return errors for malformed names,
803 # so we mock it instead. If IOStream changes to call a Resolver
804 # before sock.connect, the mock target will need to change too.
805 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
806 stream = IOStream(s)
807 stream.set_close_callback(self.stop)
808 with mock.patch(
809 "socket.socket.connect", side_effect=socket.gaierror(errno.EIO, "boom")
810 ):
811 with self.assertRaises(StreamClosedError):
812 yield stream.connect(("localhost", 80))
813 self.assertTrue(isinstance(stream.error, socket.gaierror))
814
815 @gen_test
816 def test_read_until_close_with_error(self):
817 server, client = yield self.make_iostream_pair()
818 try:
819 with mock.patch(
820 "tornado.iostream.BaseIOStream._try_inline_read",
821 side_effect=IOError("boom"),
822 ):
823 with self.assertRaisesRegex(IOError, "boom"):
824 client.read_until_close()
825 finally:
826 server.close()
827 client.close()
828
829 @skipIfNonUnix
830 @gen_test
831 def test_inline_read_error(self):
832 # An error on an inline read is raised without logging (on the
833 # assumption that it will eventually be noticed or logged further
834 # up the stack).
835 #
836 # This test is posix-only because windows os.close() doesn't work
837 # on socket FDs, but we can't close the socket object normally
838 # because we won't get the error we want if the socket knows
839 # it's closed.
840 #
841 # This test is also disabled when the
842 # AddThreadSelectorEventLoop is used, because a race between
843 # this thread closing the socket and the selector thread
844 # calling the select system call can make this test flaky.
845 # This event loop implementation is normally only used on
846 # windows, making this check redundant with skipIfNonUnix, but
847 # we sometimes enable it on other platforms for testing.
848 io_loop = IOLoop.current()
849 if isinstance(
850 io_loop.selector_loop, # type: ignore[attr-defined]
851 AddThreadSelectorEventLoop,
852 ):
853 self.skipTest("AddThreadSelectorEventLoop not supported")
854 server, client = yield self.make_iostream_pair()
855 try:
856 os.close(server.socket.fileno())
857 with self.assertRaises(socket.error):
858 server.read_bytes(1)
859 finally:
860 server.close()
861 client.close()
862
863 @gen_test
864 def test_async_read_error_logging(self):
865 # Socket errors on asynchronous reads should be logged (but only
866 # once).
867 server, client = yield self.make_iostream_pair()
868 closed = Event()
869 server.set_close_callback(closed.set)
870 try:
871 # Start a read that will be fulfilled asynchronously.
872 server.read_bytes(1)
873 client.write(b"a")
874 # Stub out read_from_fd to make it fail.
875
876 def fake_read_from_fd():
877 os.close(server.socket.fileno())
878 server.__class__.read_from_fd(server)
879
880 server.read_from_fd = fake_read_from_fd
881 # This log message is from _handle_read (not read_from_fd).
882 with ExpectLog(gen_log, "error on read"):
883 yield closed.wait()
884 finally:
885 server.close()
886 client.close()
887
888 @gen_test
889 def test_future_write(self):
890 """
891 Test that write() Futures are never orphaned.
892 """
893 # Run concurrent writers that will write enough bytes so as to
894 # clog the socket buffer and accumulate bytes in our write buffer.
895 m, n = 5000, 1000
896 nproducers = 10
897 total_bytes = m * n * nproducers
898 server, client = yield self.make_iostream_pair(max_buffer_size=total_bytes)
899
900 @gen.coroutine
901 def produce():
902 data = b"x" * m
903 for i in range(n):
904 yield server.write(data)
905
906 @gen.coroutine
907 def consume():
908 nread = 0
909 while nread < total_bytes:
910 res = yield client.read_bytes(m)
911 nread += len(res)
912
913 try:
914 yield [produce() for i in range(nproducers)] + [consume()]
915 finally:
916 server.close()
917 client.close()
918
919
920class TestIOStreamWebHTTP(AsyncHTTPTestCase, TestIOStreamWebMixin):
921 def _make_client_iostream(self):
922 return IOStream(socket.socket())
923
924 def get_app(self):
925 return self.mixin_get_app()
926
927
928class TestIOStreamWebHTTPS(AsyncHTTPSTestCase, TestIOStreamWebMixin):
929 def _make_client_iostream(self):
930 return SSLIOStream(socket.socket(), ssl_options=dict(cert_reqs=ssl.CERT_NONE))
931
932 def get_app(self):
933 return self.mixin_get_app()
934
935
936class TestIOStream(TestIOStreamMixin):
937 def _make_server_iostream(self, connection, **kwargs):
938 return IOStream(connection, **kwargs)
939
940 def _make_client_iostream(self, connection, **kwargs):
941 return IOStream(connection, **kwargs)
942
943
944class TestIOStreamSSL(TestIOStreamMixin):
945 def _make_server_iostream(self, connection, **kwargs):
946 ssl_ctx = ssl_options_to_context(_server_ssl_options(), server_side=True)
947 connection = ssl_ctx.wrap_socket(
948 connection,
949 server_side=True,
950 do_handshake_on_connect=False,
951 )
952 return SSLIOStream(connection, **kwargs)
953
954 def _make_client_iostream(self, connection, **kwargs):
955 return SSLIOStream(
956 connection, ssl_options=dict(cert_reqs=ssl.CERT_NONE), **kwargs
957 )
958
959
960# This will run some tests that are basically redundant but it's the
961# simplest way to make sure that it works to pass an SSLContext
962# instead of an ssl_options dict to the SSLIOStream constructor.
963class TestIOStreamSSLContext(TestIOStreamMixin):
964 def _make_server_iostream(self, connection, **kwargs):
965 context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
966 context.load_cert_chain(
967 os.path.join(os.path.dirname(__file__), "test.crt"),
968 os.path.join(os.path.dirname(__file__), "test.key"),
969 )
970 connection = ssl_wrap_socket(
971 connection, context, server_side=True, do_handshake_on_connect=False
972 )
973 return SSLIOStream(connection, **kwargs)
974
975 def _make_client_iostream(self, connection, **kwargs):
976 context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
977 context.check_hostname = False
978 context.verify_mode = ssl.CERT_NONE
979 return SSLIOStream(connection, ssl_options=context, **kwargs)
980
981
982class TestIOStreamStartTLS(AsyncTestCase):
983 def setUp(self):
984 try:
985 super().setUp()
986 self.listener, self.port = bind_unused_port()
987 self.server_stream = None
988 self.server_accepted = Future() # type: Future[None]
989 netutil.add_accept_handler(self.listener, self.accept)
990 self.client_stream = IOStream(
991 socket.socket()
992 ) # type: typing.Optional[IOStream]
993 self.io_loop.add_future(
994 self.client_stream.connect(("127.0.0.1", self.port)), self.stop
995 )
996 self.wait()
997 self.io_loop.add_future(self.server_accepted, self.stop)
998 self.wait()
999 except Exception as e:
1000 print(e)
1001 raise
1002
1003 def tearDown(self):
1004 if self.server_stream is not None:
1005 self.server_stream.close()
1006 if self.client_stream is not None:
1007 self.client_stream.close()
1008 self.io_loop.remove_handler(self.listener.fileno())
1009 self.listener.close()
1010 super().tearDown()
1011
1012 def accept(self, connection, address):
1013 if self.server_stream is not None:
1014 self.fail("should only get one connection")
1015 self.server_stream = IOStream(connection)
1016 self.server_accepted.set_result(None)
1017
1018 @gen.coroutine
1019 def client_send_line(self, line):
1020 assert self.client_stream is not None
1021 self.client_stream.write(line)
1022 assert self.server_stream is not None
1023 recv_line = yield self.server_stream.read_until(b"\r\n")
1024 self.assertEqual(line, recv_line)
1025
1026 @gen.coroutine
1027 def server_send_line(self, line):
1028 assert self.server_stream is not None
1029 self.server_stream.write(line)
1030 assert self.client_stream is not None
1031 recv_line = yield self.client_stream.read_until(b"\r\n")
1032 self.assertEqual(line, recv_line)
1033
1034 def client_start_tls(self, ssl_options=None, server_hostname=None):
1035 assert self.client_stream is not None
1036 client_stream = self.client_stream
1037 self.client_stream = None
1038 return client_stream.start_tls(False, ssl_options, server_hostname)
1039
1040 def server_start_tls(self, ssl_options=None):
1041 assert self.server_stream is not None
1042 server_stream = self.server_stream
1043 self.server_stream = None
1044 return server_stream.start_tls(True, ssl_options)
1045
1046 @gen_test
1047 def test_start_tls_smtp(self):
1048 # This flow is simplified from RFC 3207 section 5.
1049 # We don't really need all of this, but it helps to make sure
1050 # that after realistic back-and-forth traffic the buffers end up
1051 # in a sane state.
1052 yield self.server_send_line(b"220 mail.example.com ready\r\n")
1053 yield self.client_send_line(b"EHLO mail.example.com\r\n")
1054 yield self.server_send_line(b"250-mail.example.com welcome\r\n")
1055 yield self.server_send_line(b"250 STARTTLS\r\n")
1056 yield self.client_send_line(b"STARTTLS\r\n")
1057 yield self.server_send_line(b"220 Go ahead\r\n")
1058 client_future = self.client_start_tls(dict(cert_reqs=ssl.CERT_NONE))
1059 server_future = self.server_start_tls(_server_ssl_options())
1060 self.client_stream = yield client_future
1061 self.server_stream = yield server_future
1062 self.assertTrue(isinstance(self.client_stream, SSLIOStream))
1063 self.assertTrue(isinstance(self.server_stream, SSLIOStream))
1064 yield self.client_send_line(b"EHLO mail.example.com\r\n")
1065 yield self.server_send_line(b"250 mail.example.com welcome\r\n")
1066
1067 @gen_test
1068 def test_handshake_fail(self):
1069 server_future = self.server_start_tls(_server_ssl_options())
1070 # Certificates are verified with the default configuration.
1071 with ExpectLog(gen_log, "SSL Error"):
1072 client_future = self.client_start_tls(server_hostname="localhost")
1073 with self.assertRaises(ssl.SSLError):
1074 yield client_future
1075 with self.assertRaises((ssl.SSLError, socket.error)):
1076 yield server_future
1077
1078 @gen_test
1079 def test_check_hostname(self):
1080 # Test that server_hostname parameter to start_tls is being used.
1081 server_future = self.server_start_tls(_server_ssl_options())
1082 with ExpectLog(gen_log, "SSL Error"):
1083 client_future = self.client_start_tls(
1084 ssl.create_default_context(), server_hostname="127.0.0.1"
1085 )
1086 with self.assertRaises(ssl.SSLError):
1087 # The client fails to connect with an SSL error.
1088 yield client_future
1089 with self.assertRaises(Exception):
1090 # The server fails to connect, but the exact error is unspecified.
1091 yield server_future
1092
1093 @gen_test
1094 def test_typed_memoryview(self):
1095 # Test support of memoryviews with an item size greater than 1 byte.
1096 buf = memoryview(bytes(80)).cast("L")
1097 assert self.server_stream is not None
1098 yield self.server_stream.write(buf)
1099 assert self.client_stream is not None
1100 # This will timeout if the calculation of the buffer size is incorrect
1101 recv = yield self.client_stream.read_bytes(buf.nbytes)
1102 self.assertEqual(bytes(recv), bytes(buf))
1103
1104
1105class WaitForHandshakeTest(AsyncTestCase):
1106 @gen.coroutine
1107 def connect_to_server(self, server_cls):
1108 server = client = None
1109 try:
1110 sock, port = bind_unused_port()
1111 server = server_cls(ssl_options=_server_ssl_options())
1112 server.add_socket(sock)
1113
1114 ssl_ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
1115 ssl_ctx.check_hostname = False
1116 ssl_ctx.verify_mode = ssl.CERT_NONE
1117 # These tests fail with ConnectionAbortedErrors with TLS
1118 # 1.3 on windows python 3.7.4 (which includes an upgrade
1119 # to openssl 1.1.c. Other platforms might be affected with
1120 # newer openssl too). Disable it until we figure out
1121 # what's up.
1122 # Update 2021-12-28: Still happening with Python 3.10 on
1123 # Windows. OP_NO_TLSv1_3 now raises a DeprecationWarning.
1124 with ignore_deprecation():
1125 ssl_ctx.options |= getattr(ssl, "OP_NO_TLSv1_3", 0)
1126 client = SSLIOStream(socket.socket(), ssl_options=ssl_ctx)
1127 yield client.connect(("127.0.0.1", port))
1128 self.assertIsNotNone(client.socket.cipher())
1129 finally:
1130 if server is not None:
1131 server.stop()
1132 if client is not None:
1133 client.close()
1134
1135 @gen_test
1136 def test_wait_for_handshake_future(self):
1137 test = self
1138 handshake_future = Future() # type: Future[None]
1139
1140 class TestServer(TCPServer):
1141 def handle_stream(self, stream, address):
1142 test.assertIsNone(stream.socket.cipher())
1143 test.io_loop.spawn_callback(self.handle_connection, stream)
1144
1145 @gen.coroutine
1146 def handle_connection(self, stream):
1147 yield stream.wait_for_handshake()
1148 handshake_future.set_result(None)
1149
1150 yield self.connect_to_server(TestServer)
1151 yield handshake_future
1152
1153 @gen_test
1154 def test_wait_for_handshake_already_waiting_error(self):
1155 test = self
1156 handshake_future = Future() # type: Future[None]
1157
1158 class TestServer(TCPServer):
1159 @gen.coroutine
1160 def handle_stream(self, stream, address):
1161 fut = stream.wait_for_handshake()
1162 test.assertRaises(RuntimeError, stream.wait_for_handshake)
1163 yield fut
1164
1165 handshake_future.set_result(None)
1166
1167 yield self.connect_to_server(TestServer)
1168 yield handshake_future
1169
1170 @gen_test
1171 def test_wait_for_handshake_already_connected(self):
1172 handshake_future = Future() # type: Future[None]
1173
1174 class TestServer(TCPServer):
1175 @gen.coroutine
1176 def handle_stream(self, stream, address):
1177 yield stream.wait_for_handshake()
1178 yield stream.wait_for_handshake()
1179 handshake_future.set_result(None)
1180
1181 yield self.connect_to_server(TestServer)
1182 yield handshake_future
1183
1184
1185class TestIOStreamCheckHostname(AsyncTestCase):
1186 # This test ensures that hostname checks are working correctly after
1187 # #3337 revealed that we have no test coverage in this area, and we
1188 # removed a manual hostname check that was needed only for very old
1189 # versions of python.
1190 def setUp(self):
1191 super().setUp()
1192 self.listener, self.port = bind_unused_port()
1193
1194 def accept_callback(connection, address):
1195 ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
1196 ssl_ctx.load_cert_chain(
1197 os.path.join(os.path.dirname(__file__), "test.crt"),
1198 os.path.join(os.path.dirname(__file__), "test.key"),
1199 )
1200 connection = ssl_ctx.wrap_socket(