text: #!/usr/bin/env python
  
  # Copyright 2014 - 2017, New York University and the TUF contributors
  # SPDX-License-Identifier: MIT OR Apache-2.0
  
  """
  <Program Name>
    test_developer_tool.py.
  
  <Authors>
    Santiago Torres Arias <torresariass@gmail.com>
    Zane Fisher <zanefisher@gmail.com>
  
  <Started>
    January 22, 2014.
  
  <Copyright>
    See LICENSE-MIT OR LICENSE for licensing information.
  
  <Purpose>
    Unit test for the 'developer_tool.py' module.
  """
  
  import os
  import time
  import datetime
  import unittest
  import logging
  import tempfile
  import shutil
  import unittest
  
  import tuf
  import tuf.log
  import tuf.roledb
  import tuf.keydb
  import tuf.developer_tool as developer_tool
  import tuf.exceptions
  
  import securesystemslib
  
  from tuf.developer_tool import METADATA_DIRECTORY_NAME
  from tuf.developer_tool import TARGETS_DIRECTORY_NAME
  
  logger = logging.getLogger('tuf.test_developer_tool')
  
  developer_tool.disable_console_log_messages()
  
  class TestProject(unittest.TestCase):
  
    tmp_dir = None
  
    @classmethod
    def setUpClass(cls):
      cls.tmp_dir = tempfile.mkdtemp(dir = os.getcwd())
  
    @classmethod
    def tearDownClass(cls):
      shutil.rmtree(cls.tmp_dir)
  
    def setUp(self):
      # called before every test case
      pass
  
    def tearDown(self):
      # called after every test case
      tuf.roledb.clear_roledb(clear_all=True)
      tuf.keydb.clear_keydb(clear_all=True)
  
  
    def test_create_new_project(self):
      # Test cases for the create_new_project function. In this test we will
      # check input, correct file creation and format. We also check
      # that a proper object is generated. We will use the normal layout for this
      # test suite.
  
      # Create a local subfolder for this test.
      local_tmp = tempfile.mkdtemp(dir = self.tmp_dir)
  
      # These are the usual values we will be throwing to the function, however
      # we will swap these for nulls or malformed values every now and then to
      # test input.
      project_name = 'test_suite'
      metadata_directory = local_tmp
      location_in_repository = '/prefix'
      targets_directory = None
      key = None
  
      # Create a blank project.
      project = developer_tool.create_new_project(project_name, metadata_directory,
          location_in_repository)
  
      self.assertTrue(isinstance(project, developer_tool.Project))
      self.assertTrue(project.layout_type == 'repo-like')
      self.assertTrue(project.prefix == location_in_repository)
      self.assertTrue(project.project_name == project_name)
      self.assertTrue(project.metadata_directory ==
          os.path.join(metadata_directory,METADATA_DIRECTORY_NAME))
      self.assertTrue(project.targets_directory ==
          os.path.join(metadata_directory,TARGETS_DIRECTORY_NAME))
  
      # Create a blank project without a prefix.
      project = developer_tool.create_new_project(project_name, metadata_directory)
      self.assertTrue(isinstance(project, developer_tool.Project))
      self.assertTrue(project.layout_type == 'repo-like')
      self.assertTrue(project.prefix == '')
      self.assertTrue(project.project_name == project_name)
      self.assertTrue(project.metadata_directory ==
          os.path.join(metadata_directory,METADATA_DIRECTORY_NAME))
      self.assertTrue(project.targets_directory ==
          os.path.join(metadata_directory,TARGETS_DIRECTORY_NAME))
  
      # Create a blank project without a valid metadata directory.
      self.assertRaises(securesystemslib.exceptions.FormatError, developer_tool.create_new_project,
         0, metadata_directory, location_in_repository)
      self.assertRaises(securesystemslib.exceptions.FormatError, developer_tool.create_new_project,
         project_name, 0, location_in_repository)
      self.assertRaises(securesystemslib.exceptions.FormatError, developer_tool.create_new_project,
         project_name, metadata_directory, 0)
  
  
      # Create a new project with a flat layout.
      targets_directory = tempfile.mkdtemp(dir = local_tmp)
      metadata_directory = tempfile.mkdtemp(dir = local_tmp)
      project = developer_tool.create_new_project(project_name, metadata_directory,
          location_in_repository, targets_directory)
      self.assertTrue(isinstance(project, developer_tool.Project))
      self.assertTrue(project.layout_type == 'flat')
      self.assertTrue(project.prefix == location_in_repository)
      self.assertTrue(project.project_name == project_name)
      self.assertTrue(project.metadata_directory == metadata_directory)
      self.assertTrue(project.targets_directory == targets_directory)
  
      # Finally, check that if targets_directory is set, it is valid.
      self.assertRaises(securesystemslib.exceptions.FormatError, developer_tool.create_new_project,
          project_name, metadata_directory, location_in_repository, 0)
  
      # Copy a key to our workspace and create a new project with it.
      keystore_path = os.path.join('repository_data','keystore')
  
      # I will use the same key as the one provided in the repository
      # tool tests for the root role, but this is not a root role...
      root_key_path = os.path.join(keystore_path,'root_key.pub')
      project_key = securesystemslib.interface.import_rsa_publickey_from_file(root_key_path)
  
      # Test create new project with a key added by default.
      project = developer_tool.create_new_project(project_name, metadata_directory,
          location_in_repository, targets_directory, project_key)
  
      self.assertTrue(isinstance(project, developer_tool.Project))
      self.assertTrue(project.layout_type == 'flat')
      self.assertTrue(project.prefix == location_in_repository)
      self.assertTrue(project.project_name == project_name)
      self.assertTrue(project.metadata_directory == metadata_directory)
      self.assertTrue(project.targets_directory == targets_directory)
      self.assertTrue(len(project.keys) == 1)
      self.assertTrue(project.keys[0] == project_key['keyid'])
  
      # Try to write to an invalid location.  The OSError should be re-raised by
      # create_new_project().
      shutil.rmtree(targets_directory)
      tuf.roledb.clear_roledb()
      tuf.keydb.clear_keydb()
  
      metadata_directory = '/'
      valid_metadata_directory_name = developer_tool.METADATA_DIRECTORY_NAME
      developer_tool.METADATA_DIRECTORY_NAME = '/'
  
      try:
        developer_tool.create_new_project(project_name, metadata_directory,
            location_in_repository, targets_directory, project_key)
  
      except (OSError, securesystemslib.exceptions.RepositoryError):
        pass
  
      developer_tool.METADATA_DIRECTORY_NAME = valid_metadata_directory_name
  
  
  
    def test_load_project(self):
      # This test case will first try to load an existing project and test for
      # verify the loaded object.  It will next try to load a nonexisting project
      # and expect a correct error handler.  Finally, it will try to overwrite the
      # existing prefix of the loaded project.
  
      # Create a local subfolder for this test.
      local_tmp = tempfile.mkdtemp(dir = self.tmp_dir)
  
      # Test non-existent project filepath.
      nonexistent_path = os.path.join(local_tmp, 'nonexistent')
      self.assertRaises(IOError, developer_tool.load_project, nonexistent_path)
  
      # Copy the pregenerated metadata.
      project_data_filepath = os.path.join('repository_data', 'project')
      target_project_data_filepath = os.path.join(local_tmp, 'project')
      shutil.copytree('repository_data/project', target_project_data_filepath)
  
      # Properly load a project.
      repo_filepath = os.path.join(local_tmp, 'project', 'test-flat')
      new_targets_path = os.path.join(local_tmp, 'project', 'targets')
      project = developer_tool.load_project(repo_filepath,
          new_targets_location = new_targets_path)
      self.assertTrue(project._targets_directory == new_targets_path)
      self.assertTrue(project.layout_type == 'flat')
  
      # Load a project overwriting the prefix.
      project = developer_tool.load_project(repo_filepath, prefix='new')
      self.assertTrue(project.prefix == 'new')
  
      # Load a project with a file missing.
      file_to_corrupt = os.path.join(repo_filepath, 'test-flat.json')
      with open(file_to_corrupt, 'wt') as fp:
        fp.write('this is not a json file')
  
      self.assertRaises(securesystemslib.exceptions.Error, developer_tool.load_project, repo_filepath)
  
  
  
  
    def test_add_verification_keys(self):
      # Create a new project instance.
      project = developer_tool.Project('test_verification_keys', 'somepath',
          'someotherpath', 'prefix')
  
      # Add invalid verification key.
      self.assertRaises(securesystemslib.exceptions.FormatError, project.add_verification_key, 'invalid')
  
      # Add verification key.
      #  - load it first
      keystore_path = os.path.join('repository_data', 'keystore')
      first_verification_key_path = os.path.join(keystore_path,'root_key.pub')
      first_verification_key = \
        securesystemslib.interface.import_rsa_publickey_from_file(first_verification_key_path)
  
      project.add_verification_key(first_verification_key)
  
  
      # Add another verification key (should expect exception.)
      second_verification_key_path = os.path.join(keystore_path, 'snapshot_key.pub')
      second_verification_key = \
        securesystemslib.interface.import_ed25519_publickey_from_file(second_verification_key_path)
  
      self.assertRaises(securesystemslib.exceptions.Error,
          project.add_verification_key,(second_verification_key))
  
  
  
      # Add a verification key for the delegation.
      project.delegate('somedelegation', [], [])
      project('somedelegation').add_verification_key(first_verification_key)
      project('somedelegation').add_verification_key(second_verification_key)
  
  
      # Add another delegation of the delegation.
      project('somedelegation').delegate('somesubdelegation', [], [])
      project('somesubdelegation').add_verification_key(first_verification_key)
      project('somesubdelegation').add_verification_key(second_verification_key)
  
  
    def test_write(self):
  
      # Create tmp directory.
      local_tmp = tempfile.mkdtemp(dir=self.tmp_dir)
  
      # Create new project inside tmp directory.
      project = developer_tool.create_new_project('new_project', local_tmp,
          'prefix');
  
      # Create some target files inside the tmp directory.
      target_filepath = os.path.join(local_tmp, 'targets', 'test_target')
      with open(target_filepath, 'wt') as fp:
        fp.write('testing file')
  
  
      # Add the targets.
      project.add_target(os.path.basename(target_filepath))
  
      # Add verification keys.
      keystore_path = os.path.join('repository_data', 'keystore')
      project_key_path = os.path.join(keystore_path, 'root_key.pub')
      project_key = \
        securesystemslib.interface.import_rsa_publickey_from_file(project_key_path)
  
  
      # Call status (for the sake of doing it and to improve test coverage by
      # executing its statements.)
      project.status()
  
      project.add_verification_key(project_key)
  
  
      # Add another verification key (should expect exception.)
      delegation_key_path = os.path.join(keystore_path, 'snapshot_key.pub')
      delegation_key = \
        securesystemslib.interface.import_ed25519_publickey_from_file(delegation_key_path)
  
      # Add a subdelegation.
      subdelegation_key_path = os.path.join(keystore_path, 'timestamp_key.pub')
      subdelegation_key = \
          securesystemslib.interface.import_ed25519_publickey_from_file(subdelegation_key_path)
  
      # Add a delegation.
      project.delegate('delegation', [delegation_key], [])
      project('delegation').delegate('subdelegation', [subdelegation_key], [])
  
      # call write (except)
      self.assertRaises(securesystemslib.exceptions.Error, project.write, ())
  
      # Call status (for the sake of doing it and executing its statements.)
      project.status()
  
      # Load private keys.
      project_private_key_path = os.path.join(keystore_path, 'root_key')
      project_private_key = \
          developer_tool.import_rsa_privatekey_from_file(project_private_key_path,
              'password')
  
      delegation_private_key_path = os.path.join(keystore_path, 'snapshot_key')
      delegation_private_key = \
          developer_tool.import_ed25519_privatekey_from_file(delegation_private_key_path,
              'password')
  
      subdelegation_private_key_path =  \
          os.path.join(keystore_path, 'timestamp_key')
      subdelegation_private_key = \
          developer_tool.import_ed25519_privatekey_from_file(subdelegation_private_key_path,
              'password')
  
      # Test partial write.
      # backup everything (again)
      # + backup targets.
      targets_backup = project.target_files
  
      # + backup delegations.
      delegations_backup = \
          tuf.roledb.get_delegated_rolenames(project.project_name)
  
      # + backup layout type.
      layout_type_backup = project.layout_type
  
      # + backup keyids.
      keys_backup = project.keys
      delegation_keys_backup = project('delegation').keys
  
      # + backup the prefix.
      prefix_backup = project.prefix
  
      # + backup the name.
      name_backup = project.project_name
  
      # Write and reload.
      self.assertRaises(securesystemslib.exceptions.Error, project.write)
      project.write(write_partial=True)
  
      project = developer_tool.load_project(local_tmp)
  
      # Check against backup.
      self.assertEqual(list(project.target_files.keys()), list(targets_backup.keys()))
      new_delegations = tuf.roledb.get_delegated_rolenames(project.project_name)
      self.assertEqual(new_delegations, delegations_backup)
      self.assertEqual(project.layout_type, layout_type_backup)
      self.assertEqual(project.keys, keys_backup)
  
      self.assertEqual(project('delegation').keys, delegation_keys_backup)
  
      self.assertEqual(project.prefix, prefix_backup)
      self.assertEqual(project.project_name, name_backup)
  
      roleinfo = tuf.roledb.get_roleinfo(project.project_name)
  
      self.assertEqual(roleinfo['partial_loaded'], True)
  
  
  
      # Load_signing_keys.
      project('delegation').load_signing_key(delegation_private_key)
  
      project.status()
  
      project.load_signing_key(project_private_key)
  
      # Backup everything.
      # + backup targets.
      targets_backup = project.target_files
  
      # + backup delegations.
      delegations_backup = \
          tuf.roledb.get_delegated_rolenames(project.project_name)
  
      # + backup layout type.
      layout_type_backup = project.layout_type
  
      # + backup keyids
      keys_backup = project.keys
      delegation_keys_backup = project('delegation').keys
  
      # + backup the prefix.
      prefix_backup = project.prefix
  
      # + backup the name.
      name_backup = project.project_name
  
      # Call status (for the sake of doing it.)
      project.status()
  
      # Call write.
      project.write()
  
      # Call load.
      project = developer_tool.load_project(local_tmp)
  
  
      # Check against backup.
      self.assertEqual(list(project.target_files.keys()), list(targets_backup.keys()))
  
      new_delegations = tuf.roledb.get_delegated_rolenames(project.project_name)
      self.assertEqual(new_delegations, delegations_backup)
      self.assertEqual(project.layout_type, layout_type_backup)
      self.assertEqual(project.keys, keys_backup)
      self.assertEqual(project('delegation').keys, delegation_keys_backup)
      self.assertEqual(project.prefix, prefix_backup)
      self.assertEqual(project.project_name, name_backup)
  
  
  
  if __name__ == '__main__':
    unittest.main()
  