From 7da6a38335fb7f2098b3c576bedd0aa17c750cf2 Mon Sep 17 00:00:00 2001 From: Lukas Puehringer Date: Mon, 24 Jan 2022 11:52:38 +0100 Subject: [PATCH 1/2] test: define TESTS_DIR constant Define TESTS_DIR constant in tests/util.py as full path to the parent directory of the util module. This may be used to reliably read other files in tests dir, such es "repository_data" or "simple_server", regardless of cwd. This commit also replaces a couple of `getcwd() + "filename"` with `TESTS_DIR + filename`, so that in the future (post #1790) we should be able to invoke the tests from anywhere, not only from within the tests directory as is now the case. Signed-off-by: Lukas Puehringer --- tests/test_trusted_metadata_set.py | 6 ++++-- tests/test_updater_ng.py | 4 +++- tests/utils.py | 6 ++++-- 3 files changed, 11 insertions(+), 5 deletions(-) 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..6ff8c1809c 100644 --- a/tests/test_updater_ng.py +++ b/tests/test_updater_ng.py @@ -76,10 +76,12 @@ def setUp(self) -> None: # 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. 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, From 3ed21abf2d2efd9b7e29177805f5929ea7f9ba78 Mon Sep 17 00:00:00 2001 From: Lukas Puehringer Date: Mon, 24 Jan 2022 13:53:10 +0100 Subject: [PATCH 2/2] test: stop using unittest_toolbox in new tests Update new test modules to stop using unittest_toolbox, in preparation for its removal in #1790. The tools provided by unittest_toolbox can easily (in a more obvious way) be replaced by using the standard library modules `tempfile` and `random` (no more used) directly. In the case of tempdir and -file creation/removal, skipping the use of unittest_toolbox, which does this by default, also uncovers some test cleanup failures, which would occur when temporary test directories were removed while a test server hadn't released them. (see `except OSError: pass` in unittest_toolbox's `tearDown` method) **Change details** **test_fetcher_ng.py:** - Stop implicitly creating (setUp) and removing (tearDown) tmp test dirs. -Move now manual creation of an exemplary targets file to setUpClass, as the same file is used by all tests. And remove it explicitly in tearDownClass after killing the server (see note about failure above). - Trigger URL parsing error with a hardcoded invalid URL string instead of a random string. **test_updater_ng.py** - Stop implicitly creating (setUp) and removing (tearDown) tmp test dirs. - Explicitly create tmp test dirs in setUp, but don't remove them in tearDown to avoid above mentioned failures. They will be removed all at once when removing the tmp root test dir in tearDownClass Signed-off-by: Lukas Puehringer --- tests/test_fetcher_ng.py | 63 ++++++++++++++++------------------------ tests/test_updater_ng.py | 63 ++++++++++++---------------------------- tests/test_utils.py | 3 +- 3 files changed, 45 insertions(+), 84 deletions(-) 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_updater_ng.py b/tests/test_updater_ng.py index 6ff8c1809c..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,58 +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/'. - 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" ) @@ -100,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. @@ -128,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( @@ -139,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: