From e9abd76c339859a06e1ca87ca560c7f5cd89fa8b Mon Sep 17 00:00:00 2001 From: Vladimir Diaz Date: Mon, 13 Feb 2017 16:04:18 -0500 Subject: [PATCH] Add test case for MultiRepoUpdater and fix bugs --- tests/test_updater.py | 219 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 219 insertions(+) diff --git a/tests/test_updater.py b/tests/test_updater.py index 1163afa5..1c6fd007 100755 --- a/tests/test_updater.py +++ b/tests/test_updater.py @@ -1435,6 +1435,225 @@ def test_10__visit_child_role(self): + +class TestMultiRepoUpdater(unittest_toolbox.Modified_TestCase): + + @classmethod + def setUpClass(cls): + # setUpClass() is called before tests in an individual class are executed. + + # 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()) + + # Launch a SimpleHTTPServer (serves files in the current directory). + # 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. The test cases of 'test_updater.py' + # assume the pre-generated metadata files have a specific structure, such + # as a delegated role 'targets/role1', three target files, five key files, + # etc. + cls.SERVER_PORT = 8001 + cls.SERVER_PORT2 = 8002 + command = ['python', 'simple_server.py', str(cls.SERVER_PORT)] + command2 = ['python', 'simple_server.py', str(cls.SERVER_PORT2)] + cls.server_process = subprocess.Popen(command, stderr=subprocess.PIPE) + cls.server_process2 = subprocess.Popen(command2, stderr=subprocess.PIPE) + logger.info('\n\tServer process started.') + logger.info('\tServer process id: ' + str(cls.server_process.pid)) + logger.info('\tServing on port: ' + str(cls.SERVER_PORT)) + cls.url = 'http://localhost:' + str(cls.SERVER_PORT) + os.path.sep + + logger.info('\n\tServer process started.') + logger.info('\tServer process id: ' + str(cls.server_process2.pid)) + logger.info('\tServing on port: ' + str(cls.SERVER_PORT2)) + cls.url2 = 'http://localhost:' + str(cls.SERVER_PORT2) + os.path.sep + + # NOTE: Following error is raised if a delay is not applied: + # + time.sleep(1) + + + + @classmethod + def tearDownClass(cls): + # tearDownModule() is called after all the tests have run. + # http://docs.python.org/2/library/unittest.html#class-and-module-fixtures + + # 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) + + # Kill the SimpleHTTPServer process. + if cls.server_process.returncode is None: + logger.info('\tServer process ' + str(cls.server_process.pid) + ' terminated.') + cls.server_process.kill() + + + + def setUp(self): + # We are inheriting from custom class. + unittest_toolbox.Modified_TestCase.setUp(self) + + self.repository_name = 'test_repository' + self.repository_name2 = 'test_repository2' + + # Copy the original repository files provided in the test folder so that + # any modifications made to repository files 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) + + # The original repository, keystore, and client directories will be copied + # for each test case. + original_repository = os.path.join(original_repository_files, 'repository') + original_keystore = os.path.join(original_repository_files, 'keystore') + original_client = os.path.join(original_repository_files, 'client') + + # 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.client_metadata = os.path.join(self.client_directory, + self.repository_name, 'metadata') + self.client_metadata_current = os.path.join(self.client_metadata, + 'current') + self.client_metadata_previous = os.path.join(self.client_metadata, + 'previous') + + # Copy the original 'repository', 'client', and 'keystore' directories + # to the temporary repository the test cases can use. + shutil.copytree(original_repository, self.repository_directory) + shutil.copytree(original_client, self.client_directory) + shutil.copytree(original_keystore, self.keystore_directory) + + # 'path/to/tmp/repository' -> 'localhost:8001/tmp/repository'. + repository_basepath = self.repository_directory[len(os.getcwd()):] + url_prefix = \ + 'http://localhost:' + str(self.SERVER_PORT) + repository_basepath + + # Setting 'tuf.settings.repository_directory' with the temporary client + # directory copied from the original repository files. + tuf.settings.repositories_directory = self.client_directory + + self.repository_mirrors = {'mirror1': {'url_prefix': url_prefix, + 'metadata_path': 'metadata', + 'targets_path': 'targets', + 'confined_target_dirs': ['']}} + + self.map_file = os.path.join(self.client_directory, 'map.json') + print('map_file: ' + repr(self.map_file)) + + # Creating a repository instance. The test cases will use this client + # updater to refresh metadata, fetch target files, etc. + self.multi_repo_updater = updater.MultiRepoUpdater(self.map_file) + + # Metadata role keys are needed by the test cases to make changes to the + # repository (e.g., adding a new target file to 'targets.json' and then + # requesting a refresh()). + self.role_keys = _load_role_keys(self.keystore_directory) + + + + def tearDown(self): + # We are inheriting from custom class. + unittest_toolbox.Modified_TestCase.tearDown(self) + tuf.roledb.clear_roledb(clear_all=True) + tuf.keydb.clear_keydb(clear_all=True) + + + + + # UNIT TESTS. + def test_get_one_valid_targetinfo(self): + # The client's repository requires a metadata directory (and the 'current' + # and 'previous' sub-directories), and at least the 'root.json' file. + # setUp(), called before each test case, instantiates the required updater + # objects and keys. The needed objects/data is available in + # 'self.repository_updater', 'self.client_directory', etc. + + + # Test: Invalid arguments. + # Invalid 'updater_name' argument. String expected. + self.assertRaises(securesystemslib.exceptions.FormatError, + updater.MultiRepoUpdater, 8, self.repository_mirrors) + + # Invalid 'repository_mirrors' argument. 'tuf.formats.MIRRORDICT_SCHEMA' + # expected. + self.assertRaises(securesystemslib.exceptions.FormatError, + updater.Updater, updater.Updater, 8) + + + # 'tuf.client.updater.py' requires that the client's repositories directory + # be configured in 'tuf.settings.py'. + tuf.settings.repositories_directory = None + self.assertRaises(tuf.exceptions.RepositoryError, updater.Updater, 'test_repository', + self.repository_mirrors) + # Restore 'tuf.settings.repositories_directory' to the original client + # directory. + tuf.settings.repositories_directory = self.client_directory + + + # Test: empty client repository (i.e., no metadata directory). + metadata_backup = self.client_metadata + '.backup' + shutil.move(self.client_metadata, metadata_backup) + self.assertRaises(tuf.exceptions.RepositoryError, updater.Updater, 'test_repository', + self.repository_mirrors) + # Restore the client's metadata directory. + shutil.move(metadata_backup, self.client_metadata) + + + # Test: repository with only a '{repository_directory}/metadata' directory. + # (i.e., missing the required 'current' and 'previous' sub-directories). + current_backup = self.client_metadata_current + '.backup' + previous_backup = self.client_metadata_previous + '.backup' + + shutil.move(self.client_metadata_current, current_backup) + shutil.move(self.client_metadata_previous, previous_backup) + self.assertRaises(tuf.exceptions.RepositoryError, updater.Updater, 'test_repository', + self.repository_mirrors) + + # Restore the client's previous directory. The required 'current' directory + # is still missing. + shutil.move(previous_backup, self.client_metadata_previous) + + # Test: repository with only a '{repository_directory}/metadata/previous' + # directory. + self.assertRaises(tuf.exceptions.RepositoryError, updater.Updater, 'test_repository', + self.repository_mirrors) + # Restore the client's current directory. + shutil.move(current_backup, self.client_metadata_current) + + # Test: repository with a '{repository_directory}/metadata/current' + # directory, but the 'previous' directory is missing. + shutil.move(self.client_metadata_previous, previous_backup) + self.assertRaises(tuf.exceptions.RepositoryError, updater.Updater, 'test_repository', + self.repository_mirrors) + shutil.move(previous_backup, self.client_metadata_previous) + + # Test: repository missing the required 'root.json' file. + client_root_file = os.path.join(self.client_metadata_current, 'root.json') + backup_root_file = client_root_file + '.backup' + shutil.move(client_root_file, backup_root_file) + self.assertRaises(tuf.exceptions.RepositoryError, updater.Updater, 'test_repository', + self.repository_mirrors) + # Restore the client's 'root.json file. + shutil.move(backup_root_file, client_root_file) + + # Test: Normal 'tuf.client.updater.Updater' instantiation. + updater.Updater('test_repository', self.repository_mirrors) + + + + + def _load_role_keys(keystore_directory): # Populating 'self.role_keys' by importing the required public and private