Merge pull request #1 from theupdateframework/develop

Merge from main branch

Author: FelixWang1994

tests/test_repository_lib.py

@@ -755,14 +755,18 @@ def test_write_metadata_file(self):
                                  compression_algorithms,
                                  consistent_snapshot=False)
 
-    # Try to wirte a consistent metadate file. An exception is not raised in this case.
+    # Try to write a consistent metadate file. An exception is not raised in
+    # this case.  For testing purposes, root.json should be a hard link to the
+    # consistent metadata file.  We should verify that root.json points to
+    # the latest consistent files.
     tuf.conf.CONSISTENT_METHOD = 'hard_link'
     repo_lib.write_metadata_file(root_signable, output_filename,
                                  version_number,
                                  compression_algorithms,
                                  consistent_snapshot=True)
 
-    # Test if the consistent files are properly named (<version number>.rolename.json) or not
+    # Test if the consistent files are properly named
+    # Filename format of a consistent file: <version number>.rolename.json
     version_and_filename = str(version_number) + '.' + 'root.json'
     first_version_output_file = os.path.join(temporary_directory, version_and_filename)
     self.assertTrue(os.path.exists(first_version_output_file))
@@ -775,23 +779,34 @@ def test_write_metadata_file(self):
                                  consistent_snapshot=True)
 
     # Test if the the latest root.json points to the expected consistent file
-    # and
-    # consistent metadata do not all point to the same root.json
+    # and consistent metadata do not all point to the same root.json
     version_and_filename = str(version_number) + '.' + 'root.json'
     second_version_output_file = os.path.join(temporary_directory, version_and_filename)
     self.assertTrue(os.path.exists(second_version_output_file))
     self.assertNotEqual(os.stat(output_filename).st_ino, os.stat(first_version_output_file).st_ino)
     self.assertEqual(os.stat(output_filename).st_ino, os.stat(second_version_output_file).st_ino)
 
-    # Test improper Consistent_METHOD configuration
+    # Test for an improper tuf.conf.CONSISTENT_METHOD string value.
     tuf.conf.CONSISTENT_METHOD = 'somebadidea'
     self.assertRaises(tuf.InvalidConfigurationError, repo_lib.write_metadata_file,
                                                      root_signable, output_filename,
                                                      version_number,
                                                      compression_algorithms,
                                                      consistent_snapshot=True)
 
-    # Test unknown compression algorithm.
+    # Try to create a link to root.json when root.json doesn't exist locally.
+    # repository_lib should log a message if this is the case.
+    tuf.conf.CONSISTENT_METHOD = 'hard_link'
+    os.remove(output_filename)
+    repo_lib.write_metadata_file(root_signable, output_filename,
+                                 version_number,
+                                 compression_algorithms,
+                                 consistent_snapshot=True)
+    
+    # Reset CONSISTENT_METHOD so that subsequent tests work as expected.
+    tuf.conf.CONSISTENT_METHOD = 'copy'
+    
+    # Test for unknown compression algorithm.
     self.assertRaises(tuf.FormatError, repo_lib.write_metadata_file,
                                        root_signable, output_filename,
                                        version_number,

tests/test_repository_tool.py

@@ -194,9 +194,13 @@ def test_writeall(self):
 
 
     # (3) Load top-level signing keys.
+    repository.status()
     repository.root.load_signing_key(root_privkey)
+    repository.status()
     repository.targets.load_signing_key(targets_privkey)
+    repository.status()
     repository.snapshot.load_signing_key(snapshot_privkey)
+    repository.status()
 
     # Verify that repository.writeall() fails for insufficient threshold
     # of signatures (default threshold = 1).
@@ -343,6 +347,19 @@ def test_writeall(self):
     # successfully.
     repo_tool.load_repository(repository_directory)
 
+    # Verify the behavior of marking and unmarking roles as dirty.
+    # We begin by ensuring that writeall() cleared the list of dirty roles.. 
+    self.assertEqual([], tuf.roledb.get_dirty_roles())
+    
+    repository.mark_dirty(['root', 'timestamp'])
+    self.assertEqual(['root', 'timestamp'], sorted(tuf.roledb.get_dirty_roles()))
+    repository.unmark_dirty(['root'])
+    self.assertEqual(['timestamp'], tuf.roledb.get_dirty_roles())
+    
+    # Ensure status() does not leave behind any dirty roles.
+    repository.status()
+    self.assertEqual(['timestamp'], tuf.roledb.get_dirty_roles())
+
     # Test improperly formatted arguments.
     self.assertRaises(tuf.FormatError, repository.writeall, 3, False)
     self.assertRaises(tuf.FormatError, repository.writeall, False, 3)
@@ -1538,9 +1555,26 @@ def test_load_repository(self):
     temporary_directory = tempfile.mkdtemp(dir=self.temporary_directory) 
     original_repository_directory = os.path.join('repository_data',
                                                  'repository')
+    
     repository_directory = os.path.join(temporary_directory, 'repository')
+    metadata_directory = os.path.join(repository_directory, 'metadata.staged')
     shutil.copytree(original_repository_directory, repository_directory)
-     
+   
+    # For testing purposes, add a metadata file with an extension that is
+    # not supported, and another with invalid JSON content.
+    invalid_metadata_file = os.path.join(metadata_directory, 'root.xml')
+    root_file = os.path.join(metadata_directory, 'root.json')
+    shutil.copyfile(root_file, invalid_metadata_file)
+    bad_root_content = os.path.join(metadata_directory, 'root_bad.json') 
+    
+    with open(bad_root_content, 'wb') as file_object:
+      file_object.write(b'bad') 
+
+    # Remove the compressed version of role1 to test whether the
+    # load_repository() complains or not (it logs a message).
+    role1_path = os.path.join(metadata_directory, 'role1.json.gz')
+    os.remove(role1_path)
+    
     repository = repo_tool.load_repository(repository_directory)
     self.assertTrue(isinstance(repository, repo_tool.Repository))
 
@@ -1557,8 +1591,8 @@ def test_load_repository(self):
     self.assertTrue(len(repository.timestamp.keys))
     self.assertEqual(1, repository.targets('role1').version)
 
-    # Assumed the targets (tuf/tests/repository_data/) role contains 'file1.txt'
-    # and 'file2.txt'.
+    # It is assumed that the targets (tuf/tests/repository_data/) role contains
+    # 'file1.txt' and 'file2.txt'.
     self.assertTrue('/file1.txt' in repository.targets.target_files)
     self.assertTrue('/file2.txt' in repository.targets.target_files)
     self.assertTrue('/file3.txt' in repository.targets('role1').target_files)

tests/test_roledb.py

@@ -701,6 +701,36 @@ def test_mark_dirty(self):
     self.assertRaises(tuf.InvalidNameError, tuf.roledb.mark_dirty,
                       ['dirty_role'], 'non-existent')
 
+  
+  
+  def test_unmark_dirty(self):
+    # Add a dirty role to roledb.
+    rolename = 'targets'
+    roleinfo1 = {'keyids': ['123'], 'threshold': 1}
+    tuf.roledb.add_role(rolename, roleinfo1)
+    rolename2 = 'dirty_role'
+    roleinfo2 = {'keyids': ['123'], 'threshold': 2}
+    tuf.roledb.add_role(rolename2, roleinfo2)
+    mark_role_as_dirty = True
+    tuf.roledb.update_roleinfo(rolename, roleinfo1, mark_role_as_dirty)
+    # Note: The 'default' repository is searched if the repository name is
+    # not given to get_dirty_roles().
+    self.assertEqual([rolename], tuf.roledb.get_dirty_roles())
+    tuf.roledb.update_roleinfo(rolename2, roleinfo2, mark_role_as_dirty)
+    
+    tuf.roledb.unmark_dirty(['dirty_role'])
+    self.assertEqual([rolename], sorted(tuf.roledb.get_dirty_roles()))
+    tuf.roledb.unmark_dirty(['targets'])
+    self.assertEqual([], sorted(tuf.roledb.get_dirty_roles()))
+
+    # What happens for a role that isn't dirty?  unmark_dirty() should just
+    # log a message.
+    tuf.roledb.unmark_dirty(['unknown_role'])
+
+    # Verify that a role cannot be unmarked as dirty for a non-existent
+    # repository.
+    self.assertRaises(tuf.InvalidNameError, tuf.roledb.unmark_dirty,
+                      ['dirty_role'], 'non-existent')
 
 
   def _test_rolename(self, test_function):

tests/test_sig.py

@@ -177,6 +177,16 @@ def test_get_signature_status_single_key(self):
     self.assertEqual([], sig_status['unknown_method_sigs'])
 
     self.assertTrue(tuf.sig.verify(signable, 'Root'))
+   
+    # Test for an unknown signature when 'role' is left unspecified.
+    sig_status = tuf.sig.get_signature_status(signable)
+    
+    self.assertEqual(0, sig_status['threshold'])
+    self.assertEqual([], sig_status['good_sigs'])
+    self.assertEqual([], sig_status['bad_sigs'])
+    self.assertEqual([KEYS[0]['keyid']], sig_status['unknown_sigs'])
+    self.assertEqual([], sig_status['untrusted_sigs'])
+    self.assertEqual([], sig_status['unknown_method_sigs'])
 
     # Done.  Let's remove the added key(s) from the key database.
     tuf.keydb.remove_key(KEYS[0]['keyid'])
@@ -281,6 +291,9 @@ def test_get_signature_status_below_threshold_unauthorized_sigs(self):
     self.assertEqual([], sig_status['unknown_method_sigs'])
 
     self.assertFalse(tuf.sig.verify(signable, 'Root'))
+    
+    self.assertRaises(tuf.UnknownRoleError,
+                      tuf.sig.get_signature_status, signable, 'unknown_role')
 
     # Done.  Let's remove the added key(s) from the key database.
     tuf.keydb.remove_key(KEYS[0]['keyid'])