raise ValueError('You must set server_hostname '
'when using ssl without a host')
server_hostname = host
-
+
if host is not None or port is not None:
if sock is not None:
raise ValueError(
@tasks.coroutine
def start_serving_pipe(self, protocol_factory, address):
server = PipeServer(address)
+
def loop(f=None):
pipe = None
try:
pipe.close()
else:
f.add_done_callback(loop)
+
self.call_soon(loop)
return [server]
ov.WSARecv(conn.fileno(), nbytes, flags)
else:
ov.ReadFile(conn.fileno(), nbytes)
+
def finish(trans, key, ov):
try:
return ov.getresult()
raise ConnectionResetError(*exc.args)
else:
raise
+
return self._register(ov, conn, finish)
def send(self, conn, buf, flags=0):
ov.WSASend(conn.fileno(), buf, flags)
else:
ov.WriteFile(conn.fileno(), buf)
+
def finish(trans, key, ov):
try:
return ov.getresult()
raise ConnectionResetError(*exc.args)
else:
raise
+
return self._register(ov, conn, finish)
def accept(self, listener):
conn = self._get_accept_socket(listener.family)
ov = _overlapped.Overlapped(NULL)
ov.AcceptEx(listener.fileno(), conn.fileno())
+
def finish_accept(trans, key, ov):
ov.getresult()
# Use SO_UPDATE_ACCEPT_CONTEXT so getsockname() etc work.
_overlapped.SO_UPDATE_ACCEPT_CONTEXT, buf)
conn.settimeout(listener.gettimeout())
return conn, conn.getpeername()
+
return self._register(ov, listener, finish_accept)
def connect(self, conn, address):
raise
ov = _overlapped.Overlapped(NULL)
ov.ConnectEx(conn.fileno(), address)
+
def finish_connect(trans, key, ov):
ov.getresult()
# Use SO_UPDATE_CONNECT_CONTEXT so getsockname() etc work.
conn.setsockopt(socket.SOL_SOCKET,
_overlapped.SO_UPDATE_CONNECT_CONTEXT, 0)
return conn
+
return self._register(ov, conn, finish_connect)
def accept_pipe(self, pipe):
self._register_with_iocp(pipe)
ov = _overlapped.Overlapped(NULL)
ov.ConnectNamedPipe(pipe.fileno())
+
def finish(trans, key, ov):
ov.getresult()
return pipe
+
return self._register(ov, pipe, finish)
def connect_pipe(self, address):
ov = _overlapped.Overlapped(NULL)
ov.WaitNamedPipeAndConnect(address, self._iocp, ov.address)
+
def finish(err, handle, ov):
# err, handle were arguments passed to PostQueuedCompletionStatus()
# in a function run in a thread pool.
raise OSError(0, msg, None, err)
else:
return windows_utils.PipeHandle(handle)
+
return self._register(ov, None, finish, wait_for_post=True)
def wait_for_handle(self, handle, timeout=None):
self._proc = windows_utils.Popen(
args, shell=shell, stdin=stdin, stdout=stdout, stderr=stderr,
bufsize=bufsize, **kwargs)
+
def callback(f):
returncode = self._proc.poll()
self._process_exited(returncode)
+
f = self._loop._proactor.wait_for_handle(int(self._proc._handle))
f.add_done_callback(callback)
__all__ = ['socketpair', 'pipe', 'Popen', 'PIPE', 'PipeHandle']
-#
+
# Constants/globals
-#
+
BUFSIZE = 8192
PIPE = subprocess.PIPE
STDOUT = subprocess.STDOUT
_mmap_counter = itertools.count()
-#
+
# Replacement for socket.socketpair()
-#
+
def socketpair(family=socket.AF_INET, type=socket.SOCK_STREAM, proto=0):
"""A socket pair usable as a self-pipe, for Windows.
lsock.close()
return (ssock, csock)
-#
+
# Replacement for os.pipe() using handles instead of fds
-#
+
def pipe(*, duplex=False, overlapped=(True, True), bufsize=BUFSIZE):
"""Like os.pipe() but with overlapped support and using handles not fds."""
_winapi.CloseHandle(h2)
raise
-#
+
# Wrapper for a pipe handle
-#
+
class PipeHandle:
"""Wrapper for an overlapped pipe handle which is vaguely file-object like.
def __exit__(self, t, v, tb):
self.close()
-#
+
# Replacement for subprocess.Popen using overlapped pipe handles
-#
+
class Popen(subprocess.Popen):
"""Replacement for subprocess.Popen using overlapped pipe handles.
def test_create_connection_server_hostname_default(self):
self.loop.getaddrinfo = unittest.mock.Mock()
+
def mock_getaddrinfo(*args, **kwds):
f = futures.Future(loop=self.loop)
f.set_result([(socket.AF_INET, socket.SOCK_STREAM,
socket.SOL_TCP, '', ('1.2.3.4', 80))])
return f
+
self.loop.getaddrinfo.side_effect = mock_getaddrinfo
self.loop.sock_connect = unittest.mock.Mock()
self.loop.sock_connect.return_value = ()
self.loop._make_ssl_transport = unittest.mock.Mock()
- def mock_make_ssl_transport(sock, protocol, sslcontext, waiter, **kwds):
+
+ def mock_make_ssl_transport(sock, protocol, sslcontext, waiter,
+ **kwds):
waiter.set_result(None)
+
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)
- self.loop._make_ssl_transport.assert_called_with(ANY, ANY, ANY, ANY,
- server_side=False,
- server_hostname='python.org')
+ self.loop._make_ssl_transport.assert_called_with(
+ ANY, ANY, ANY, ANY,
+ server_side=False,
+ server_hostname='python.org')
# Next try an explicit server_hostname.
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)
- self.loop._make_ssl_transport.assert_called_with(ANY, ANY, ANY, ANY,
- server_side=False,
- server_hostname='perl.com')
+ self.loop._make_ssl_transport.assert_called_with(
+ ANY, ANY, ANY, ANY,
+ server_side=False,
+ server_hostname='perl.com')
# Finally try an explicit empty server_hostname.
self.loop._make_ssl_transport.reset_mock()
coro = self.loop.create_connection(MyProto, 'python.org', 80, ssl=True,
def test_create_connection_server_hostname_errors(self):
# When not using ssl, server_hostname must be None (but '' is OK).
- coro = self.loop.create_connection(MyProto, 'python.org', 80, server_hostname='')
+ coro = self.loop.create_connection(MyProto, 'python.org', 80,
+ server_hostname='')
self.assertRaises(ValueError, self.loop.run_until_complete, coro)
- coro = self.loop.create_connection(MyProto, 'python.org', 80, server_hostname='python.org')
+ coro = self.loop.create_connection(MyProto, 'python.org', 80,
+ server_hostname='python.org')
self.assertRaises(ValueError, self.loop.run_until_complete, coro)
# When using ssl, server_hostname may be None if host is non-empty.
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)
- coro = self.loop.create_connection(MyProto, None, None, ssl=True, sock=socket.socket())
+ coro = self.loop.create_connection(MyProto, None, None,
+ ssl=True, sock=socket.socket())
self.assertRaises(ValueError, self.loop.run_until_complete, coro)
def test_create_server_empty_host(self):
def create_event_loop(self):
return windows_events.SelectorEventLoop()
-
class ProactorEventLoopTests(EventLoopTestsMixin,
SubprocessTestsMixin,
unittest.TestCase):
stream_reader = streams.StreamReader(loop=self.loop)
protocol = streams.StreamReaderProtocol(stream_reader)
trans, proto = yield from self.loop.create_pipe_connection(
- lambda:protocol, ADDRESS)
+ lambda: protocol, ADDRESS)
self.assertIsInstance(trans, transports.Transport)
self.assertEqual(protocol, proto)
clients.append((stream_reader, trans))
projects / python / commitdiff