self.loop.sock_connect.return_value = ()
self.loop._make_ssl_transport = unittest.mock.Mock()
+ class _SelectorTransportMock:
+ _sock = None
+
+ def close(self):
+ self._sock.close()
+
def mock_make_ssl_transport(sock, protocol, sslcontext, waiter,
**kwds):
waiter.set_result(None)
+ transport = _SelectorTransportMock()
+ transport._sock = sock
+ return transport
self.loop._make_ssl_transport.side_effect = mock_make_ssl_transport
ANY = unittest.mock.ANY
# First try the default server_hostname.
self.loop._make_ssl_transport.reset_mock()
coro = self.loop.create_connection(MyProto, 'python.org', 80, ssl=True)
- self.loop.run_until_complete(coro)
+ transport, _ = self.loop.run_until_complete(coro)
+ transport.close()
self.loop._make_ssl_transport.assert_called_with(
ANY, ANY, ANY, ANY,
server_side=False,
self.loop._make_ssl_transport.reset_mock()
coro = self.loop.create_connection(MyProto, 'python.org', 80, ssl=True,
server_hostname='perl.com')
- self.loop.run_until_complete(coro)
+ transport, _ = self.loop.run_until_complete(coro)
+ transport.close()
self.loop._make_ssl_transport.assert_called_with(
ANY, ANY, ANY, ANY,
server_side=False,
self.loop._make_ssl_transport.reset_mock()
coro = self.loop.create_connection(MyProto, 'python.org', 80, ssl=True,
server_hostname='')
- self.loop.run_until_complete(coro)
+ transport, _ = self.loop.run_until_complete(coro)
+ transport.close()
self.loop._make_ssl_transport.assert_called_with(ANY, ANY, ANY, ANY,
server_side=False,
server_hostname='')
self.assertRaises(ValueError, self.loop.run_until_complete, coro)
coro = self.loop.create_connection(MyProto, None, 80, ssl=True)
self.assertRaises(ValueError, self.loop.run_until_complete, coro)
+ sock = socket.socket()
coro = self.loop.create_connection(MyProto, None, None,
- ssl=True, sock=socket.socket())
+ ssl=True, sock=sock)
+ self.addCleanup(sock.close)
self.assertRaises(ValueError, self.loop.run_until_complete, coro)
def test_create_server_empty_host(self):