codekingpro/portable-devtools
114k
1#
2# Copyright 2014 Facebook
3#
4# Licensed under the Apache License, Version 2.0 (the "License"); you may
5# not use this file except in compliance with the License. You may obtain
6# a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13# License for the specific language governing permissions and limitations
14# under the License.
15from contextlib import closing
16import getpass
17import socket
18import unittest
19
20from tornado.concurrent import Future
21from tornado.netutil import bind_sockets, Resolver
22from tornado.queues import Queue
23from tornado.tcpclient import TCPClient, _Connector
24from tornado.tcpserver import TCPServer
25from tornado.testing import AsyncTestCase, gen_test
26from tornado.test.util import skipIfNoIPv6, refusing_port, skipIfNonUnix
27from tornado.gen import TimeoutError
28
29import typing
30
31if typing.TYPE_CHECKING:
32 from tornado.iostream import IOStream # noqa: F401
33 from typing import List, Dict, Tuple # noqa: F401
34
35# Fake address families for testing. Used in place of AF_INET
36# and AF_INET6 because some installations do not have AF_INET6.
37AF1, AF2 = 1, 2
38
39
40class TestTCPServer(TCPServer):
41 def __init__(self, family):
42 super().__init__()
43 self.streams = [] # type: List[IOStream]
44 self.queue = Queue() # type: Queue[IOStream]
45 sockets = bind_sockets(0, "localhost", family)
46 self.add_sockets(sockets)
47 self.port = sockets[0].getsockname()[1]
48
49 def handle_stream(self, stream, address):
50 self.streams.append(stream)
51 self.queue.put(stream)
52
53 def stop(self):
54 super().stop()
55 for stream in self.streams:
56 stream.close()
57
58
59class TCPClientTest(AsyncTestCase):
60 def setUp(self):
61 super().setUp()
62 self.server = None
63 self.client = TCPClient()
64
65 def start_server(self, family):
66 self.server = TestTCPServer(family)
67 return self.server.port
68
69 def stop_server(self):
70 if self.server is not None:
71 self.server.stop()
72 self.server = None
73
74 def tearDown(self):
75 self.client.close()
76 self.stop_server()
77 super().tearDown()
78
79 def skipIfLocalhostV4(self):
80 # The port used here doesn't matter, but some systems require it
81 # to be non-zero if we do not also pass AI_PASSIVE.
82 addrinfo = self.io_loop.run_sync(lambda: Resolver().resolve("localhost", 80))
83 families = {addr[0] for addr in addrinfo}
84 if socket.AF_INET6 not in families:
85 self.skipTest("localhost does not resolve to ipv6")
86
87 @gen_test
88 def do_test_connect(self, family, host, source_ip=None, source_port=None):
89 port = self.start_server(family)
90 stream = yield self.client.connect(
91 host,
92 port,
93 source_ip=source_ip,
94 source_port=source_port,
95 af=family,
96 )
97 assert self.server is not None
98 server_stream = yield self.server.queue.get()
99 with closing(stream):
100 stream.write(b"hello")
101 data = yield server_stream.read_bytes(5)
102 self.assertEqual(data, b"hello")
103
104 def test_connect_ipv4_ipv4(self):
105 self.do_test_connect(socket.AF_INET, "127.0.0.1")
106
107 def test_connect_ipv4_dual(self):
108 self.do_test_connect(socket.AF_INET, "localhost")
109
110 @skipIfNoIPv6
111 def test_connect_ipv6_ipv6(self):
112 self.skipIfLocalhostV4()
113 self.do_test_connect(socket.AF_INET6, "::1")
114
115 @skipIfNoIPv6
116 def test_connect_ipv6_dual(self):
117 self.skipIfLocalhostV4()
118 self.do_test_connect(socket.AF_INET6, "localhost")
119
120 def test_connect_unspec_ipv4(self):
121 self.do_test_connect(socket.AF_UNSPEC, "127.0.0.1")
122
123 @skipIfNoIPv6
124 def test_connect_unspec_ipv6(self):
125 self.skipIfLocalhostV4()
126 self.do_test_connect(socket.AF_UNSPEC, "::1")
127
128 def test_connect_unspec_dual(self):
129 self.do_test_connect(socket.AF_UNSPEC, "localhost")
130
131 @gen_test
132 def test_refused_ipv4(self):
133 cleanup_func, port = refusing_port()
134 self.addCleanup(cleanup_func)
135 with self.assertRaises(IOError):
136 yield self.client.connect("127.0.0.1", port)
137
138 def test_source_ip_fail(self):
139 """Fail when trying to use the source IP Address '8.8.8.8'."""
140 self.assertRaises(
141 socket.error,
142 self.do_test_connect,
143 socket.AF_INET,
144 "127.0.0.1",
145 source_ip="8.8.8.8",
146 )
147
148 def test_source_ip_success(self):
149 """Success when trying to use the source IP Address '127.0.0.1'."""
150 self.do_test_connect(socket.AF_INET, "127.0.0.1", source_ip="127.0.0.1")
151
152 @skipIfNonUnix
153 def test_source_port_fail(self):
154 """Fail when trying to use source port 1."""
155 if getpass.getuser() == "root":
156 # Root can use any port so we can't easily force this to fail.
157 # This is mainly relevant for docker.
158 self.skipTest("running as root")
159 self.assertRaises(
160 socket.error,
161 self.do_test_connect,
162 socket.AF_INET,
163 "127.0.0.1",
164 source_port=1,
165 )
166
167 @gen_test
168 def test_connect_timeout(self):
169 timeout = 0.05
170
171 class TimeoutResolver(Resolver):
172 def resolve(self, *args, **kwargs):
173 return Future() # never completes
174
175 with self.assertRaises(TimeoutError):
176 yield TCPClient(resolver=TimeoutResolver()).connect(
177 "1.2.3.4", 12345, timeout=timeout
178 )
179
180
181class TestConnectorSplit(unittest.TestCase):
182 def test_one_family(self):
183 # These addresses aren't in the right format, but split doesn't care.
184 primary, secondary = _Connector.split([(AF1, "a"), (AF1, "b")])
185 self.assertEqual(primary, [(AF1, "a"), (AF1, "b")])
186 self.assertEqual(secondary, [])
187
188 def test_mixed(self):
189 primary, secondary = _Connector.split(
190 [(AF1, "a"), (AF2, "b"), (AF1, "c"), (AF2, "d")]
191 )
192 self.assertEqual(primary, [(AF1, "a"), (AF1, "c")])
193 self.assertEqual(secondary, [(AF2, "b"), (AF2, "d")])
194
195
196class ConnectorTest(AsyncTestCase):
197 class FakeStream:
198 def __init__(self):
199 self.closed = False
200
201 def close(self):
202 self.closed = True
203
204 def setUp(self):
205 super().setUp()
206 self.connect_futures = (
207 {}
208 ) # type: Dict[Tuple[int, typing.Any], Future[ConnectorTest.FakeStream]]
209 self.streams = {} # type: Dict[typing.Any, ConnectorTest.FakeStream]
210 self.addrinfo = [(AF1, "a"), (AF1, "b"), (AF2, "c"), (AF2, "d")]
211
212 def tearDown(self):
213 # Unless explicitly checked (and popped) in the test, we shouldn't
214 # be closing any streams
215 for stream in self.streams.values():
216 self.assertFalse(stream.closed)
217 super().tearDown()
218
219 def create_stream(self, af, addr):
220 stream = ConnectorTest.FakeStream()
221 self.streams[addr] = stream
222 future = Future() # type: Future[ConnectorTest.FakeStream]
223 self.connect_futures[(af, addr)] = future
224 return stream, future
225
226 def assert_pending(self, *keys):
227 self.assertEqual(sorted(self.connect_futures.keys()), sorted(keys))
228
229 def resolve_connect(self, af, addr, success):
230 future = self.connect_futures.pop((af, addr))
231 if success:
232 future.set_result(self.streams[addr])
233 else:
234 self.streams.pop(addr)
235 future.set_exception(IOError())
236 # Run the loop to allow callbacks to be run.
237 self.io_loop.add_callback(self.stop)
238 self.wait()
239
240 def assert_connector_streams_closed(self, conn):
241 for stream in conn.streams:
242 self.assertTrue(stream.closed)
243
244 def start_connect(self, addrinfo):
245 conn = _Connector(addrinfo, self.create_stream)
246 # Give it a huge timeout; we'll trigger timeouts manually.
247 future = conn.start(3600, connect_timeout=self.io_loop.time() + 3600)
248 return conn, future
249
250 def test_immediate_success(self):
251 conn, future = self.start_connect(self.addrinfo)
252 self.assertEqual(list(self.connect_futures.keys()), [(AF1, "a")])
253 self.resolve_connect(AF1, "a", True)
254 self.assertEqual(future.result(), (AF1, "a", self.streams["a"]))
255
256 def test_immediate_failure(self):
257 # Fail with just one address.
258 conn, future = self.start_connect([(AF1, "a")])
259 self.assert_pending((AF1, "a"))
260 self.resolve_connect(AF1, "a", False)
261 self.assertRaises(IOError, future.result)
262
263 def test_one_family_second_try(self):
264 conn, future = self.start_connect([(AF1, "a"), (AF1, "b")])
265 self.assert_pending((AF1, "a"))
266 self.resolve_connect(AF1, "a", False)
267 self.assert_pending((AF1, "b"))
268 self.resolve_connect(AF1, "b", True)
269 self.assertEqual(future.result(), (AF1, "b", self.streams["b"]))
270
271 def test_one_family_second_try_failure(self):
272 conn, future = self.start_connect([(AF1, "a"), (AF1, "b")])
273 self.assert_pending((AF1, "a"))
274 self.resolve_connect(AF1, "a", False)
275 self.assert_pending((AF1, "b"))
276 self.resolve_connect(AF1, "b", False)
277 self.assertRaises(IOError, future.result)
278
279 def test_one_family_second_try_timeout(self):
280 conn, future = self.start_connect([(AF1, "a"), (AF1, "b")])
281 self.assert_pending((AF1, "a"))
282 # trigger the timeout while the first lookup is pending;
283 # nothing happens.
284 conn.on_timeout()
285 self.assert_pending((AF1, "a"))
286 self.resolve_connect(AF1, "a", False)
287 self.assert_pending((AF1, "b"))
288 self.resolve_connect(AF1, "b", True)
289 self.assertEqual(future.result(), (AF1, "b", self.streams["b"]))
290
291 def test_two_families_immediate_failure(self):
292 conn, future = self.start_connect(self.addrinfo)
293 self.assert_pending((AF1, "a"))
294 self.resolve_connect(AF1, "a", False)
295 self.assert_pending((AF1, "b"), (AF2, "c"))
296 self.resolve_connect(AF1, "b", False)
297 self.resolve_connect(AF2, "c", True)
298 self.assertEqual(future.result(), (AF2, "c", self.streams["c"]))
299
300 def test_two_families_timeout(self):
301 conn, future = self.start_connect(self.addrinfo)
302 self.assert_pending((AF1, "a"))
303 conn.on_timeout()
304 self.assert_pending((AF1, "a"), (AF2, "c"))
305 self.resolve_connect(AF2, "c", True)
306 self.assertEqual(future.result(), (AF2, "c", self.streams["c"]))
307 # resolving 'a' after the connection has completed doesn't start 'b'
308 self.resolve_connect(AF1, "a", False)
309 self.assert_pending()
310
311 def test_success_after_timeout(self):
312 conn, future = self.start_connect(self.addrinfo)
313 self.assert_pending((AF1, "a"))
314 conn.on_timeout()
315 self.assert_pending((AF1, "a"), (AF2, "c"))
316 self.resolve_connect(AF1, "a", True)
317 self.assertEqual(future.result(), (AF1, "a", self.streams["a"]))
318 # resolving 'c' after completion closes the connection.
319 self.resolve_connect(AF2, "c", True)
320 self.assertTrue(self.streams.pop("c").closed)
321
322 def test_all_fail(self):
323 conn, future = self.start_connect(self.addrinfo)
324 self.assert_pending((AF1, "a"))
325 conn.on_timeout()
326 self.assert_pending((AF1, "a"), (AF2, "c"))
327 self.resolve_connect(AF2, "c", False)
328 self.assert_pending((AF1, "a"), (AF2, "d"))
329 self.resolve_connect(AF2, "d", False)
330 # one queue is now empty
331 self.assert_pending((AF1, "a"))
332 self.resolve_connect(AF1, "a", False)
333 self.assert_pending((AF1, "b"))
334 self.assertFalse(future.done())
335 self.resolve_connect(AF1, "b", False)
336 self.assertRaises(IOError, future.result)
337
338 def test_one_family_timeout_after_connect_timeout(self):
339 conn, future = self.start_connect([(AF1, "a"), (AF1, "b")])
340 self.assert_pending((AF1, "a"))
341 conn.on_connect_timeout()
342 # the connector will close all streams on connect timeout, we
343 # should explicitly pop the connect_future.
344 self.connect_futures.pop((AF1, "a"))
345 self.assertTrue(self.streams.pop("a").closed)
346 conn.on_timeout()
347 # if the future is set with TimeoutError, we will not iterate next
348 # possible address.
349 self.assert_pending()
350 self.assertEqual(len(conn.streams), 1)
351 self.assert_connector_streams_closed(conn)
352 self.assertRaises(TimeoutError, future.result)
353
354 def test_one_family_success_before_connect_timeout(self):
355 conn, future = self.start_connect([(AF1, "a"), (AF1, "b")])
356 self.assert_pending((AF1, "a"))
357 self.resolve_connect(AF1, "a", True)
358 conn.on_connect_timeout()
359 self.assert_pending()
360 self.assertFalse(self.streams["a"].closed)
361 # success stream will be pop
362 self.assertEqual(len(conn.streams), 0)
363 # streams in connector should be closed after connect timeout
364 self.assert_connector_streams_closed(conn)
365 self.assertEqual(future.result(), (AF1, "a", self.streams["a"]))
366
367 def test_one_family_second_try_after_connect_timeout(self):
368 conn, future = self.start_connect([(AF1, "a"), (AF1, "b")])
369 self.assert_pending((AF1, "a"))
370 self.resolve_connect(AF1, "a", False)
371 self.assert_pending((AF1, "b"))
372 conn.on_connect_timeout()
373 self.connect_futures.pop((AF1, "b"))
374 self.assertTrue(self.streams.pop("b").closed)
375 self.assert_pending()
376 self.assertEqual(len(conn.streams), 2)
377 self.assert_connector_streams_closed(conn)
378 self.assertRaises(TimeoutError, future.result)
379
380 def test_one_family_second_try_failure_before_connect_timeout(self):
381 conn, future = self.start_connect([(AF1, "a"), (AF1, "b")])
382 self.assert_pending((AF1, "a"))
383 self.resolve_connect(AF1, "a", False)
384 self.assert_pending((AF1, "b"))
385 self.resolve_connect(AF1, "b", False)
386 conn.on_connect_timeout()
387 self.assert_pending()
388 self.assertEqual(len(conn.streams), 2)
389 self.assert_connector_streams_closed(conn)
390 self.assertRaises(IOError, future.result)
391
392 def test_two_family_timeout_before_connect_timeout(self):
393 conn, future = self.start_connect(self.addrinfo)
394 self.assert_pending((AF1, "a"))
395 conn.on_timeout()
396 self.assert_pending((AF1, "a"), (AF2, "c"))
397 conn.on_connect_timeout()
398 self.connect_futures.pop((AF1, "a"))
399 self.assertTrue(self.streams.pop("a").closed)
400 self.connect_futures.pop((AF2, "c"))
401 self.assertTrue(self.streams.pop("c").closed)
402 self.assert_pending()
403 self.assertEqual(len(conn.streams), 2)
404 self.assert_connector_streams_closed(conn)
405 self.assertRaises(TimeoutError, future.result)
406
407 def test_two_family_success_after_timeout(self):
408 conn, future = self.start_connect(self.addrinfo)
409 self.assert_pending((AF1, "a"))
410 conn.on_timeout()
411 self.assert_pending((AF1, "a"), (AF2, "c"))
412 self.resolve_connect(AF1, "a", True)
413 # if one of streams succeed, connector will close all other streams
414 self.connect_futures.pop((AF2, "c"))
415 self.assertTrue(self.streams.pop("c").closed)
416 self.assert_pending()
417 self.assertEqual(len(conn.streams), 1)
418 self.assert_connector_streams_closed(conn)
419 self.assertEqual(future.result(), (AF1, "a", self.streams["a"]))
420
421 def test_two_family_timeout_after_connect_timeout(self):
422 conn, future = self.start_connect(self.addrinfo)
423 self.assert_pending((AF1, "a"))
424 conn.on_connect_timeout()
425 self.connect_futures.pop((AF1, "a"))
426 self.assertTrue(self.streams.pop("a").closed)
427 self.assert_pending()
428 conn.on_timeout()
429 # if the future is set with TimeoutError, connector will not
430 # trigger secondary address.
431 self.assert_pending()
432 self.assertEqual(len(conn.streams), 1)
433 self.assert_connector_streams_closed(conn)
434 self.assertRaises(TimeoutError, future.result)
435