(self, family, host, source_ip=None, source_port=None)
| 89 | |
| 90 | @gen_test |
| 91 | def do_test_connect(self, family, host, source_ip=None, source_port=None): |
| 92 | port = self.start_server(family) |
| 93 | stream = yield self.client.connect( |
| 94 | host, port, source_ip=source_ip, source_port=source_port |
| 95 | ) |
| 96 | assert self.server is not None |
| 97 | server_stream = yield self.server.queue.get() |
| 98 | with closing(stream): |
| 99 | stream.write(b"hello") |
| 100 | data = yield server_stream.read_bytes(5) |
| 101 | self.assertEqual(data, b"hello") |
| 102 | |
| 103 | def test_connect_ipv4_ipv4(self): |
| 104 | self.do_test_connect(socket.AF_INET, "127.0.0.1") |
no test coverage detected