From ccbadb31d606c5e1d05ed25c954905eebede4130 Mon Sep 17 00:00:00 2001
From: Mike McLean <mikem@redhat.com>
Date: Jan 11 2024 16:21:34 +0000
Subject: [PATCH 1/2] Backup signature headers in delete_rpm_sig


---

diff --git a/kojihub/kojihub.py b/kojihub/kojihub.py
index 54d8b68..d883038 100644
--- a/kojihub/kojihub.py
+++ b/kojihub/kojihub.py
@@ -7954,63 +7954,67 @@ def delete_rpm_sig(rpminfo, sigkey=None, all_sigs=False):
     elif not sigkey:
         raise koji.GenericError("No signature specified")
     rinfo = get_rpm(rpminfo, strict=True)
+    nvra = "%(name)s-%(version)s-%(release)s.%(arch)s" % rinfo
     if rinfo['external_repo_id']:
         raise koji.GenericError("Not an internal rpm: %s (from %s)"
                                 % (rpminfo, rinfo['external_repo_name']))
 
+    # Determine what signature we have
     rpm_query_result = query_rpm_sigs(rpm_id=rinfo['id'], sigkey=sigkey)
     if not rpm_query_result:
-        nvra = "%(name)s-%(version)s-%(release)s.%(arch)s" % rinfo
         raise koji.GenericError("%s has no matching signatures to delete" % nvra)
+    found_keys = [r['sigkey'] for r in rpm_query_result]
 
-    clauses = ["rpm_id=%(rpm_id)i"]
-    if sigkey is not None:
-        clauses.append("sigkey=%(sigkey)s")
+    # Delete signature entries from db
+    clauses = ["rpm_id=%(rpm_id)s", "sigkey IN %(found_keys)s"]
     rpm_id = rinfo['id']
     delete = DeleteProcessor(table='rpmsigs', clauses=clauses, values=locals())
     delete.execute()
     delete = DeleteProcessor(table='rpm_checksum', clauses=clauses, values=locals())
     delete.execute()
-    binfo = get_build(rinfo['build_id'])
+
+    # Get the base build dir for our paths
+    binfo = get_build(rinfo['build_id'], strict=True)
     builddir = koji.pathinfo.build(binfo)
-    list_sigcaches = []
-    list_sighdrs = []
+
+    # Delete signed copies
+    # We do these first since they are the lowest risk
     for rpmsig in rpm_query_result:
-        list_sigcaches.append(joinpath(builddir, koji.pathinfo.sighdr(rinfo, rpmsig['sigkey'])))
-        list_sighdrs.append(joinpath(builddir, koji.pathinfo.signed(rinfo, rpmsig['sigkey'])))
-    list_paths = list_sighdrs + list_sigcaches
-    count = 0
-    logged_user = get_user(context.session.user_id)
-    for file_path in list_paths:
+        signed_path = joinpath(builddir, koji.pathinfo.signed(rinfo, rpmsig['sigkey']))
+        if not os.path.exists(signed_path):
+            # signed copies might not exist
+            continue
         try:
-            os.remove(file_path)
-            count += 1
+            os.remove(signed_path)
+            logger.warning(f"Deleted signed copy {signed_path}")
         except FileNotFoundError:
-            logger.warning("User %s has deleted file %s", logged_user['name'], file_path)
+            # possibly a race?
+            # not fatal -- log and continue
+            logger.error(f"Signed copy disappeared during call: {signed_path}")
         except Exception:
-            logger.error("An error happens when deleting %s, %s deleting are deleted, "
-                         "%s deleting are skipped, the original request is %s rpm "
-                         "and %s sigkey", file_path,
-                         list_paths[:count], list_paths[count:], rpminfo, sigkey, exc_info=True)
-            raise koji.GenericError("File %s cannot be deleted." % file_path)
-
-    for path in list_paths:
-        basedir = os.path.dirname(path)
-        if os.path.isdir(basedir) and not os.listdir(basedir):
-            try:
-                os.rmdir(basedir)
-            except OSError:
-                logger.warning("An error happens when deleting %s directory",
-                               basedir, exc_info=True)
-        sigdir = os.path.dirname(basedir)
-        if os.path.isdir(sigdir) and not os.listdir(sigdir):
-            try:
-                os.rmdir(sigdir)
-            except OSError:
-                logger.warning("An error happens when deleting %s directory",
-                               sigdir, exc_info=True)
-    logger.warning("Signed RPM %s with sigkey %s is deleted by %s", rinfo['id'], sigkey,
-                   logged_user['name'])
+            logger.error(f"Failed to delete {signed_path}", exc_info=True)
+            raise koji.GenericError(f"Failed to delete {signed_path}")
+
+    # Backup header files
+    for rpmsig in rpm_query_result:
+        hdr_path = joinpath(builddir, koji.pathinfo.sighdr(rinfo, rpmsig['sigkey']))
+        backup_path = hdr_path + f".{rpmsig['sighash']}.save"
+        if os.path.exists(backup_path):
+            # Likely residue of previous failed deletion
+            raise koji.GenericError(f"Stray header backup file: {backup_path}")
+        if not os.path.exists(hdr_path):
+            logger.error(f'Missing signature header file: {hdr_path}')
+            # not fatal
+        elif not os.path.isfile(hdr_path):
+            raise koji.GenericError(f"Not a regular file: {hdr_path}")
+        else:
+            os.rename(hdr_path, backup_path)
+            logger.debug(f"Signature header saved to {backup_path}")
+
+    # Note: we do not delete any empty parent dirs as the primary use case for deleting these
+    # signatures is to allow the import of new, overlapping ones
+
+    logger.warning("Deleted signatures %s for rpm %s", found_keys, rinfo['id'])
 
 
 def _scan_sighdr(sighdr, fn):
@@ -8185,7 +8189,7 @@ def write_signed_rpm(an_rpm, sigkey, force=False):
     # make sure we have it in the db
     rpm_id = rinfo['id']
     query = QueryProcessor(tables=['rpmsigs'], columns=['sighash'],
-                           clauses=['rpm_id=%(rpm_id)i', 'sigkey=%(sigkey)s'],
+                           clauses=['rpm_id=%(rpm_id)s', 'sigkey=%(sigkey)s'],
                            values={'rpm_id': rpm_id, 'sigkey': sigkey})
     sighash = query.singleValue(strict=False)
     if not sighash:
diff --git a/tests/test_hub/test_delete_rpm_sig.py b/tests/test_hub/test_delete_rpm_sig.py
index 5840f56..88796ac 100644
--- a/tests/test_hub/test_delete_rpm_sig.py
+++ b/tests/test_hub/test_delete_rpm_sig.py
@@ -1,9 +1,13 @@
+import os
+import tempfile
+import shutil
 import unittest
 
 import mock
 
 import koji
 import kojihub
+from koji.util import joinpath
 
 DP = kojihub.DeleteProcessor
 
@@ -17,6 +21,9 @@ class TestDeleteRPMSig(unittest.TestCase):
         return delete
 
     def setUp(self):
+        self.tempdir = tempfile.mkdtemp()
+        self.pathinfo = koji.PathInfo(self.tempdir)
+        mock.patch('koji.pathinfo', new=self.pathinfo).start()
         self.DeleteProcessor = mock.patch('kojihub.kojihub.DeleteProcessor',
                                           side_effect=self.getDelete).start()
         self.deletes = []
@@ -58,9 +65,31 @@ class TestDeleteRPMSig(unittest.TestCase):
                               'sigkey': '2f86d6a1'}]
         self.userinfo = {'authtype': 2, 'id': 1, 'krb_principal': None, 'krb_principals': [],
                          'name': 'testuser', 'status': 0, 'usertype': 0}
+        self.set_up_files()
+
+    def set_up_files(self):
+        builddir = self.pathinfo.build(self.buildinfo)
+        os.makedirs(builddir)
+        self.builddir = builddir
+        self.signed = {}
+        self.sighdr = {}
+        for sig in self.queryrpmsigs:
+            key = sig['sigkey']
+            signed = joinpath(builddir, self.pathinfo.signed(self.rinfo, key))
+            self.signed[key] = signed
+            koji.ensuredir(os.path.dirname(signed))
+            with open(signed, 'wt') as fo:
+                fo.write('SIGNED COPY\n')
+
+            sighdr = joinpath(builddir, self.pathinfo.sighdr(self.rinfo, key))
+            self.sighdr[key] = sighdr
+            koji.ensuredir(os.path.dirname(sighdr))
+            with open(sighdr, 'wt') as fo:
+                fo.write('DETACHED SIGHDR\n')
 
     def tearDown(self):
         mock.patch.stopall()
+        shutil.rmtree(self.tempdir)
 
     def test_rpm_not_existing(self):
         rpm_id = 1234
@@ -84,7 +113,8 @@ class TestDeleteRPMSig(unittest.TestCase):
 
     def test_external_repo(self):
         rpminfo = 1234
-        rinfo = {'external_repo_id': 1, 'external_repo_name': 'INTERNAL'}
+        rinfo = self.rinfo.copy()
+        rinfo.update({'external_repo_id': 1, 'external_repo_name': 'INTERNAL'})
         self.get_rpm.return_value = rinfo
         with self.assertRaises(koji.GenericError) as ex:
             kojihub.delete_rpm_sig(rpminfo, all_sigs=True)
@@ -108,39 +138,67 @@ class TestDeleteRPMSig(unittest.TestCase):
         self.get_rpm.assert_called_once_with(rpminfo, strict=True)
         self.query_rpm_sigs.assert_called_once_with(rpm_id=self.rinfo['id'], sigkey=None)
 
-    @mock.patch('koji.pathinfo.build', return_value='fakebuildpath')
-    @mock.patch('os.remove')
-    def test_file_not_found_error(self, os_remove, pb):
-        rpminfo = 2
-        os_remove.side_effect = FileNotFoundError()
+    def test_file_not_found_error(self):
+        rpminfo = self.rinfo['id']
         self.get_rpm.return_value = self.rinfo
         self.get_build.return_value = self.buildinfo
         self.get_user.return_value = self.userinfo
         self.query_rpm_sigs.return_value = self.queryrpmsigs
+
+        # a missing signed copy or header should not error
+        builddir = self.pathinfo.build(self.buildinfo)
+        sigkey = '2f86d6a1'
+        os.remove(self.signed[sigkey])
+        os.remove(self.sighdr[sigkey])
         r = kojihub.delete_rpm_sig(rpminfo, sigkey='testkey')
         self.assertEqual(r, None)
 
+        # the files should still be gone
+        for sigkey in self.signed:
+            if os.path.exists(self.signed[sigkey]):
+                raise Exception('signed copy not deleted')
+        for sigkey in self.sighdr:
+            if os.path.exists(self.sighdr[sigkey]):
+                raise Exception('header still in place')
+
         self.assertEqual(len(self.deletes), 2)
         delete = self.deletes[0]
         self.assertEqual(delete.table, 'rpmsigs')
-        self.assertEqual(delete.clauses, ["rpm_id=%(rpm_id)i", "sigkey=%(sigkey)s"])
+        self.assertEqual(delete.clauses, ["rpm_id=%(rpm_id)s", "sigkey IN %(found_keys)s"])
 
         delete = self.deletes[1]
         self.assertEqual(delete.table, 'rpm_checksum')
-        self.assertEqual(delete.clauses, ["rpm_id=%(rpm_id)i", "sigkey=%(sigkey)s"])
+        self.assertEqual(delete.clauses, ["rpm_id=%(rpm_id)s", "sigkey IN %(found_keys)s"])
         self.get_rpm.assert_called_once_with(rpminfo, strict=True)
         self.query_rpm_sigs.assert_called_once_with(rpm_id=self.rinfo['id'], sigkey='testkey')
-        self.get_build.assert_called_once_with(self.rinfo['build_id'])
+        self.get_build.assert_called_once_with(self.rinfo['build_id'], strict=True)
+
+    def test_stray_backup(self):
+        rpminfo = self.rinfo['id']
+        self.get_rpm.return_value = self.rinfo
+        self.get_build.return_value = self.buildinfo
+        self.get_user.return_value = self.userinfo
+        self.query_rpm_sigs.return_value = self.queryrpmsigs
+
+        # a missing signed copy or header should not error
+        siginfo = self.queryrpmsigs[0]
+        sigkey = siginfo['sigkey']
+        backup = "%s.%s.save" % (self.sighdr[sigkey], siginfo['sighash'])
+        with open(backup, 'wt') as fo:
+            fo.write('STRAY FILE\n')
+        with self.assertRaises(koji.GenericError) as ex:
+            r = kojihub.delete_rpm_sig(rpminfo, sigkey='testkey')
+        expected_msg = "Stray header backup file: %s" % backup
+        self.assertEqual(ex.exception.args[0], expected_msg)
 
-    @mock.patch('koji.pathinfo.build', return_value='fakebuildpath')
     @mock.patch('os.remove', side_effect=OSError)
-    def test_not_valid(self, os_remove, pb):
+    def test_not_valid(self, os_remove):
         rpminfo = 2
-        filepath = 'fakebuildpath/data/signed/x86_64/fs_mark-3.3-20.el8.x86_64.rpm'
+        filepath = '%s/packages/fs_mark/3.3/20.el8/data/signed/x86_64/fs_mark-3.3-20.el8.x86_64.rpm' % self.tempdir
         self.get_rpm.return_value = self.rinfo
         self.get_build.return_value = self.buildinfo
         self.query_rpm_sigs.return_value = self.queryrpmsigs
-        expected_msg = "File %s cannot be deleted." % filepath
+        expected_msg = "Failed to delete %s" % filepath
         with self.assertRaises(koji.GenericError) as ex:
             kojihub.delete_rpm_sig(rpminfo, all_sigs=True)
         self.assertEqual(ex.exception.args[0], expected_msg)
@@ -148,18 +206,16 @@ class TestDeleteRPMSig(unittest.TestCase):
         self.assertEqual(len(self.deletes), 2)
         delete = self.deletes[0]
         self.assertEqual(delete.table, 'rpmsigs')
-        self.assertEqual(delete.clauses, ["rpm_id=%(rpm_id)i"])
+        self.assertEqual(delete.clauses, ["rpm_id=%(rpm_id)s", "sigkey IN %(found_keys)s"])
 
         delete = self.deletes[1]
         self.assertEqual(delete.table, 'rpm_checksum')
-        self.assertEqual(delete.clauses, ["rpm_id=%(rpm_id)i"])
+        self.assertEqual(delete.clauses, ["rpm_id=%(rpm_id)s", "sigkey IN %(found_keys)s"])
         self.get_rpm.assert_called_once_with(rpminfo, strict=True)
         self.query_rpm_sigs.assert_called_once_with(rpm_id=self.rinfo['id'], sigkey=None)
-        self.get_build.assert_called_once_with(self.rinfo['build_id'])
+        self.get_build.assert_called_once_with(self.rinfo['build_id'], strict=True)
 
-    @mock.patch('koji.pathinfo.build', return_value='fakebuildpath')
-    @mock.patch('os.remove')
-    def test_valid(self, os_remove, pb):
+    def test_valid(self):
         rpminfo = 2
         self.get_rpm.return_value = self.rinfo
         self.get_build.return_value = self.buildinfo
@@ -167,14 +223,29 @@ class TestDeleteRPMSig(unittest.TestCase):
         self.query_rpm_sigs.return_value = self.queryrpmsigs
         kojihub.delete_rpm_sig(rpminfo, all_sigs=True)
 
+        # the files should be gone
+        for sigkey in self.signed:
+            if os.path.exists(self.signed[sigkey]):
+                raise Exception('signed copy not deleted')
+        for sigkey in self.sighdr:
+            if os.path.exists(self.sighdr[sigkey]):
+                raise Exception('header still in place')
+
+        # the sighdrs should be saved
+        for siginfo in self.queryrpmsigs:
+            sigkey = siginfo['sigkey']
+            backup = "%s.%s.save" % (self.sighdr[sigkey], siginfo['sighash'])
+            with open(backup, 'rt') as fo:
+                self.assertEqual(fo.read(), 'DETACHED SIGHDR\n')
+
         self.assertEqual(len(self.deletes), 2)
         delete = self.deletes[0]
         self.assertEqual(delete.table, 'rpmsigs')
-        self.assertEqual(delete.clauses, ["rpm_id=%(rpm_id)i"])
+        self.assertEqual(delete.clauses, ["rpm_id=%(rpm_id)s", "sigkey IN %(found_keys)s"])
 
         delete = self.deletes[1]
         self.assertEqual(delete.table, 'rpm_checksum')
-        self.assertEqual(delete.clauses, ["rpm_id=%(rpm_id)i"])
+        self.assertEqual(delete.clauses, ["rpm_id=%(rpm_id)s", "sigkey IN %(found_keys)s"])
         self.get_rpm.assert_called_once_with(rpminfo, strict=True)
         self.query_rpm_sigs.assert_called_once_with(rpm_id=self.rinfo['id'], sigkey=None)
-        self.get_build.assert_called_once_with(self.rinfo['build_id'])
+        self.get_build.assert_called_once_with(self.rinfo['build_id'], strict=True)

From 3ac51624f3fbc9cb5b4e34272c2eb16dfb919df8 Mon Sep 17 00:00:00 2001
From: Mike McLean <mikem@redhat.com>
Date: Jan 11 2024 16:21:34 +0000
Subject: [PATCH 2/2] delete_rpm_sig: check header files first


also fixup some error logging and update unit tests

---

diff --git a/kojihub/kojihub.py b/kojihub/kojihub.py
index d883038..543ddf3 100644
--- a/kojihub/kojihub.py
+++ b/kojihub/kojihub.py
@@ -30,6 +30,7 @@ import calendar
 import copy
 import datetime
 import fcntl
+import filecmp
 import fnmatch
 import functools
 import hashlib
@@ -7977,6 +7978,32 @@ def delete_rpm_sig(rpminfo, sigkey=None, all_sigs=False):
     binfo = get_build(rinfo['build_id'], strict=True)
     builddir = koji.pathinfo.build(binfo)
 
+    # Check header files
+    hdr_renames = []
+    hdr_deletes = []
+    for rpmsig in rpm_query_result:
+        hdr_path = joinpath(builddir, koji.pathinfo.sighdr(rinfo, rpmsig['sigkey']))
+        backup_path = hdr_path + f".{rpmsig['sighash']}.save"
+        if not os.path.exists(hdr_path):
+            logger.error(f'Missing signature header file: {hdr_path}')
+            # this doesn't prevent us from deleting the signature
+            # it just means we have nothing to back up
+            continue
+        if not os.path.isfile(hdr_path):
+            # this should not happen and requires human intervention
+            raise koji.GenericError(f"Not a regular file: {hdr_path}")
+        if os.path.exists(backup_path):
+            # Likely residue of previous failed deletion
+            if filecmp.cmp(hdr_path, backup_path, shallow=False):
+                # same file contents, so we're already backed up
+                logger.warning(f"Signature header already backed up: {backup_path}")
+                hdr_deletes.append([rpmsig, hdr_path])
+            else:
+                # this shouldn't happen
+                raise koji.GenericError(f"Stray header backup file: {backup_path}")
+        else:
+            hdr_renames.append([rpmsig, hdr_path, backup_path])
+
     # Delete signed copies
     # We do these first since they are the lowest risk
     for rpmsig in rpm_query_result:
@@ -7987,29 +8014,28 @@ def delete_rpm_sig(rpminfo, sigkey=None, all_sigs=False):
         try:
             os.remove(signed_path)
             logger.warning(f"Deleted signed copy {signed_path}")
-        except FileNotFoundError:
-            # possibly a race?
-            # not fatal -- log and continue
-            logger.error(f"Signed copy disappeared during call: {signed_path}")
         except Exception:
             logger.error(f"Failed to delete {signed_path}", exc_info=True)
             raise koji.GenericError(f"Failed to delete {signed_path}")
 
     # Backup header files
-    for rpmsig in rpm_query_result:
-        hdr_path = joinpath(builddir, koji.pathinfo.sighdr(rinfo, rpmsig['sigkey']))
-        backup_path = hdr_path + f".{rpmsig['sighash']}.save"
-        if os.path.exists(backup_path):
-            # Likely residue of previous failed deletion
-            raise koji.GenericError(f"Stray header backup file: {backup_path}")
-        if not os.path.exists(hdr_path):
-            logger.error(f'Missing signature header file: {hdr_path}')
-            # not fatal
-        elif not os.path.isfile(hdr_path):
-            raise koji.GenericError(f"Not a regular file: {hdr_path}")
-        else:
+    for rpmsig, hdr_path, backup_path in hdr_renames:
+        # sanity checked above
+        try:
             os.rename(hdr_path, backup_path)
-            logger.debug(f"Signature header saved to {backup_path}")
+            logger.warning(f"Signature header saved to {backup_path}")
+        except Exception:
+            logger.error(f"Failed to rename {hdr_path} to {backup_path}", exc_info=True)
+
+    # Delete already backed-up headers
+    for rpmsig, hdr_path in hdr_deletes:
+        # verified backup above
+        try:
+            os.remove(hdr_path)
+            logger.warning(f"Deleted signature header {hdr_path}")
+        except Exception:
+            logger.error(f"Failed to delete {hdr_path}", exc_info=True)
+            raise koji.GenericError(f"Failed to delete {hdr_path}")
 
     # Note: we do not delete any empty parent dirs as the primary use case for deleting these
     # signatures is to allow the import of new, overlapping ones
diff --git a/tests/test_hub/test_delete_rpm_sig.py b/tests/test_hub/test_delete_rpm_sig.py
index 88796ac..907c033 100644
--- a/tests/test_hub/test_delete_rpm_sig.py
+++ b/tests/test_hub/test_delete_rpm_sig.py
@@ -173,6 +173,35 @@ class TestDeleteRPMSig(unittest.TestCase):
         self.query_rpm_sigs.assert_called_once_with(rpm_id=self.rinfo['id'], sigkey='testkey')
         self.get_build.assert_called_once_with(self.rinfo['build_id'], strict=True)
 
+    def test_header_not_a_file(self):
+        rpminfo = self.rinfo['id']
+        self.get_rpm.return_value = self.rinfo
+        self.get_build.return_value = self.buildinfo
+        self.get_user.return_value = self.userinfo
+        self.query_rpm_sigs.return_value = self.queryrpmsigs
+
+        # we should error, without making any changes, if a header is not a regular file
+        builddir = self.pathinfo.build(self.buildinfo)
+        bad_sigkey = '2f86d6a1'
+        bad_hdr= self.sighdr[bad_sigkey]
+        os.remove(bad_hdr)
+        os.mkdir(bad_hdr)
+        with self.assertRaises(koji.GenericError) as ex:
+            r = kojihub.delete_rpm_sig(rpminfo, sigkey='testkey')
+        expected_msg = "Not a regular file: %s" % bad_hdr
+        self.assertEqual(ex.exception.args[0], expected_msg)
+
+        # the files should still be there
+        for sigkey in self.signed:
+            if not os.path.exists(self.signed[sigkey]):
+                raise Exception('signed copy was deleted')
+        for sigkey in self.sighdr:
+            if not os.path.exists(self.sighdr[sigkey]):
+                raise Exception('header was deleted')
+        if not os.path.isdir(bad_hdr):
+            # the function should not have touched the invalid path
+            raise Exception('bad header file was removed')
+
     def test_stray_backup(self):
         rpminfo = self.rinfo['id']
         self.get_rpm.return_value = self.rinfo
@@ -180,16 +209,54 @@ class TestDeleteRPMSig(unittest.TestCase):
         self.get_user.return_value = self.userinfo
         self.query_rpm_sigs.return_value = self.queryrpmsigs
 
-        # a missing signed copy or header should not error
         siginfo = self.queryrpmsigs[0]
         sigkey = siginfo['sigkey']
         backup = "%s.%s.save" % (self.sighdr[sigkey], siginfo['sighash'])
         with open(backup, 'wt') as fo:
             fo.write('STRAY FILE\n')
+            # different contents
         with self.assertRaises(koji.GenericError) as ex:
             r = kojihub.delete_rpm_sig(rpminfo, sigkey='testkey')
         expected_msg = "Stray header backup file: %s" % backup
         self.assertEqual(ex.exception.args[0], expected_msg)
+        # files should not have been removed
+        for sigkey in self.signed:
+            if not os.path.exists(self.signed[sigkey]):
+                raise Exception('signed copy was deleted incorrectly')
+        for sigkey in self.sighdr:
+            if not os.path.exists(self.sighdr[sigkey]):
+                raise Exception('header was deleted incorrectly')
+
+    def test_dup_backup(self):
+        rpminfo = self.rinfo['id']
+        self.get_rpm.return_value = self.rinfo
+        self.get_build.return_value = self.buildinfo
+        self.get_user.return_value = self.userinfo
+        self.query_rpm_sigs.return_value = self.queryrpmsigs
+
+        siginfo = self.queryrpmsigs[0]
+        sigkey = siginfo['sigkey']
+        backup = "%s.%s.save" % (self.sighdr[sigkey], siginfo['sighash'])
+        with open(backup, 'wt') as fo:
+            fo.write('DETACHED SIGHDR\n')
+            # SAME contents
+
+        r = kojihub.delete_rpm_sig(rpminfo, sigkey='testkey')
+
+        # the files should be gone
+        for sigkey in self.signed:
+            if os.path.exists(self.signed[sigkey]):
+                raise Exception('signed copy not deleted')
+        for sigkey in self.sighdr:
+            if os.path.exists(self.sighdr[sigkey]):
+                raise Exception('header still in place')
+
+        # the sighdrs should be saved
+        for siginfo in self.queryrpmsigs:
+            sigkey = siginfo['sigkey']
+            backup = "%s.%s.save" % (self.sighdr[sigkey], siginfo['sighash'])
+            with open(backup, 'rt') as fo:
+                self.assertEqual(fo.read(), 'DETACHED SIGHDR\n')
 
     @mock.patch('os.remove', side_effect=OSError)
     def test_not_valid(self, os_remove):

