Fix test.support imports

Author: fanninpm

Lib/test/ssl_servers.py

@@ -9,10 +9,11 @@
     SimpleHTTPRequestHandler, BaseHTTPRequestHandler)
 
 from test import support
+from test.support import socket_helper
 
 here = os.path.dirname(__file__)
 
-HOST = support.HOST
+HOST = socket_helper.HOST
 CERTFILE = os.path.join(here, 'keycert.pem')
 
 # This one's based on HTTPServer, which is based on socketserver
@@ -49,7 +50,7 @@ class RootedHTTPRequestHandler(SimpleHTTPRequestHandler):
     server_version = "TestHTTPS/1.0"
     root = here
     # Avoid hanging when a request gets interrupted by the client
-    timeout = 5
+    timeout = support.LOOPBACK_TIMEOUT
 
     def translate_path(self, path):
         """Translate a /-separated PATH to the local filename syntax.

Lib/test/test_asynchat.py

@@ -1,6 +1,8 @@
 # test asynchat
 
 from test import support
+from test.support import socket_helper
+
 
 import asynchat
 import asyncore
@@ -12,7 +14,7 @@
 import unittest
 import unittest.mock
 
-HOST = support.HOST
+HOST = socket_helper.HOST
 SERVER_QUIT = b'QUIT\n'
 TIMEOUT = 3.0
 
@@ -26,7 +28,7 @@ def __init__(self, event):
         threading.Thread.__init__(self)
         self.event = event
         self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        self.port = support.bind_port(self.sock)
+        self.port = socket_helper.bind_port(self.sock)
         # This will be set if the client wants us to wait before echoing
         # data back.
         self.start_resend_event = None

Lib/test/test_asyncore.py

@@ -11,6 +11,7 @@
 
 from test import support
 from test.support import os_helper
+from test.support import socket_helper
 from io import BytesIO
 
 if support.PGO:
@@ -93,7 +94,7 @@ def bind_af_aware(sock, addr):
     if HAS_UNIX_SOCKETS and sock.family == socket.AF_UNIX:
         # Make sure the path doesn't exist.
         os_helper.unlink(addr)
-        support.bind_unix_socket(sock, addr)
+        socket_helper.bind_unix_socket(sock, addr)
     else:
         sock.bind(addr)
 
@@ -329,7 +330,7 @@ def test_send(self):
         evt = threading.Event()
         sock = socket.socket()
         sock.settimeout(3)
-        port = support.bind_port(sock)
+        port = socket_helper.bind_port(sock)
 
         cap = BytesIO()
         args = (evt, cap, sock)
@@ -343,7 +344,7 @@ def test_send(self):
             data = b"Suppose there isn't a 16-ton weight?"
             d = dispatcherwithsend_noread()
             d.create_socket()
-            d.connect((support.HOST, port))
+            d.connect((socket_helper.HOST, port))
 
             # give time for socket to connect
             time.sleep(0.1)
@@ -793,12 +794,12 @@ def test_quick_connect(self):
 
 class TestAPI_UseIPv4Sockets(BaseTestAPI):
     family = socket.AF_INET
-    addr = (support.HOST, 0)
+    addr = (socket_helper.HOST, 0)
 
-@unittest.skipUnless(support.IPV6_ENABLED, 'IPv6 support required')
+@unittest.skipUnless(socket_helper.IPV6_ENABLED, 'IPv6 support required')
 class TestAPI_UseIPv6Sockets(BaseTestAPI):
     family = socket.AF_INET6
-    addr = (support.HOSTv6, 0)
+    addr = (socket_helper.HOSTv6, 0)
 
 @unittest.skipUnless(HAS_UNIX_SOCKETS, 'Unix sockets required')
 class TestAPI_UseUnixSockets(BaseTestAPI):

Lib/test/test_ftplib.py

@@ -20,15 +20,17 @@
 
 from unittest import TestCase, skipUnless
 from test import support
-from test.support import HOST, HOSTv6
+from test.support import socket_helper
+from test.support.socket_helper import HOST, HOSTv6
 
 import sys
 if sys.platform == 'win32':
     raise unittest.SkipTest("test_ftplib not working on windows")
 if getattr(sys, '_rustpython_debugbuild', False):
     raise unittest.SkipTest("something's weird on debug builds")
 
-TIMEOUT = 3
+
+TIMEOUT = support.LOOPBACK_TIMEOUT
 # the dummy data returned by server over the data channel when
 # RETR, LIST, NLST, MLSD commands are issued
 RETR_DATA = 'abcde12345\r\n' * 1000
@@ -750,7 +752,7 @@ def is_client_connected():
 
     def test_source_address(self):
         self.client.quit()
-        port = support.find_unused_port()
+        port = socket_helper.find_unused_port()
         try:
             self.client.connect(self.server.host, self.server.port,
                                 source_address=(HOST, port))
@@ -762,7 +764,7 @@ def test_source_address(self):
             raise
 
     def test_source_address_passive_connection(self):
-        port = support.find_unused_port()
+        port = socket_helper.find_unused_port()
         self.client.source_address = (HOST, port)
         try:
             with self.client.transfercmd('list') as sock:
@@ -799,7 +801,7 @@ def test_storlines_too_long(self):
         self.assertRaises(ftplib.Error, self.client.storlines, 'stor', f)
 
 
-@skipUnless(support.IPV6_ENABLED, "IPv6 not enabled")
+@skipUnless(socket_helper.IPV6_ENABLED, "IPv6 not enabled")
 class TestIPv6Environment(TestCase):
 
     def setUp(self):
@@ -987,7 +989,7 @@ def setUp(self):
         self.evt = threading.Event()
         self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
         self.sock.settimeout(20)
-        self.port = support.bind_port(self.sock)
+        self.port = socket_helper.bind_port(self.sock)
         self.server_thread = threading.Thread(target=self.server)
         self.server_thread.daemon = True
         self.server_thread.start()

Lib/test/test_httplib.py

@@ -14,6 +14,7 @@
 
 from test import support
 from test.support import os_helper
+from test.support import socket_helper
 
 here = os.path.dirname(__file__)
 # Self-signed cert file for 'localhost'
@@ -43,7 +44,7 @@
 trailers = "X-Dummy: foo\r\nX-Dumm2: bar\r\n"
 chunked_end = "\r\n"
 
-HOST = support.HOST
+HOST = socket_helper.HOST
 
 class FakeSocket:
     def __init__(self, text, fileclass=io.BytesIO, host=None, port=None):
@@ -1499,8 +1500,8 @@ def test_client_constants(self):
 class SourceAddressTest(TestCase):
     def setUp(self):
         self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        self.port = support.bind_port(self.serv)
-        self.source_port = support.find_unused_port()
+        self.port = socket_helper.bind_port(self.serv)
+        self.source_port = socket_helper.find_unused_port()
         self.serv.listen()
         self.conn = None
 
@@ -1532,7 +1533,7 @@ class TimeoutTest(TestCase):
 
     def setUp(self):
         self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        TimeoutTest.PORT = support.bind_port(self.serv)
+        TimeoutTest.PORT = socket_helper.bind_port(self.serv)
         self.serv.listen()
 
     def tearDown(self):
@@ -1665,7 +1666,7 @@ def test_networked(self):
         # Default settings: requires a valid cert from a trusted CA
         import ssl
         support.requires('network')
-        with support.transient_internet('self-signed.pythontest.net'):
+        with socket_helper.transient_internet('self-signed.pythontest.net'):
             h = client.HTTPSConnection('self-signed.pythontest.net', 443)
             with self.assertRaises(ssl.SSLError) as exc_info:
                 h.request('GET', '/')
@@ -1675,7 +1676,7 @@ def test_networked_noverification(self):
         # Switch off cert verification
         import ssl
         support.requires('network')
-        with support.transient_internet('self-signed.pythontest.net'):
+        with socket_helper.transient_internet('self-signed.pythontest.net'):
             context = ssl._create_unverified_context()
             h = client.HTTPSConnection('self-signed.pythontest.net', 443,
                                        context=context)
@@ -1689,7 +1690,7 @@ def test_networked_noverification(self):
     def test_networked_trusted_by_default_cert(self):
         # Default settings: requires a valid cert from a trusted CA
         support.requires('network')
-        with support.transient_internet('www.python.org'):
+        with socket_helper.transient_internet('www.python.org'):
             h = client.HTTPSConnection('www.python.org', 443)
             h.request('GET', '/')
             resp = h.getresponse()
@@ -1703,7 +1704,7 @@ def test_networked_good_cert(self):
         import ssl
         support.requires('network')
         selfsigned_pythontestdotnet = 'self-signed.pythontest.net'
-        with support.transient_internet(selfsigned_pythontestdotnet):
+        with socket_helper.transient_internet(selfsigned_pythontestdotnet):
             context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
             self.assertEqual(context.verify_mode, ssl.CERT_REQUIRED)
             self.assertEqual(context.check_hostname, True)
@@ -1735,7 +1736,7 @@ def test_networked_bad_cert(self):
         # We feed a "CA" cert that is unrelated to the server's cert
         import ssl
         support.requires('network')
-        with support.transient_internet('self-signed.pythontest.net'):
+        with socket_helper.transient_internet('self-signed.pythontest.net'):
             context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
             context.load_verify_locations(CERT_localhost)
             h = client.HTTPSConnection('self-signed.pythontest.net', 443, context=context)

Lib/test/test_os.py

@@ -33,8 +33,8 @@
 from test import support
 from test.support import import_helper
 from test.support import os_helper
+from test.support import socket_helper
 # TODO: RUSTPYTHON
-# from test.support import socket_helper
 # from test.support import threading_helper
 # from test.support import warnings_helper
 from platform import win32_is_iot
@@ -3352,7 +3352,7 @@ def tearDownClass(cls):
         os_helper.unlink(os_helper.TESTFN)
 
     def setUp(self):
-        self.server = SendfileTestServer((support.HOST, 0))
+        self.server = SendfileTestServer((socket_helper.HOST, 0))
         self.server.start()
         self.client = socket.socket()
         self.client.connect((self.server.host, self.server.port))

Lib/test/test_robotparser.py

@@ -4,6 +4,7 @@
 import unittest
 import urllib.robotparser
 from test import support
+from test.support import socket_helper
 from http.server import BaseHTTPRequestHandler, HTTPServer
 
 
@@ -309,7 +310,7 @@ def log_message(self, format, *args):
 class PasswordProtectedSiteTestCase(unittest.TestCase):
 
     def setUp(self):
-        self.server = HTTPServer((support.HOST, 0), RobotHandler)
+        self.server = HTTPServer((socket_helper.HOST, 0), RobotHandler)
 
         self.t = threading.Thread(
             name='HTTPServer serving',
@@ -329,7 +330,7 @@ def tearDown(self):
     @support.reap_threads
     def testPasswordProtectedSite(self):
         addr = self.server.server_address
-        url = 'http://' + support.HOST + ':' + str(addr[1])
+        url = 'http://' + socket_helper.HOST + ':' + str(addr[1])
         robots_url = url + "/robots.txt"
         parser = urllib.robotparser.RobotFileParser()
         parser.set_url(url)
@@ -345,7 +346,7 @@ class NetworkTestCase(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
         support.requires('network')
-        with support.transient_internet(cls.base_url):
+        with socket_helper.transient_internet(cls.base_url):
             cls.parser = urllib.robotparser.RobotFileParser(cls.robots_txt)
             cls.parser.read()