Add validation of existing packages to sync

common/pulp_deb/common/constants.py

@@ -25,6 +25,7 @@
 CONFIG_ALLOWED_KEYS = 'allowed_keys'
 CONFIG_KEYSERVER = 'keyserver'
 CONFIG_KEYSERVER_DEFAULT = 'hkp://wwwkeys.pgp.net'
+CONFIG_REPAIR_SYNC = 'repair_sync'
 
 # Distributor configuration key names
 CONFIG_SERVE_HTTP = 'serve_http'

plugins/pulp_deb/plugins/db/models.py

@@ -285,7 +285,7 @@ def associate(self, repo):
             repository=repo, unit=self)
         return self
 
-    def save_and_associate(self, file_path, repo):
+    def save_and_associate(self, file_path, repo, force=False):
         filename = self.filename_from_unit_key(self.unit_key)
         self.set_storage_path(filename)
         unit = self
@@ -294,6 +294,8 @@ def save_and_associate(self, file_path, repo):
             self.safe_import_content(file_path)
         except NotUniqueError:
             unit = self.__class__.objects.filter(**unit.unit_key).first()
+            if force:
+                self.import_content(file_path)
         unit.associate(repo)
         return unit
 

plugins/pulp_deb/plugins/importers/sync.py

@@ -33,6 +33,19 @@
     ["repo_id", "feed_url", "filename", "checksum_expected", "checksum_actual"])
 
 
+def verify_unit_callback(unit):
+    if not os.path.isfile(unit.storage_path):
+        return False
+    checksums = unit.calculate_deb_checksums(unit.storage_path)
+    if unit.sha1 and unit.sha1 != checksums['sha1']:
+        return False
+    if unit.md5sum and unit.md5sum != checksums['md5sum']:
+        return False
+    if unit.sha256 and unit.sha256 != checksums['sha256']:
+        return False
+    return True
+
+
 class RepoSync(publish_step.PluginStep):
     Type_Class_Map = {
         models.DebPackage.TYPE_ID: models.DebPackage,
@@ -59,6 +72,7 @@ def __init__(self, repo, conduit, config):
         self.components = split_or_none(self.get_config().get('components'))
         self.remove_missing = self.get_config().get_boolean(
             constants.CONFIG_REMOVE_MISSING_UNITS, constants.CONFIG_REMOVE_MISSING_UNITS_DEFAULT)
+        self.repair_sync = self.get_config().get_boolean(constants.CONFIG_REPAIR_SYNC, False)
 
         self.unit_relative_urls = {}
         self.available_units = None
@@ -111,8 +125,12 @@ def __init__(self, repo, conduit, config):
         self.add_child(ParsePackagesStep(constants.SYNC_STEP_PACKAGES_PARSE))
 
         #  packages
+        if self.repair_sync:
+            verify_unit = verify_unit_callback
+        else:
+            verify_unit = None
         self.step_local_units = publish_step.GetLocalUnitsStep(
-            importer_type=ids.TYPE_ID_IMPORTER)
+            importer_type=ids.TYPE_ID_IMPORTER, verify_unit=verify_unit)
         self.add_child(self.step_local_units)
 
         self.add_child(CreateRequestsUnitsToDownload(
@@ -344,7 +362,7 @@ def verify_checksum(self, calculated, from_metadata, path):
     def process_main(self, item=None):
         path_to_unit = self.parent.step_download_units.path_to_unit
         repo = self.get_repo().repo_obj
-        for path, unit in sorted(path_to_unit.items()):
+        for path, unit in path_to_unit.items():
             checksums = unit.calculate_deb_checksums(path)
             if unit.sha1:
                 self.verify_checksum(checksums['sha1'], unit.sha1, path)
@@ -359,7 +377,7 @@ def process_main(self, item=None):
             else:
                 unit.sha256 = checksums['sha256']
             self.verify_checksum(unit.checksum, unit.sha256, path)
-            unit.save_and_associate(path, repo)
+            unit.save_and_associate(path, repo, force=self.parent.repair_sync)
 
 
 class SaveMetadataStep(publish_step.PluginStep):

plugins/test/unit/plugins/importers/test_sync.py

@@ -12,12 +12,14 @@
 
 from pulp_deb.common import constants
 from pulp_deb.common import ids
+from pulp_deb.plugins.db import models
 from pulp_deb.plugins.importers import sync
 
 
 class _TestSyncBase(testbase.TestCase):
     remove_missing = False
     release = None
+    repair_sync = False
 
     def setUp(self):
         super(_TestSyncBase, self).setUp()
@@ -33,6 +35,7 @@ def setUp(self):
             importer_constants.KEY_FEED: 'http://example.com/deb',
             constants.CONFIG_REQUIRE_SIGNATURE: False,
             importer_constants.KEY_UNITS_REMOVE_MISSING: self.remove_missing,
+            constants.CONFIG_REPAIR_SYNC: self.repair_sync,
         }
         if self.release:
             plugin_config['releases'] = self.release
@@ -166,6 +169,39 @@ def test_ParsePackagesStep(self):
             set([x.checksum for x in self.step.available_units]))
         self.assertEquals(len(self.step.component_packages[self.release]['main']), 2)
 
+    @mock.patch('os.path.isfile', return_value=True)
+    @mock.patch('pulp.plugins.util.publish_step.repo_controller.associate_single_unit')
+    @mock.patch('pulp.plugins.util.publish_step.units_controller.find_units')
+    def test_getLocalUnits(self, _find_units, _associate_single_unit, _isfile):
+        self.step.conduit.remove_unit = mock.MagicMock()
+        pkgs = self._mock_repometa()
+        units = [
+            models.DebPackage.from_packages_paragraph(pkg)
+            for pkg in pkgs
+        ]
+        if self.repair_sync:
+            for unit in units:
+                checksums = {
+                    'md5sum': unit.md5sum,
+                    'sha1': unit.sha1,
+                    'sha256': unit.sha256,
+                }
+                unit.calculate_deb_checksums = mock.MagicMock(return_value=checksums)
+            # Break the first one
+            units[0].calculate_deb_checksums.return_value['sha1'] = "invalid"
+        self.step.available_units = units
+        _find_units.return_value = units
+        step = self.step.children[4]
+        step.process_lifecycle()
+        if self.repair_sync:
+            self.assertEqual(_associate_single_unit.call_count, len(units) - 1)
+            self.assertEqual(len(step.units_to_download), 1)
+            self.step.conduit.remove_unit.assert_not_called()
+        else:
+            self.assertEqual(_associate_single_unit.call_count, len(units))
+            self.assertEqual(len(step.units_to_download), 0)
+            self.step.conduit.remove_unit.assert_called_once()
+
     @mock.patch('pulp_deb.plugins.importers.sync.misc.mkdir')
     def test_CreateRequestsUnitsToDownload(self, _mkdir):
         pkgs = self._mock_repometa()
@@ -230,7 +266,7 @@ def test_SaveDownloadedUnits(self):
 
         repo = self.repo.repo_obj
         for path, unit in path_to_unit.items():
-            unit.save_and_associate.assert_called_once_with(path, repo)
+            unit.save_and_associate.assert_called_once_with(path, repo, force=self.repair_sync)
 
     def test_SaveDownloadedUnits_bad_sha256(self):
         self.repo.repo_obj = mock.MagicMock(repo_id=self.repo.id)
@@ -313,3 +349,7 @@ class TestSyncRemoveMissing(_TestSyncBase):
 
 class TestSyncNestedDistribution(_TestSyncBase):
     release = 'stable/updates'
+
+
+class TestSyncRepairSync(_TestSyncBase):
+    repair_sync = True