mirror of
https://github.com/theupdateframework/python-tuf
synced 2026-05-24 10:08:28 +00:00
Add test case for MultiRepoUpdater and fix bugs
This commit is contained in:
Vladimir Diaz
parent
ad29eee3aa
commit
e9abd76c33
1 changed files with 219 additions and 0 deletions
|
|
@ -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:
|
||||
# <urlopen error [Errno 111] Connection refused>
|
||||
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
|
||||
|
|
|
|||
Loading…
Reference in a new issue