Lint additional four files in the tests folder

tests/aggregate_tests.py

@@ -29,13 +29,16 @@
 import sys
 import unittest
 
-if __name__ == '__main__':
-  suite = unittest.TestLoader().discover(".")
-  all_tests_passed = unittest.TextTestRunner(
-      verbosity=1, buffer=True).run(suite).wasSuccessful()
-
-  if not all_tests_passed:
-    sys.exit(1)
-
-  else:
-    sys.exit(0)
+if __name__ == "__main__":
+    suite = unittest.TestLoader().discover(".")
+    all_tests_passed = (
+        unittest.TextTestRunner(verbosity=1, buffer=True)
+        .run(suite)
+        .wasSuccessful()
+    )
+
+    if not all_tests_passed:
+        sys.exit(1)
+
+    else:
+        sys.exit(0)

tests/simple_server.py

@@ -25,40 +25,42 @@
     http://docs.python.org/library/simplehttpserver.html#module-SimpleHTTPServer
 """
 
-import sys
-import random
 import socketserver
+import sys
 from http.server import SimpleHTTPRequestHandler
+from typing import Type, Union
 
 
 class QuietHTTPRequestHandler(SimpleHTTPRequestHandler):
-  """A SimpleHTTPRequestHandler that does not write incoming requests to
-  stderr. """
-  def log_request(self, code='-', size='-'):
-    pass
+    """A SimpleHTTPRequestHandler that does not write incoming requests to
+    stderr."""
+
+    def log_request(
+        self, code: Union[int, str] = "-", size: Union[int, str] = "-"
+    ) -> None:
+        pass
+
 
 # NOTE: On Windows/Python2 tests that use this simple_server.py in a
 # subprocesses hang after a certain amount of requests (~68), if a PIPE is
 # passed as Popen's stderr argument. This problem doesn't emerge if
 # we silence the HTTP messages.
 # If you decide to receive the HTTP messages, then this bug
 # could reappear.
-use_quiet_http_request_handler = True
 
-if len(sys.argv) > 2:
-  use_quiet_http_request_handler = sys.argv[2]
+# pylint: disable=invalid-name
+handler: Type[Union[SimpleHTTPRequestHandler, QuietHTTPRequestHandler]]
 
-if use_quiet_http_request_handler:
-  handler = QuietHTTPRequestHandler
+if len(sys.argv) > 2 and sys.argv[2]:
+    handler = QuietHTTPRequestHandler
 else:
-  handler = SimpleHTTPRequestHandler
+    handler = SimpleHTTPRequestHandler
 
 # Allow re-use so you can re-run tests as often as you want even if the
 # tests re-use ports. Otherwise TCP TIME-WAIT prevents reuse for ~1 minute
 socketserver.TCPServer.allow_reuse_address = True
 
-httpd = socketserver.TCPServer(('localhost', 0), handler)
-port_message = 'bind succeeded, server port is: ' \
-    + str(httpd.server_address[1])
+httpd = socketserver.TCPServer(("localhost", 0), handler)
+port_message = "bind succeeded, server port is: " + str(httpd.server_address[1])
 print(port_message)
 httpd.serve_forever()

tests/test_utils.py

@@ -20,104 +20,108 @@
   Provide tests for some of the functions in utils.py module.
 """
 
-import os
 import logging
-import unittest
+import os
 import socket
 import sys
-
-import tuf.unittest_toolbox as unittest_toolbox
+import unittest
 
 from tests import utils
+from tuf import unittest_toolbox
 
 logger = logging.getLogger(__name__)
 
-class TestServerProcess(unittest_toolbox.Modified_TestCase):
-
-  def tearDown(self):
-    # Make sure we are calling clean on existing attribute.
-    if hasattr(self, 'server_process_handler'):
-      self.server_process_handler.clean()
-
 
-  def can_connect(self):
+def can_connect(port: int) -> bool:
+    """Check if a socket can connect on the given port"""
     try:
-      sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-      sock.connect(('localhost', self.server_process_handler.port))
-      return True
-    except:
-      return False
+        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        sock.connect(("localhost", port))
+        return True
+    # pylint: disable=broad-except
+    except Exception:
+        return False
     finally:
-      # The process will always enter in finally even we return.
-      if sock:
-        sock.close()
-
-
-  def test_simple_server_startup(self):
-    # Test normal case
-    self.server_process_handler = utils.TestServerProcess(log=logger)
-
-    # Make sure we can connect to the server
-    self.assertTrue(self.can_connect())
-
-
-  def test_simple_https_server_startup(self):
-    # Test normal case
-    good_cert_path = os.path.join('ssl_certs', 'ssl_cert.crt')
-    self.server_process_handler = utils.TestServerProcess(log=logger,
-        server='simple_https_server.py', extra_cmd_args=[good_cert_path])
-
-    # Make sure we can connect to the server
-    self.assertTrue(self.can_connect())
-    self.server_process_handler.clean()
-
-    # Test when no cert file is provided
-    self.server_process_handler = utils.TestServerProcess(log=logger,
-        server='simple_https_server.py')
+        # The process will always enter in finally even after return.
+        if sock:
+            sock.close()
 
-    # Make sure we can connect to the server
-    self.assertTrue(self.can_connect())
-    self.server_process_handler.clean()
 
-    # Test with a non existing cert file.
-    non_existing_cert_path = os.path.join('ssl_certs', 'non_existing.crt')
-    self.server_process_handler = utils.TestServerProcess(log=logger,
-        server='simple_https_server.py',
-        extra_cmd_args=[non_existing_cert_path])
-
-    # Make sure we can connect to the server
-    self.assertTrue(self.can_connect())
-
-
-  def test_slow_retrieval_server_startup(self):
-    # Test normal case
-    self.server_process_handler = utils.TestServerProcess(log=logger,
-        server='slow_retrieval_server.py')
-
-    # Make sure we can connect to the server
-    self.assertTrue(self.can_connect())
-
-
-  def test_cleanup(self):
-    # Test normal case
-    self.server_process_handler = utils.TestServerProcess(log=logger,
-        server='simple_server.py')
-
-    self.server_process_handler.clean()
-
-    # Check if the process has successfully been killed.
-    self.assertFalse(self.server_process_handler.is_process_running())
-
-
-  def test_server_exit_before_timeout(self):
-    with self.assertRaises(utils.TestServerProcessError):
-      utils.TestServerProcess(logger, server='non_existing_server.py')
-
-    # Test starting a server which immediately exits."
-    with self.assertRaises(utils.TestServerProcessError):
-      utils.TestServerProcess(logger, server='fast_server_exit.py')
-
-
-if __name__ == '__main__':
-  utils.configure_test_logging(sys.argv)
-  unittest.main()
+class TestServerProcess(unittest_toolbox.Modified_TestCase):
+    """Test functionality provided in TestServerProcess from tests/utils.py."""
+
+    def test_simple_server_startup(self) -> None:
+        # Test normal case
+        server_process_handler = utils.TestServerProcess(log=logger)
+
+        # Make sure we can connect to the server
+        self.assertTrue(can_connect(server_process_handler.port))
+        server_process_handler.clean()
+
+    def test_simple_https_server_startup(self) -> None:
+        # Test normal case
+        good_cert_path = os.path.join("ssl_certs", "ssl_cert.crt")
+        server_process_handler = utils.TestServerProcess(
+            log=logger,
+            server="simple_https_server.py",
+            extra_cmd_args=[good_cert_path],
+        )
+
+        # Make sure we can connect to the server
+        self.assertTrue(can_connect(server_process_handler.port))
+        server_process_handler.clean()
+
+        # Test when no cert file is provided
+        server_process_handler = utils.TestServerProcess(
+            log=logger, server="simple_https_server.py"
+        )
+
+        # Make sure we can connect to the server
+        self.assertTrue(can_connect(server_process_handler.port))
+        server_process_handler.clean()
+
+        # Test with a non existing cert file.
+        non_existing_cert_path = os.path.join("ssl_certs", "non_existing.crt")
+        server_process_handler = utils.TestServerProcess(
+            log=logger,
+            server="simple_https_server.py",
+            extra_cmd_args=[non_existing_cert_path],
+        )
+
+        # Make sure we can connect to the server
+        self.assertTrue(can_connect(server_process_handler.port))
+        server_process_handler.clean()
+
+    def test_slow_retrieval_server_startup(self) -> None:
+        # Test normal case
+        server_process_handler = utils.TestServerProcess(
+            log=logger, server="slow_retrieval_server.py"
+        )
+
+        # Make sure we can connect to the server
+        self.assertTrue(can_connect(server_process_handler.port))
+        server_process_handler.clean()
+
+    def test_cleanup(self) -> None:
+        # Test normal case
+        server_process_handler = utils.TestServerProcess(
+            log=logger, server="simple_server.py"
+        )
+
+        server_process_handler.clean()
+
+        # Check if the process has successfully been killed.
+        self.assertFalse(server_process_handler.is_process_running())
+
+    def test_server_exit_before_timeout(self) -> None:
+        with self.assertRaises(utils.TestServerProcessError):
+            utils.TestServerProcess(logger, server="non_existing_server.py")
+
+        # Test starting a server which immediately exits."
+        with self.assertRaises(utils.TestServerProcessError):
+            utils.TestServerProcess(logger, server="fast_server_exit.py")
+
+
+if __name__ == "__main__":
+    utils.configure_test_logging(sys.argv)
+    unittest.main()