Add connect_accepted_socket.

Author: 1st1

tests/test_tcp.py

@@ -3,11 +3,47 @@
 import socket
 import unittest.mock
 import uvloop
+import ssl
 import sys
+import threading
 
 from uvloop import _testbase as tb
 
 
+class MyBaseProto(asyncio.Protocol):
+    connected = None
+    done = None
+
+    def __init__(self, loop=None):
+        self.transport = None
+        self.state = 'INITIAL'
+        self.nbytes = 0
+        if loop is not None:
+            self.connected = asyncio.Future(loop=loop)
+            self.done = asyncio.Future(loop=loop)
+
+    def connection_made(self, transport):
+        self.transport = transport
+        assert self.state == 'INITIAL', self.state
+        self.state = 'CONNECTED'
+        if self.connected:
+            self.connected.set_result(None)
+
+    def data_received(self, data):
+        assert self.state == 'CONNECTED', self.state
+        self.nbytes += len(data)
+
+    def eof_received(self):
+        assert self.state == 'CONNECTED', self.state
+        self.state = 'EOF'
+
+    def connection_lost(self, exc):
+        assert self.state in ('CONNECTED', 'EOF'), self.state
+        self.state = 'CLOSED'
+        if self.done:
+            self.done.set_result(None)
+
+
 class _TestTCP:
     def test_create_server_1(self):
         if self.is_asyncio_loop() and sys.version_info[:3] == (3, 5, 2):
@@ -699,6 +735,62 @@ async def runner():
         srv.close()
         self.loop.run_until_complete(srv.wait_closed())
 
+    def test_connect_accepted_socket(self, server_ssl=None, client_ssl=None):
+        loop = self.loop
+
+        class MyProto(MyBaseProto):
+
+            def connection_lost(self, exc):
+                super().connection_lost(exc)
+                loop.call_soon(loop.stop)
+
+            def data_received(self, data):
+                super().data_received(data)
+                self.transport.write(expected_response)
+
+        lsock = socket.socket()
+        lsock.bind(('127.0.0.1', 0))
+        lsock.listen(1)
+        addr = lsock.getsockname()
+
+        message = b'test data'
+        response = None
+        expected_response = b'roger'
+
+        def client():
+            nonlocal response
+            try:
+                csock = socket.socket()
+                if client_ssl is not None:
+                    csock = client_ssl.wrap_socket(csock)
+                csock.connect(addr)
+                csock.sendall(message)
+                response = csock.recv(99)
+                csock.close()
+            except Exception as exc:
+                print(
+                    "Failure in client thread in test_connect_accepted_socket",
+                    exc)
+
+        thread = threading.Thread(target=client, daemon=True)
+        thread.start()
+
+        conn, _ = lsock.accept()
+        proto = MyProto(loop=loop)
+        proto.loop = loop
+        loop.create_task(
+            loop.connect_accepted_socket(
+                (lambda: proto), conn, ssl=server_ssl))
+        loop.run_forever()
+        conn.close()
+        lsock.close()
+
+        thread.join(1)
+        self.assertFalse(thread.is_alive())
+        self.assertEqual(proto.state, 'CLOSED')
+        self.assertEqual(proto.nbytes, len(message))
+        self.assertEqual(response, expected_response)
+
 
 class Test_AIO_TCP(_TestTCP, tb.AIOTestCase):
     pass
@@ -864,7 +956,22 @@ def run(coro):
 
 
 class Test_UV_TCPSSL(_TestSSL, tb.UVTestCase):
-    pass
+
+    def test_ssl_connect_accepted_socket(self):
+        server_context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
+        server_context.load_cert_chain(self.ONLYCERT, self.ONLYKEY)
+        if hasattr(server_context, 'check_hostname'):
+            server_context.check_hostname = False
+        server_context.verify_mode = ssl.CERT_NONE
+
+        client_context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
+        if hasattr(server_context, 'check_hostname'):
+            client_context.check_hostname = False
+        client_context.verify_mode = ssl.CERT_NONE
+
+        Test_UV_TCP.test_connect_accepted_socket(
+            self, server_context, client_context)
+
 
 
 class Test_AIO_TCPSSL(_TestSSL, tb.AIOTestCase):

uvloop/handles/handle.pxd

@@ -39,3 +39,5 @@ cdef class UVSocketHandle(UVHandle):
     cdef _new_socket(self)
     cdef inline _get_socket(self)
     cdef inline _attach_fileobj(self, object file)
+
+    cdef _open(self, int sockfd)

uvloop/handles/handle.pyx

@@ -274,6 +274,9 @@ cdef class UVSocketHandle(UVHandle):
         finally:
             UVHandle._close(self)
 
+    cdef _open(self, int sockfd):
+        raise NotImplementedError
+
 
 cdef inline bint __ensure_handle_data(uv.uv_handle_t* handle,
                                       const char* handle_ctx):

uvloop/handles/pipe.pxd

@@ -1,7 +1,6 @@
 cdef class UnixServer(UVStreamServer):
 
     cdef bind(self, str path)
