diff --git a/tests/test_fetcher_ng.py b/tests/test_fetcher_ng.py index 78a8f1c68a..e384cc6a2d 100644 --- a/tests/test_fetcher_ng.py +++ b/tests/test_fetcher_ng.py @@ -20,58 +20,49 @@ 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: @@ -79,9 +70,7 @@ def test_fetch(self) -> None: 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: @@ -89,7 +78,7 @@ def test_fetch_in_chunks(self) -> None: # 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: diff --git a/tests/test_trusted_metadata_set.py b/tests/test_trusted_metadata_set.py index b891aafdf7..2d9a81b251 100644 --- a/tests/test_trusted_metadata_set.py +++ b/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" diff --git a/tests/test_updater_ng.py b/tests/test_updater_ng.py index 2e0d41b21c..69cb21814d 100644 --- a/tests/test_updater_ng.py +++ b/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() diff --git a/tests/test_utils.py b/tests/test_utils.py index 6ad06a0fbb..df4d06e667 100644 --- a/tests/test_utils.py +++ b/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: diff --git a/tests/utils.py b/tests/utils.py index e8060d6cf3..1456b091e2 100644 --- a/tests/utils.py +++ b/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,