Prevent concurrent import with an advisory lock

It uses the new OCA/connector#138, OCA/connector#139 feature.

magentoerpconnect/tests/test_synchronization.py

@@ -18,11 +18,23 @@
 #    along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 ##############################################################################
+import mock
 
-from openerp.addons.connector.exception import InvalidDataError
+from openerp import api
+from openerp.modules.registry import RegistryManager
+from openerp.tests.common import get_db_name
+from openerp.addons.connector.connector import ConnectorEnvironment
+from openerp.addons.connector.exception import (
+    InvalidDataError,
+    RetryableJobError,
+)
+from openerp.addons.connector.session import ConnectorSession
 from openerp.addons.magentoerpconnect.unit.import_synchronizer import (
     import_batch,
     import_record)
+from openerp.addons.magentoerpconnect.product_category import (
+    ProductCategoryImporter,
+)
 from .common import (mock_api,
                      mock_urlopen_image,
                      SetUpMagentoBase,
@@ -264,3 +276,43 @@ def test_import_sale_order_with_discount(self):
             else:
                 self.fail('encountered unexpected sale '
                           'order line %s' % line.name)
+
+
+class TestImportMagentoConcurrentSync(SetUpMagentoSynchronized):
+
+    def setUp(self):
+        super(TestImportMagentoConcurrentSync, self).setUp()
+        self.registry2 = RegistryManager.get(get_db_name())
+        self.cr2 = self.registry2.cursor()
+        self.env2 = api.Environment(self.cr2, self.env.uid, {})
+        backend2 = mock.Mock(name='Backend Record')
+        backend2._name = 'magento.backend'
+        backend2.id = self.backend_id
+        self.backend2 = backend2
+        self.connector_session2 = ConnectorSession.from_env(self.env2)
+
+        @self.addCleanup
+        def reset_cr2():
+            # rollback and close the cursor, and reset the environments
+            self.env2.reset()
+            self.cr2.rollback()
+            self.cr2.close()
+
+    def test_concurrent_import(self):
+        connector_env = ConnectorEnvironment(
+            self.backend,
+            self.session,
+            'magento.product.category'
+        )
+        importer = ProductCategoryImporter(connector_env)
+        with mock_api(magento_base_responses):
+            importer.run(1)
+
+        connector_env2 = ConnectorEnvironment(
+            self.backend2,
+            self.connector_session2,
+            'magento.product.category'
+        )
+        importer2 = ProductCategoryImporter(connector_env2)
+        with self.assertRaises(RetryableJobError):
+            importer2.run(1)

magentoerpconnect/unit/import_synchronizer.py

@@ -193,6 +193,15 @@ def run(self, magento_id, force=False):
         :param magento_id: identifier of the record on Magento
         """
         self.magento_id = magento_id
+        lock_name = 'import({}, {}, {}, {})'.format(
+            self.backend_record._name,
+            self.backend_record.id,
+            self.model._name,
+            magento_id,
+        )
+        # Keep a lock on this import until the transaction is committed
+        self.advisory_lock_or_retry(lock_name)
+
         try:
             self.magento_record = self._get_magento_data()
         except IDMissingInBackend: