move test case, remove test support code from asyncio

Author: nkinnan

Lib/asyncio/base_events.py

@@ -200,17 +200,6 @@ def _set_nodelay(sock):
         pass
 
 
-if hasattr(socket, 'TCP_QUICKACK'):
-    def _set_quickack(sock, val):
-        if (sock.family in {socket.AF_INET, socket.AF_INET6} and
-                sock.type == socket.SOCK_STREAM and
-                sock.proto == socket.IPPROTO_TCP):
-            sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_QUICKACK, val)
-else:
-    def _set_quickack(sock, val):
-        pass
-
-
 def _check_ssl_socket(sock):
     if ssl is not None and isinstance(sock, ssl.SSLSocket):
         raise TypeError("Socket cannot be of type SSLSocket")

Lib/test/test_asyncio/test_base_events.py

@@ -2228,35 +2228,5 @@ def test_set_nodelay(self):
                 self.check_set_nodelay(sock)
 
 
-    def check_set_quickack(self, sock):
-        # quickack already true by default on some OSes
-        opt = sock.getsockopt(socket.IPPROTO_TCP, socket.TCP_QUICKACK)
-        if opt:
-            base_events._set_quickack(sock, 0)
-
-        opt = sock.getsockopt(socket.IPPROTO_TCP, socket.TCP_QUICKACK)
-        self.assertFalse(opt)
-
-        base_events._set_quickack(sock, 1)
-
-        opt = sock.getsockopt(socket.IPPROTO_TCP, socket.TCP_QUICKACK)
-        self.assertTrue(opt)
-
-    @unittest.skipUnless(hasattr(socket, 'TCP_QUICKACK'),
-                         'need socket.TCP_QUICKACK')
-    def test_set_quickack(self):
-        with self.subTest('non-blocking'):
-            sock = socket.socket(family=socket.AF_INET, type=socket.SOCK_STREAM,
-                                 proto=socket.IPPROTO_TCP)
-            with sock:
-                self.check_set_quickack(sock)
-        with self.subTest('blocking'):
-            sock = socket.socket(family=socket.AF_INET, type=socket.SOCK_STREAM,
-                                 proto=socket.IPPROTO_TCP)
-            with sock:
-                sock.setblocking(False)
-                self.check_set_quickack(sock)
-
-
 if __name__ == '__main__':
     unittest.main()

Lib/test/test_socket.py

@@ -6826,6 +6826,35 @@ class TestMacOSTCPFlags(unittest.TestCase):
     def test_tcp_keepalive(self):
         self.assertTrue(socket.TCP_KEEPALIVE)
 
+class TestQuickackFlag(unittest.TestCase):
+    def test_tcp_keepalive(self):
+        self.assertTrue(socket.TCP_KEEPALIVE)
+
+@unittest.skipUnless(hasattr(socket, 'TCP_QUICKACK'), 'need socket.TCP_QUICKACK')
+class TestQuickackFlag(unittest.TestCase):
+    def check_set_quickack(self, sock):
+        # quickack already true by default on some OS distributions
+        opt = sock.getsockopt(socket.IPPROTO_TCP, socket.TCP_QUICKACK)
+        if opt:
+            sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_QUICKACK, 0)
+
+        opt = sock.getsockopt(socket.IPPROTO_TCP, socket.TCP_QUICKACK)
+        self.assertFalse(opt)
+
+        sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_QUICKACK, 1)
+
+        opt = sock.getsockopt(socket.IPPROTO_TCP, socket.TCP_QUICKACK)
+        self.assertTrue(opt)
+
+    def test_set_quickack(self):
+        for blocking in (False, True):
+            with self.subTest(blocking=blocking):
+                sock = socket.socket(family=socket.AF_INET, type=socket.SOCK_STREAM,
+                                     proto=socket.IPPROTO_TCP)
+                with sock:
+                    sock.setblocking(blocking)
+                    self.check_set_quickack(sock)
+
 
 @unittest.skipUnless(sys.platform.startswith("win"), "requires Windows")
 class TestMSWindowsTCPFlags(unittest.TestCase):