-    cdef open(self, int sockfd)
 
     @staticmethod
     cdef UnixServer new(Loop loop, object protocol_factory, Server server,
@@ -14,7 +13,6 @@ cdef class UnixTransport(UVStream):
     cdef UnixTransport new(Loop loop, object protocol, Server server,
                            object waiter)
 
-    cdef open(self, int sockfd)
     cdef connect(self, char* addr)
 
 
@@ -24,13 +22,9 @@ cdef class ReadUnixTransport(UVStream):
     cdef ReadUnixTransport new(Loop loop, object protocol, Server server,
                                object waiter)
 
-    cdef open(self, int sockfd)
-
 
 cdef class WriteUnixTransport(UVStream):
 
     @staticmethod
     cdef WriteUnixTransport new(Loop loop, object protocol, Server server,
                                 object waiter)
-
-    cdef open(self, int sockfd)

uvloop/handles/pipe.pyx

@@ -50,7 +50,7 @@ cdef class UnixServer(UVStreamServer):
     cdef _new_socket(self):
         return __pipe_get_socket(<UVSocketHandle>self)
 
-    cdef open(self, int sockfd):
+    cdef _open(self, int sockfd):
         self._ensure_alive()
         __pipe_open(<UVStream>self, sockfd)
         self._mark_as_open()
@@ -89,7 +89,7 @@ cdef class UnixTransport(UVStream):
     cdef _new_socket(self):
         return __pipe_get_socket(<UVSocketHandle>self)
 
-    cdef open(self, int sockfd):
+    cdef _open(self, int sockfd):
         __pipe_open(<UVStream>self, sockfd)
 
     cdef connect(self, char* addr):
@@ -113,7 +113,7 @@ cdef class ReadUnixTransport(UVStream):
     cdef _new_socket(self):
         return __pipe_get_socket(<UVSocketHandle>self)
 
-    cdef open(self, int sockfd):
+    cdef _open(self, int sockfd):
         __pipe_open(<UVStream>self, sockfd)
 
     def get_write_buffer_limits(self):
@@ -163,7 +163,7 @@ cdef class WriteUnixTransport(UVStream):
     cdef _new_socket(self):
         return __pipe_get_socket(<UVSocketHandle>self)
 
-    cdef open(self, int sockfd):
+    cdef _open(self, int sockfd):
         __pipe_open(<UVStream>self, sockfd)
 
     def pause_reading(self):

uvloop/handles/process.pyx

@@ -401,7 +401,7 @@ cdef class UVProcessTransport(UVProcess):
                 self._stdin = WriteUnixTransport.new(
                     self._loop, self.stdin_proto, None, waiter)
                 self._init_futs.append(waiter)
-                self._stdin.open(w)
+                self._stdin._open(w)
                 self._stdin._init_protocol()
             elif _stdin == subprocess_DEVNULL:
                 io[0] = self._file_devnull()
@@ -430,7 +430,7 @@ cdef class UVProcessTransport(UVProcess):
                 self._stdout = ReadUnixTransport.new(
                     self._loop, self.stdout_proto, None, waiter)
                 self._init_futs.append(waiter)
-                self._stdout.open(r)
+                self._stdout._open(r)
                 self._stdout._init_protocol()
             elif _stdout == subprocess_DEVNULL:
                 io[1] = self._file_devnull()
@@ -453,7 +453,7 @@ cdef class UVProcessTransport(UVProcess):
                 self._stderr = ReadUnixTransport.new(
                     self._loop, self.stderr_proto, None, waiter)
                 self._init_futs.append(waiter)
-                self._stderr.open(r)
+                self._stderr._open(r)
                 self._stderr._init_protocol()
             elif _stderr == subprocess_STDOUT:
                 if io[1] is None:

uvloop/handles/tcp.pxd

@@ -1,5 +1,4 @@
 cdef class TCPServer(UVStreamServer):
-    cdef open(self, int sockfd)
     cdef bind(self, system.sockaddr* addr, unsigned int flags=*)
 
     @staticmethod
@@ -15,7 +14,6 @@ cdef class TCPTransport(UVStream):
         system.sockaddr_storage __sockname
 
     cdef bind(self, system.sockaddr* addr, unsigned int flags=*)
-    cdef open(self, int sockfd)
     cdef connect(self, system.sockaddr* addr)
 
     @staticmethod

uvloop/handles/tcp.pyx

@@ -69,7 +69,7 @@ cdef class TCPServer(UVStreamServer):
     cdef _new_socket(self):
         return __tcp_get_socket(<UVSocketHandle>self)
 
-    cdef open(self, int sockfd):
+    cdef _open(self, int sockfd):
         self._ensure_alive()
         try:
             __tcp_open(<UVStream>self, sockfd)
@@ -158,7 +158,7 @@ cdef class TCPTransport(UVStream):
         self._ensure_alive()
         __tcp_bind(<UVStream>self, addr, flags)
 
-    cdef open(self, int sockfd):
+    cdef _open(self, int sockfd):
         self._ensure_alive()
         __tcp_open(<UVStream>self, sockfd)
 

uvloop/handles/udp.pxd

@@ -13,7 +13,7 @@ cdef class UDPTransport(UVBaseTransport):
                              size_t addr_len)
 
     cdef _bind(self, system.sockaddr* addr, bint reuse_addr)
-    cdef _open(self, int family, int sockfd)
+    cdef open(self, int family, int sockfd)
     cdef _set_broadcast(self, bint on)
 
     cdef inline __receiving_started(self)

uvloop/handles/udp.pyx

@@ -86,7 +86,7 @@ cdef class UDPTransport(UVBaseTransport):
         memcpy(&self._address, addr, addr_len)
         self._address_set = 1
 
-    cdef _open(self, int family, int sockfd):
+    cdef open(self, int family, int sockfd):
         if family in (uv.AF_INET, uv.AF_INET6):
             self._family = family
         else: