]> git.ipfire.org Git - thirdparty/Python/cpython.git/commitdiff
add tests for `asyncio` transport sockets (#100263)
authorKumar Aditya <59607654+kumaraditya303@users.noreply.github.com>
Wed, 28 Dec 2022 04:37:55 +0000 (10:07 +0530)
committerGitHub <noreply@github.com>
Wed, 28 Dec 2022 04:37:55 +0000 (10:07 +0530)
Lib/test/test_asyncio/test_events.py

index 153b2de81722732aec37729d1d5616a3c05556bd..e7771edd2e4afbfe6d5240f3ec0e716ef7d36f3e 100644 (file)
@@ -823,6 +823,29 @@ class EventLoopTestsMixin:
         # close server
         server.close()
 
+    def test_create_server_trsock(self):
+        proto = MyProto(self.loop)
+        f = self.loop.create_server(lambda: proto, '0.0.0.0', 0)
+        server = self.loop.run_until_complete(f)
+        self.assertEqual(len(server.sockets), 1)
+        sock = server.sockets[0]
+        self.assertIsInstance(sock, asyncio.trsock.TransportSocket)
+        host, port = sock.getsockname()
+        self.assertEqual(host, '0.0.0.0')
+        dup = sock.dup()
+        self.addCleanup(dup.close)
+        self.assertIsInstance(dup, socket.socket)
+        self.assertFalse(sock.get_inheritable())
+        with self.assertRaises(ValueError):
+            sock.settimeout(1)
+        sock.settimeout(0)
+        self.assertEqual(sock.gettimeout(), 0)
+        with self.assertRaises(ValueError):
+            sock.setblocking(True)
+        sock.setblocking(False)
+        server.close()
+
+
     @unittest.skipUnless(hasattr(socket, 'SO_REUSEPORT'), 'No SO_REUSEPORT')
     def test_create_server_reuse_port(self):
         proto = MyProto(self.loop)