Issue #28652: Make loop methods reject socket kinds they do not support.

3 files changed, 90 insertions, 11 deletions

diff --git a/Lib/test/test_asyncio/test_base_events.py b/Lib/test/test_asyncio/test_base_events.py
index cdbd587..2a93923 100644
--- a/Lib/test/test_asyncio/test_base_events.py
+++ b/Lib/test/test_asyncio/test_base_events.py
@@ -116,6 +116,13 @@ class BaseEventTests(test_utils.TestCase):
         self.assertIsNone(
             base_events._ipaddr_info('::3%lo0', 1, INET6, STREAM, TCP))
 
+        if hasattr(socket, 'SOCK_NONBLOCK'):
+            self.assertEqual(
+                None,
+                base_events._ipaddr_info(
+                    '1.2.3.4', 1, INET, STREAM | socket.SOCK_NONBLOCK, TCP))
+
+
     def test_port_parameter_types(self):
         # Test obscure kinds of arguments for "port".
         INET = socket.AF_INET
@@ -1040,6 +1047,43 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase):
             MyProto, 'example.com', 80, sock=object())
         self.assertRaises(ValueError, self.loop.run_until_complete, coro)
 
+    @unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'no Unix sockets')
+    def test_create_connection_wrong_sock(self):
+        sock = socket.socket(socket.AF_UNIX)
+        with sock:
+            coro = self.loop.create_connection(MyProto, sock=sock)
+            with self.assertRaisesRegex(ValueError,
+                                        'A TCP Stream Socket was expected'):
+                self.loop.run_until_complete(coro)
+
+    @unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'no Unix sockets')
+    def test_create_server_wrong_sock(self):
+        sock = socket.socket(socket.AF_UNIX)
+        with sock:
+            coro = self.loop.create_server(MyProto, sock=sock)
+            with self.assertRaisesRegex(ValueError,
+                                        'A TCP Stream Socket was expected'):
+                self.loop.run_until_complete(coro)
+
+    @unittest.skipUnless(hasattr(socket, 'SOCK_NONBLOCK'),
+                         'no socket.SOCK_NONBLOCK (linux only)')
+    def test_create_server_stream_bittype(self):
+        sock = socket.socket(
+            socket.AF_INET, socket.SOCK_STREAM | socket.SOCK_NONBLOCK)
+        with sock:
+            coro = self.loop.create_server(lambda: None, sock=sock)
+            srv = self.loop.run_until_complete(coro)
+            srv.close()
+            self.loop.run_until_complete(srv.wait_closed())
+
+    def test_create_datagram_endpoint_wrong_sock(self):
+        sock = socket.socket(socket.AF_INET)
+        with sock:
+            coro = self.loop.create_datagram_endpoint(MyProto, sock=sock)
+            with self.assertRaisesRegex(ValueError,
+                                        'A UDP Socket was expected'):
+                self.loop.run_until_complete(coro)
+
     def test_create_connection_no_host_port_sock(self):
         coro = self.loop.create_connection(MyProto)
         self.assertRaises(ValueError, self.loop.run_until_complete, coro)
@@ -1487,36 +1531,39 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase):
         self.assertEqual('CLOSED', protocol.state)
 
     def test_create_datagram_endpoint_sock_sockopts(self):
+        class FakeSock:
+            type = socket.SOCK_DGRAM
+
         fut = self.loop.create_datagram_endpoint(
-            MyDatagramProto, local_addr=('127.0.0.1', 0), sock=object())
+            MyDatagramProto, local_addr=('127.0.0.1', 0), sock=FakeSock())
         self.assertRaises(ValueError, self.loop.run_until_complete, fut)
 
         fut = self.loop.create_datagram_endpoint(
-            MyDatagramProto, remote_addr=('127.0.0.1', 0), sock=object())
+            MyDatagramProto, remote_addr=('127.0.0.1', 0), sock=FakeSock())
         self.assertRaises(ValueError, self.loop.run_until_complete, fut)
 
         fut = self.loop.create_datagram_endpoint(
-            MyDatagramProto, family=1, sock=object())
+            MyDatagramProto, family=1, sock=FakeSock())
         self.assertRaises(ValueError, self.loop.run_until_complete, fut)
 
         fut = self.loop.create_datagram_endpoint(
-            MyDatagramProto, proto=1, sock=object())
+            MyDatagramProto, proto=1, sock=FakeSock())
         self.assertRaises(ValueError, self.loop.run_until_complete, fut)
 
         fut = self.loop.create_datagram_endpoint(
-            MyDatagramProto, flags=1, sock=object())
+            MyDatagramProto, flags=1, sock=FakeSock())
         self.assertRaises(ValueError, self.loop.run_until_complete, fut)
 
         fut = self.loop.create_datagram_endpoint(
-            MyDatagramProto, reuse_address=True, sock=object())
+            MyDatagramProto, reuse_address=True, sock=FakeSock())
         self.assertRaises(ValueError, self.loop.run_until_complete, fut)
 
         fut = self.loop.create_datagram_endpoint(
-            MyDatagramProto, reuse_port=True, sock=object())
+            MyDatagramProto, reuse_port=True, sock=FakeSock())
         self.assertRaises(ValueError, self.loop.run_until_complete, fut)
 
         fut = self.loop.create_datagram_endpoint(
-            MyDatagramProto, allow_broadcast=True, sock=object())
+            MyDatagramProto, allow_broadcast=True, sock=FakeSock())
         self.assertRaises(ValueError, self.loop.run_until_complete, fut)
 
     def test_create_datagram_endpoint_sockopts(self):
diff --git a/Lib/test/test_asyncio/test_events.py b/Lib/test/test_asyncio/test_events.py
index 5b32332..28d92a9 100644
--- a/Lib/test/test_asyncio/test_events.py
+++ b/Lib/test/test_asyncio/test_events.py
@@ -791,9 +791,9 @@ class EventLoopTestsMixin:
         conn, _ = lsock.accept()
         proto = MyProto(loop=loop)
         proto.loop = loop
-        f = loop.create_task(
+        loop.run_until_complete(
             loop.connect_accepted_socket(
-                (lambda : proto), conn, ssl=server_ssl))
+                (lambda: proto), conn, ssl=server_ssl))
         loop.run_forever()
         proto.transport.close()
         lsock.close()
@@ -1377,6 +1377,11 @@ class EventLoopTestsMixin:
         server.transport.close()
 
     def test_create_datagram_endpoint_sock(self):
+        if (sys.platform == 'win32' and
+                isinstance(self.loop, proactor_events.BaseProactorEventLoop)):
+            raise unittest.SkipTest(
+                'UDP is not supported with proactor event loops')
+
         sock = None
         local_address = ('127.0.0.1', 0)
         infos = self.loop.run_until_complete(
@@ -1394,7 +1399,7 @@ class EventLoopTestsMixin:
         else:
             assert False, 'Can not create socket.'
 
-        f = self.loop.create_connection(
+        f = self.loop.create_datagram_endpoint(
             lambda: MyDatagramProto(loop=self.loop), sock=sock)
         tr, pr = self.loop.run_until_complete(f)
         self.assertIsInstance(tr, asyncio.Transport)
diff --git a/Lib/test/test_asyncio/test_unix_events.py b/Lib/test/test_asyncio/test_unix_events.py
index 83a035e..89c6eed 100644
--- a/Lib/test/test_asyncio/test_unix_events.py
+++ b/Lib/test/test_asyncio/test_unix_events.py
@@ -280,6 +280,33 @@ class SelectorEventLoopUnixSocketTests(test_utils.TestCase):
                                         'A UNIX Domain Stream.*was expected'):
                 self.loop.run_until_complete(coro)
 
+    def test_create_unix_server_path_dgram(self):
+        sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
+        with sock:
+            coro = self.loop.create_unix_server(lambda: None, path=None,
+                                                sock=sock)
+            with self.assertRaisesRegex(ValueError,
+                                        'A UNIX Domain Stream.*was expected'):
+                self.loop.run_until_complete(coro)
+
+    @unittest.skipUnless(hasattr(socket, 'SOCK_NONBLOCK'),
+                         'no socket.SOCK_NONBLOCK (linux only)')
+    def test_create_unix_server_path_stream_bittype(self):
+        sock = socket.socket(
+            socket.AF_UNIX, socket.SOCK_STREAM | socket.SOCK_NONBLOCK)
+        with tempfile.NamedTemporaryFile() as file:
+            fn = file.name
+        try:
+            with sock:
+                sock.bind(fn)
+                coro = self.loop.create_unix_server(lambda: None, path=None,
+                                                    sock=sock)
+                srv = self.loop.run_until_complete(coro)
+                srv.close()
+                self.loop.run_until_complete(srv.wait_closed())
+        finally:
+            os.unlink(fn)
+
     def test_create_unix_connection_path_inetsock(self):
         sock = socket.socket()
         with sock: