test: stop using unittest_toolbox in new tests

tests/test_fetcher_ng.py

@@ -20,76 +20,65 @@
 import urllib3.exceptions
 
 from tests import utils
-from tuf import unittest_toolbox
 from tuf.api import exceptions
 from tuf.ngclient._internal.requests_fetcher import RequestsFetcher
 
 logger = logging.getLogger(__name__)
 
 
-class TestFetcher(unittest_toolbox.Modified_TestCase):
+class TestFetcher(unittest.TestCase):
     """Test RequestsFetcher class."""
 
     server_process_handler: ClassVar[utils.TestServerProcess]
 
     @classmethod
     def setUpClass(cls) -> None:
-        # Launch a SimpleHTTPServer (serves files in the current dir).
-        cls.server_process_handler = utils.TestServerProcess(log=logger)
-
-    @classmethod
-    def tearDownClass(cls) -> None:
-        # Stop server process and perform clean up.
-        cls.server_process_handler.clean()
-
-    def setUp(self) -> None:
         """
         Create a temporary file and launch a simple server in the
         current working directory.
         """
+        cls.server_process_handler = utils.TestServerProcess(log=logger)
 
-        unittest_toolbox.Modified_TestCase.setUp(self)
-
-        # Making a temporary data file.
-        current_dir = os.getcwd()
-        target_filepath = self.make_temp_data_file(directory=current_dir)
-
-        with open(target_filepath, "r", encoding="utf8") as target_fileobj:
-            self.file_contents = target_fileobj.read()
-            self.file_length = len(self.file_contents)
+        cls.file_contents = b"junk data"
+        cls.file_length = len(cls.file_contents)
+        with tempfile.NamedTemporaryFile(
+            dir=os.getcwd(), delete=False
+        ) as cls.target_file:
+            cls.target_file.write(cls.file_contents)
 
-        self.rel_target_filepath = os.path.basename(target_filepath)
-        self.url_prefix = (
+        cls.url_prefix = (
             f"http://{utils.TEST_HOST_ADDRESS}:"
-            f"{str(self.server_process_handler.port)}"
+            f"{str(cls.server_process_handler.port)}"
         )
-        self.url = f"{self.url_prefix}/{self.rel_target_filepath}"
+        target_filename = os.path.basename(cls.target_file.name)
+        cls.url = f"{cls.url_prefix}/{target_filename}"
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        # Stop server process and perform clean up.
+        cls.server_process_handler.clean()
+        os.remove(cls.target_file.name)
 
+    def setUp(self) -> None:
         # Instantiate a concrete instance of FetcherInterface
         self.fetcher = RequestsFetcher()
 
-    def tearDown(self) -> None:
-        # Remove temporary directory
-        unittest_toolbox.Modified_TestCase.tearDown(self)
-
     # Simple fetch.
     def test_fetch(self) -> None:
         with tempfile.TemporaryFile() as temp_file:
             for chunk in self.fetcher.fetch(self.url):
                 temp_file.write(chunk)
 
             temp_file.seek(0)
-            self.assertEqual(
-                self.file_contents, temp_file.read().decode("utf-8")
-            )
+            self.assertEqual(self.file_contents, temp_file.read())
 
     # URL data downloaded in more than one chunk
     def test_fetch_in_chunks(self) -> None:
         # Set a smaller chunk size to ensure that the file will be downloaded
         # in more than one chunk
         self.fetcher.chunk_size = 4
 
-        # expected_chunks_count: 3
+        # expected_chunks_count: 3 (depends on length of self.file_length)
         expected_chunks_count = math.ceil(
             self.file_length / self.fetcher.chunk_size
         )
@@ -102,16 +91,14 @@ def test_fetch_in_chunks(self) -> None:
                 chunks_count += 1
 
             temp_file.seek(0)
-            self.assertEqual(
-                self.file_contents, temp_file.read().decode("utf-8")
-            )
+            self.assertEqual(self.file_contents, temp_file.read())
             # Check that we calculate chunks as expected
             self.assertEqual(chunks_count, expected_chunks_count)
 
     # Incorrect URL parsing
     def test_url_parsing(self) -> None:
         with self.assertRaises(exceptions.DownloadError):
-            self.fetcher.fetch(self.random_string())
+            self.fetcher.fetch("missing-scheme-and-hostname-in-url")
 
     # File not found error
     def test_http_error(self) -> None:
@@ -148,12 +135,12 @@ def test_session_get_timeout(self, mock_session_get: Any) -> None:
     # Simple bytes download
     def test_download_bytes(self) -> None:
         data = self.fetcher.download_bytes(self.url, self.file_length)
-        self.assertEqual(self.file_contents, data.decode("utf-8"))
+        self.assertEqual(self.file_contents, data)
 
     # Download file smaller than required max_length
     def test_download_bytes_upper_length(self) -> None:
         data = self.fetcher.download_bytes(self.url, self.file_length + 4)
-        self.assertEqual(self.file_contents, data.decode("utf-8"))
+        self.assertEqual(self.file_contents, data)
 
     # Download a file bigger than expected
     def test_download_bytes_length_mismatch(self) -> None:

tests/test_trusted_metadata_set.py

@@ -54,7 +54,7 @@ def modify_metadata(
     @classmethod
     def setUpClass(cls) -> None:
         cls.repo_dir = os.path.join(
-            os.getcwd(), "repository_data", "repository", "metadata"
+            utils.TESTS_DIR, "repository_data", "repository", "metadata"
         )
         cls.metadata = {}
         for md in [
@@ -68,7 +68,9 @@ def setUpClass(cls) -> None:
             with open(os.path.join(cls.repo_dir, f"{md}.json"), "rb") as f:
                 cls.metadata[md] = f.read()
 
-        keystore_dir = os.path.join(os.getcwd(), "repository_data", "keystore")
+        keystore_dir = os.path.join(
+            utils.TESTS_DIR, "repository_data", "keystore"
+        )
         cls.keystore = {}
         root_key_dict = import_rsa_privatekey_from_file(
             os.path.join(keystore_dir, Root.type + "_key"), password="password"

tests/test_updater_ng.py

@@ -19,7 +19,7 @@
 from securesystemslib.signer import SSlibSigner
 
 from tests import utils
-from tuf import ngclient, unittest_toolbox
+from tuf import ngclient
 from tuf.api import exceptions
 from tuf.api.metadata import (
     Metadata,
@@ -33,56 +33,41 @@
 logger = logging.getLogger(__name__)
 
 
-class TestUpdater(unittest_toolbox.Modified_TestCase):
+class TestUpdater(unittest.TestCase):
     """Test the Updater class from 'tuf/ngclient/updater.py'."""
 
-    temporary_directory: ClassVar[str]
+    # pylint: disable=too-many-instance-attributes
     server_process_handler: ClassVar[utils.TestServerProcess]
 
     @classmethod
     def setUpClass(cls) -> None:
-        # Create a temporary directory to store the repository, metadata, and
-        # target files. 'temporary_directory' must be deleted in
-        # TearDownModule() so that temporary files are always removed, even when
-        # exceptions occur.
-        cls.temporary_directory = tempfile.mkdtemp(dir=os.getcwd())
-
-        # Needed because in some tests simple_server.py cannot be found.
-        # The reason is that the current working directory
-        # has been changed when executing a subprocess.
-        simple_server_path = os.path.join(os.getcwd(), "simple_server.py")
-
-        # Launch a SimpleHTTPServer (serves files in the current directory).
+        cls.tmp_test_root_dir = tempfile.mkdtemp(dir=os.getcwd())
+
+        # Launch a SimpleHTTPServer
         # Test cases will request metadata and target files that have been
-        # pre-generated in 'tuf/tests/repository_data', which will be served
-        # by the SimpleHTTPServer launched here.
-        cls.server_process_handler = utils.TestServerProcess(
-            log=logger, server=simple_server_path
-        )
+        # pre-generated in 'tuf/tests/repository_data', and are copied to
+        # CWD/tmp_test_root_dir/*
+        cls.server_process_handler = utils.TestServerProcess(log=logger)
 
     @classmethod
     def tearDownClass(cls) -> None:
-        # Cleans the resources and flush the logged lines (if any).
+        # Cleans resources, flush the logged lines (if any) and remove test dir
         cls.server_process_handler.clean()
-
-        # Remove the temporary repository directory, which should contain all
-        # the metadata, targets, and key files generated for the test cases
-        shutil.rmtree(cls.temporary_directory)
+        shutil.rmtree(cls.tmp_test_root_dir)
 
     def setUp(self) -> None:
-        # We are inheriting from custom class.
-        unittest_toolbox.Modified_TestCase.setUp(self)
+        # Create tmp test dir inside of tmp test root dir to independently serve
+        # new repository files for each test. We delete all tmp dirs at once in
+        # tearDownClass after the server has released all resources.
+        self.tmp_test_dir = tempfile.mkdtemp(dir=self.tmp_test_root_dir)
 
         # Copy the original repository files provided in the test folder so that
         # any modifications are restricted to the copies.
         # The 'repository_data' directory is expected to exist in 'tuf.tests/'.
-        original_repository_files = os.path.join(os.getcwd(), "repository_data")
-        temporary_repository_root = self.make_temp_directory(
-            directory=self.temporary_directory
+        original_repository_files = os.path.join(
+            utils.TESTS_DIR, "repository_data"
         )
 
-        # The original repository, keystore, and client directories will be
-        # copied for each test case.
         original_repository = os.path.join(
             original_repository_files, "repository"
         )
@@ -98,15 +83,10 @@ def setUp(self) -> None:
         # Save references to the often-needed client repository directories.
         # Test cases need these references to access metadata and target files.
         self.repository_directory = os.path.join(
-            temporary_repository_root, "repository"
-        )
-        self.keystore_directory = os.path.join(
-            temporary_repository_root, "keystore"
-        )
-
-        self.client_directory = os.path.join(
-            temporary_repository_root, "client"
+            self.tmp_test_dir, "repository"
         )
+        self.keystore_directory = os.path.join(self.tmp_test_dir, "keystore")
+        self.client_directory = os.path.join(self.tmp_test_dir, "client")
 
         # Copy the original 'repository', 'client', and 'keystore' directories
         # to the temporary repository the test cases can use.
@@ -126,7 +106,7 @@ def setUp(self) -> None:
 
         self.metadata_url = f"{url_prefix}/metadata/"
         self.targets_url = f"{url_prefix}/targets/"
-        self.dl_dir = self.make_temp_directory()
+        self.dl_dir = tempfile.mkdtemp(dir=self.tmp_test_dir)
         # Creating a repository instance.  The test cases will use this client
         # updater to refresh metadata, fetch target files, etc.
         self.updater = ngclient.Updater(
@@ -137,9 +117,6 @@ def setUp(self) -> None:
         )
 
     def tearDown(self) -> None:
-        # We are inheriting from custom class.
-        unittest_toolbox.Modified_TestCase.tearDown(self)
-
         # Logs stdout and stderr from the sever subprocess.
         self.server_process_handler.flush_log()
 

tests/test_utils.py

@@ -27,7 +27,6 @@
 import unittest
 
 from tests import utils
-from tuf import unittest_toolbox
 
 logger = logging.getLogger(__name__)
 
@@ -47,7 +46,7 @@ def can_connect(port: int) -> bool:
             sock.close()
 
 
-class TestServerProcess(unittest_toolbox.Modified_TestCase):
+class TestServerProcess(unittest.TestCase):
     """Test functionality provided in TestServerProcess from tests/utils.py."""
 
     def test_simple_server_startup(self) -> None:

tests/utils.py

@@ -39,10 +39,12 @@
 
 logger = logging.getLogger(__name__)
 
+# May may be used to reliably read other files in tests dir regardless of cwd
+TESTS_DIR = os.path.dirname(os.path.realpath(__file__))
+
 # Used when forming URLs on the client side
 TEST_HOST_ADDRESS = "127.0.0.1"
 
-
 # DataSet is only here so type hints can be used.
 DataSet = Dict[str, Any]
 
@@ -183,7 +185,7 @@ class TestServerProcess:
     def __init__(
         self,
         log: logging.Logger,
-        server: str = "simple_server.py",
+        server: str = os.path.join(TESTS_DIR, "simple_server.py"),
         timeout: int = 10,
         popen_cwd: str = ".",
         extra_cmd_args: Optional[List[str]] = None,