This fixes a problem that SQLQueue increases the priority of a message even for conflict errors.

git-svn-id: https://svn.erp5.org/repos/public/erp5/trunk@12021 20353a03-c40f-0410-a6d1-a30d3c3de9de

product/CMFActivity/Activity/SQLQueue.py

@@ -33,7 +33,7 @@ from DateTime import DateTime
 from Queue import VALID, INVALID_ORDER, INVALID_PATH, EXCEPTION, MAX_PROCESSING_TIME, VALIDATION_ERROR_DELAY
 from Products.CMFActivity.ActiveObject import DISTRIBUTABLE_STATE, INVOKE_ERROR_STATE, VALIDATE_ERROR_STATE
 from ZODB.POSException import ConflictError
-from types import StringType
+from types import StringType, ClassType
 import sys
 
 try:
@@ -159,9 +159,15 @@ class SQLQueue(RAMQueue):
           LOG('SQLQueue', WARNING, 'abort failed, thus some objects may be modified accidentally')
           pass
 
-        if line.priority > MAX_PRIORITY:
+        if type(m.exc_type) is ClassType \
+                and issubclass(m.exc_type, ConflictError):
+          activity_tool.SQLQueue_setPriority(uid = line.uid, 
+                                             date = next_processing_date,
+                                             priority = line.priority)
+        elif line.priority > MAX_PRIORITY:
           # This is an error
-          activity_tool.SQLQueue_assignMessage(uid=line.uid, processing_node = INVOKE_ERROR_STATE)
+          activity_tool.SQLQueue_assignMessage(uid = line.uid, 
+                                               processing_node = INVOKE_ERROR_STATE)
                                                                             # Assign message back to 'error' state
           m.notifyUser(activity_tool)                                       # Notify Error
         else:

product/CMFActivity/tests/testCMFActivity.py

@@ -53,6 +53,7 @@ from DateTime import DateTime
 from Acquisition import aq_base, aq_inner
 from zLOG import LOG
 import time
+from ZODB.POSException import ConflictError
 
 try:
   from transaction import get as get_transaction
@@ -642,6 +643,36 @@ class TestCMFActivity(ERP5TypeTestCase):
     self.assertEquals(o.getTitle(), 'a')
     self.assertEquals(portal_activities.countMessageWithTag('toto'), 0)
 
+  def TryConflictErrorsWhileProcessing(self, activity):
+    """Try to execute active objects which may throw conflict errors
+    while processing, and check if they are still executed."""
+    # Make sure that no active object is installed.
+    activity_tool = self.getPortal().portal_activities
+    activity_tool.manageClearActivities(keep=0)
+
+    # Need an object.
+    organisation_module = self.getOrganisationModule()
+    if not organisation_module.hasContent(self.company_id):
+      organisation_module.newContent(id=self.company_id)
+    o = organisation_module._getOb(self.company_id)
+    get_transaction().commit()
+    self.flushAllActivities(silent = 1, loop_size = 10)
+    self.assertEquals(len(activity_tool.getMessageList()), 0)
+
+    # Monkey patch Organisation to induce conflict errors artificially.
+    def induceConflictErrors(self, limit):
+      if self.__class__.current_num_conflict_errors < limit:
+        self.__class__.current_num_conflict_errors += 1
+        raise ConflictError
+    Organisation.induceConflictErrors = induceConflictErrors
+
+    # Test some range of conflict error occurences.
+    for i in xrange(10):
+      Organisation.current_num_conflict_errors = 0
+      o.activate(activity = activity).induceConflictErrors(i)
+      get_transaction().commit()
+      self.flushAllActivities(silent = 1, loop_size = i + 10)
+      self.assertEquals(len(activity_tool.getMessageList()), 0)
 
   def test_01_DeferedSetTitleSQLDict(self, quiet=0, run=run_all_test):
     # Test if we can add a complete sales order
@@ -1409,6 +1440,28 @@ class TestCMFActivity(ERP5TypeTestCase):
     get_transaction().commit()
     self.assertEquals(len(activity_tool.getMessageList()), 0)
 
+  def test_68_TestConflictErrorsWhileProcessingWithSQLDict(self, quiet=0, run=run_all_test):
+    """
+      Test if conflict errors spoil out active objects with SQLDict.
+    """
+    if not run: return
+    if not quiet:
+      message = '\nTest Conflict Errors While Processing With SQLDict'
+      ZopeTestCase._print(message)
+      LOG('Testing... ', 0, message)
+    self.TryConflictErrorsWhileProcessing('SQLDict')
+
+  def test_69_TestConflictErrorsWhileProcessingWithSQLQueue(self, quiet=0, run=run_all_test):
+    """
+      Test if conflict errors spoil out active objects with SQLQueue.
+    """
+    if not run: return
+    if not quiet:
+      message = '\nTest Conflict Errors While Processing With SQLQueue'
+      ZopeTestCase._print(message)
+      LOG('Testing... ', 0, message)
+    self.TryConflictErrorsWhileProcessing('SQLQueue')
+
 
 if __name__ == '__main__':
     framework()