##############################################################################
#
# Copyright (c) 2004 Nexedi SARL and Contributors. All Rights Reserved.
# Sebastien Robin <seb@nexedi.com>
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsability of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# garantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
import six
import inspect
import warnings
from functools import wraps
from itertools import product
from AccessControl.SecurityManagement import getSecurityManager
from AccessControl.SecurityManagement import setSecurityManager
from AccessControl.SecurityManagement import newSecurityManager
from Acquisition import aq_base, aq_parent
from Products.ERP5Type.tests.utils import LogInterceptor
from Testing import ZopeTestCase
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from Products.ERP5Type.tests.utils import createZODBPythonScript
from Products.ERP5Type.Base import Base
from Products.CMFActivity import ActivityTool
from Products.CMFActivity.Activity.SQLBase import INVOKE_ERROR_STATE
from Products.CMFActivity.Activity.Queue import VALIDATION_ERROR_DELAY
from Products.CMFActivity.Activity.SQLDict import SQLDict
from Products.CMFActivity.Errors import ActivityPendingError, ActivityFlushError
from Products.PluggableAuthService.PropertiedUser import PropertiedUser
from erp5.portal_type import Organisation
from AccessControl.SecurityManagement import newSecurityManager
from zLOG import LOG
from ZODB.POSException import ConflictError
from DateTime import DateTime
from Products.CMFActivity.ActivityTool import (
cancelProcessShutdown, Message, getCurrentNode, getServerAddress)
from MySQLdb import OperationalError
from Products.ZMySQLDA.db import DB
import gc
import random
import threading
import weakref
import transaction
from App.config import getConfiguration
import socket
class CommitFailed(Exception):
pass
def for_each_activity(wrapped):
def wrapper(self):
getMessageList = self.portal.portal_activities.getMessageList
for activity in ActivityTool.activity_dict:
wrapped(self, activity)
self.abort()
self.assertFalse([
x.__dict__ for x in getMessageList()
])
return wraps(wrapped)(wrapper)
def registerFailingTransactionManager(*args, **kw):
from Shared.DC.ZRDB.TM import TM
class dummy_tm(TM):
def tpc_vote(self, *ignored):
raise CommitFailed
def _finish(self):
pass
def _abort(self):
pass
dummy_tm()._register()
class LockOnce(object):
def __init__(self):
self.acquire = threading.Lock().acquire
def release(self):
pass
class TestCMFActivity(ERP5TypeTestCase, LogInterceptor):
# Different variables used for this test
company_id = 'Nexedi'
title1 = 'title1'
title2 = 'title2'
company_id2 = 'Coramy'
company_id3 = 'toto'
def getTitle(self):
return "CMFActivity"
def getBusinessTemplateList(self):
"""
Return the list of business templates.
"""
return ('erp5_base', 'erp5_joblib')
def getOrganisationModule(self):
return self.portal.organisation_module
def getOrganisation(self):
return self.getOrganisationModule()._getOb(self.company_id)
def afterSetUp(self):
super(TestCMFActivity, self).afterSetUp()
from Products.CMFActivity.ActivityRuntimeEnvironment import BaseMessage
# Set 'max_retry' to a known value so that we can test the feature
BaseMessage.max_retry = property(lambda self:
self.activity_kw.get('max_retry', 5))
self.login()
# Then add new components
organisation_module = self.getOrganisationModule()
if not(organisation_module.hasContent(self.company_id)):
o1 = organisation_module.newContent(id=self.company_id)
self.tic()
def tearDown(self):
# Override ERP5 tearDown to make sure that tests do not leave unprocessed
# activity messages. We are testing CMFActivity so it's important to check
# that everything works as expected on this subject.
try:
if self._resultForDoCleanups.wasSuccessful():
getMessageList = self.portal.portal_activities.getMessageList
self.assertFalse(getMessageList())
# Also check if a test drop them without committing.
self.abort()
self.assertFalse(getMessageList())
finally:
ERP5TypeTestCase.tearDown(self)
def getMessageList(self, activity, **kw):
return ActivityTool.activity_dict[activity].getMessageList(
self.portal.portal_activities, **kw)
def deleteMessageList(self, activity, message_list):
ActivityTool.activity_dict[activity].deleteMessageList(
self.portal.portal_activities.getSQLConnection(),
[m.uid for m in message_list])
self.commit()
def login(self):
uf = self.portal.acl_users
uf._doAddUser('seb', '', ['Manager'], [])
uf._doAddUser('ERP5TypeTestCase', '', ['Manager'], [])
user = uf.getUserById('seb').__of__(uf)
newSecurityManager(None, user)
def ticOnce(self, *args, **kw):
is_running_lock = ActivityTool.is_running_lock
try:
ActivityTool.is_running_lock = LockOnce()
self.portal.portal_activities.tic(*args, **kw)
finally:
ActivityTool.is_running_lock = is_running_lock
@for_each_activity
def testInvokeAndCancelActivity(self, activity):
"""
Simple test where we invoke and cancel an activity
"""
activity_tool = self.portal.portal_activities
organisation = self.getOrganisation()
organisation._setTitle(self.title1)
self.assertEqual(self.title1,organisation.getTitle())
organisation.activate(activity=activity)._setTitle(self.title2)
# Needed so that the message are commited into the queue
self.commit()
message_list = activity_tool.getMessageList()
self.assertEqual(len(message_list),1)
activity_tool.manageCancel(organisation.getPhysicalPath(),'_setTitle')
# Needed so that the message are removed from the queue
self.commit()
self.assertEqual(self.title1,organisation.getTitle())
message_list = activity_tool.getMessageList()
self.assertEqual(len(message_list),0)
organisation.activate(activity=activity)._setTitle(self.title2)
# Needed so that the message are commited into the queue
self.commit()
message_list = activity_tool.getMessageList()
self.assertEqual(len(message_list),1)
activity_tool.manageInvoke(organisation.getPhysicalPath(),'_setTitle')
# Needed so that the message are removed from the queue
self.commit()
self.assertEqual(self.title2,organisation.getTitle())
@for_each_activity
def testDeferredSetTitleActivity(self, activity):
"""
We check that the title is changed only after that
the activity was called
"""
activity_tool = self.portal.portal_activities
organisation = self.getOrganisation()
organisation._setTitle(self.title1)
self.assertEqual(self.title1,organisation.getTitle())
organisation.activate(activity=activity)._setTitle(self.title2)
# Needed so that the message are commited into the queue
self.commit()
self.assertEqual(self.title1,organisation.getTitle())
activity_tool.tic()
self.assertEqual(self.title2,organisation.getTitle())
@for_each_activity
def testCallOnceWithActivity(self, activity):
"""
With this test we can check if methods are called
only once (sometimes it was twice !!!)
"""
activity_tool = self.portal.portal_activities
def setFoobar(self):
if hasattr(self,'foobar'):
self.foobar = self.foobar + 1
else:
self.foobar = 1
def getFoobar(self):
return (getattr(self,'foobar',0))
organisation = self.getOrganisation()
Organisation.setFoobar = setFoobar
Organisation.getFoobar = getFoobar
organisation.foobar = 0
organisation._setTitle(self.title1)
self.assertEqual(0,organisation.getFoobar())
organisation.activate(activity=activity).setFoobar()
# Needed so that the message are commited into the queue
self.commit()
message_list = activity_tool.getMessageList()
self.assertEqual(len(message_list),1)
activity_tool.tic()
self.assertEqual(1,organisation.getFoobar())
message_list = activity_tool.getMessageList()
self.assertEqual(len(message_list),0)
organisation.activate(activity=activity).setFoobar()
# Needed so that the message are commited into the queue
self.commit()
message_list = activity_tool.getMessageList()
self.assertEqual(len(message_list),1)
activity_tool.manageInvoke(organisation.getPhysicalPath(),'setFoobar')
# Needed so that the message are commited into the queue
self.commit()
self.assertEqual(2,organisation.getFoobar())
@for_each_activity
def testTryFlushActivity(self, activity):
"""
Check the method flush
"""
organisation = self.getOrganisation()
organisation._setTitle(self.title1)
organisation.activate(activity=activity)._setTitle(self.title2)
organisation.flushActivity(invoke=1)
self.assertEqual(organisation.getTitle(),self.title2)
self.commit()
message_list = self.portal.portal_activities.getMessageList()
self.assertEqual(len(message_list),0)
self.assertEqual(organisation.getTitle(),self.title2)
# Try again with different commit order
organisation._setTitle(self.title1)
organisation.activate(activity=activity)._setTitle(self.title2)
self.commit()
organisation.flushActivity(invoke=1)
self.assertEqual(len(message_list),0)
self.assertEqual(organisation.getTitle(),self.title2)
self.commit()
@for_each_activity
def testTryActivateInsideFlush(self, activity):
"""
Create a new activity inside a flush action
"""
activity_tool = self.portal.portal_activities
def DeferredSetTitle(self,value):
self.activate(activity=activity)._setTitle(value)
Organisation.DeferredSetTitle = DeferredSetTitle
organisation = self.getOrganisation()
organisation._setTitle(self.title1)
organisation.activate(activity=activity).DeferredSetTitle(self.title2)
organisation.flushActivity(invoke=1)
self.commit()
activity_tool.tic()
self.commit()
self.assertEqual(organisation.getTitle(),self.title2)
@for_each_activity
def testTryTwoMethods(self, activity):
"""
Try several activities
"""
activity_tool = self.portal.portal_activities
def DeferredSetDescription(self,value):
self._setDescription(value)
def DeferredSetTitle(self,value):
self._setTitle(value)
Organisation.DeferredSetTitle = DeferredSetTitle
Organisation.DeferredSetDescription = DeferredSetDescription
organisation = self.getOrganisation()
organisation._setTitle(None)
organisation.setDescription(None)
organisation.activate(activity=activity).DeferredSetTitle(self.title1)
organisation.activate(activity=activity).DeferredSetDescription(self.title1)
self.commit()
activity_tool.distribute()
activity_tool.tic()
self.commit()
self.assertEqual(organisation.getTitle(),self.title1)
self.assertEqual(organisation.getDescription(),self.title1)
self.tic()
@for_each_activity
def testTryTwoMethodsAndFlushThem(self, activity):
"""
make sure flush works with several activities
"""
activity_tool = self.portal.portal_activities
def DeferredSetTitle(self,value):
self.activate(activity=activity)._setTitle(value)
def DeferredSetDescription(self,value):
self.activate(activity=activity)._setDescription(value)
Organisation.DeferredSetTitle = DeferredSetTitle
Organisation.DeferredSetDescription = DeferredSetDescription
organisation = self.getOrganisation()
organisation._setTitle(None)
organisation.setDescription(None)
organisation.activate(activity=activity).DeferredSetTitle(self.title1)
organisation.activate(activity=activity).DeferredSetDescription(self.title1)
organisation.flushActivity(invoke=1)
self.commit()
activity_tool.distribute()
activity_tool.tic()
self.commit()
self.assertEqual(organisation.getTitle(),self.title1)
self.assertEqual(organisation.getDescription(),self.title1)
def TryActivateFlushActivateTic(self, activity,second=None,commit_sub=0):
"""
try to commit sub transactions
"""
activity_tool = self.portal.portal_activities
def DeferredSetTitle(self,value,commit_sub=0):
if commit_sub:
transaction.savepoint(optimistic=True)
self.activate(activity=second or activity,priority=4)._setTitle(value)
def DeferredSetDescription(self,value,commit_sub=0):
if commit_sub:
transaction.savepoint(optimistic=True)
self.activate(activity=second or activity,priority=4)._setDescription(value)
Organisation.DeferredSetTitle = DeferredSetTitle
Organisation.DeferredSetDescription = DeferredSetDescription
organisation = self.getOrganisation()
organisation._setTitle(None)
organisation.setDescription(None)
organisation.activate(activity=activity).DeferredSetTitle(self.title1,commit_sub=commit_sub)
organisation.flushActivity(invoke=1)
organisation.activate(activity=activity).DeferredSetDescription(self.title1,commit_sub=commit_sub)
self.commit()
activity_tool.distribute()
activity_tool.tic()
self.commit()
self.assertEqual(organisation.getTitle(),self.title1)
self.assertEqual(organisation.getDescription(),self.title1)
@for_each_activity
def testTryMessageWithErrorOnActivity(self, activity):
"""
Make sure that message with errors are not deleted
"""
activity_tool = self.portal.portal_activities
def crashThisActivity(self):
self.IWillCrash()
organisation = self.getOrganisation()
Organisation.crashThisActivity = crashThisActivity
organisation.activate(activity=activity).crashThisActivity()
# Needed so that the message are commited into the queue
self.commit()
message_list = activity_tool.getMessageList()
LOG('Before MessageWithErrorOnActivityFails, message_list',0,[x.__dict__ for x in message_list])
self.assertEqual(len(message_list),1)
activity_tool.tic()
# XXX HERE WE SHOULD USE TIME SHIFT IN ORDER TO SIMULATE MULTIPLE TICS
# Test if there is still the message after it crashed
message_list = activity_tool.getMessageList()
self.assertEqual(len(message_list),1)
activity_tool.manageCancel(organisation.getPhysicalPath(),'crashThisActivity')
# Needed so that the message are commited into the queue
self.commit()
@for_each_activity
def testDeferredSetTitleWithRenamedObject(self, activity):
"""
make sure that it is impossible to rename an object
if some activities are still waiting for this object
"""
organisation = self.getOrganisation()
organisation._setTitle(self.title1)
self.assertEqual(self.title1,organisation.getTitle())
organisation.activate(activity=activity)._setTitle(self.title2)
# Needed so that the message are commited into the queue
self.commit()
self.assertEqual(self.title1,organisation.getTitle())
self.assertRaises(ActivityPendingError,organisation.edit,id=self.company_id2)
self.portal.portal_activities.tic()
def TryActiveProcess(self, activity):
"""
Try to store the result inside an active process
"""
activity_tool = self.portal.portal_activities
organisation = self.getOrganisation()
organisation._setTitle(self.title1)
active_process = activity_tool.newActiveProcess()
self.assertEqual(self.title1,organisation.getTitle())
organisation.activate(activity=activity,active_process=active_process).getTitle()
# Needed so that the message are commited into the queue
self.commit()
activity_tool.distribute()
activity_tool.tic()
self.assertEqual(self.title1,organisation.getTitle())
result = active_process.getResultList()[0]
self.assertEqual(result.method_id , 'getTitle')
self.assertEqual(result.result , self.title1)
# Execute any further activity which may have been spawned by activity
# execution (ex: fulltext indeation of the active process).
self.tic()
def TryActiveProcessWithResultDict(self, activity):
"""
Try to store the result inside an active process using result list
"""
activity_tool = self.portal.portal_activities
organisation = self.getOrganisation()
organisation._setTitle(self.title1)
active_process = activity_tool.newActiveProcess()
self.assertEqual(self.title1,organisation.getTitle())
# Post SQLjoblib tasks with explicit signature
organisation.activate(activity=activity,active_process=active_process, signature=1).getTitle()
organisation.activate(activity=activity,active_process=active_process, signature=2).getTitle()
organisation.activate(activity=activity,active_process=active_process, signature=3).getTitle()
self.commit()
activity_tool.distribute()
activity_tool.tic()
result_dict = active_process.getResultDict()
result = result_dict[1]
self.assertEqual(result_dict[1].method_id, 'getTitle')
self.assertEqual(result.result , self.title1)
result = result_dict[2]
self.assertEqual(result_dict[2].method_id, 'getTitle')
self.assertEqual(result.result , self.title1)
result = result_dict[3]
self.assertEqual(result_dict[3].method_id, 'getTitle')
self.assertEqual(result.result , self.title1)
# Execute any further activity which may have been spawned by activity
# execution (ex: fulltext indeation of the active process).
self.tic()
@for_each_activity
def testTryMethodAfterMethod(self, activity):
"""
Ensure the order of an execution by a method id
"""
o = self.getOrganisation()
o.setTitle('a')
self.assertEqual(o.getTitle(), 'a')
self.tic()
def toto(self, value):
self.setTitle(self.getTitle() + value)
o.__class__.toto = toto
def titi(self, value):
self.setTitle(self.getTitle() + value)
o.__class__.titi = titi
o.activate(after_method_id = 'titi', activity = activity).toto('b')
o.activate(activity = activity).titi('c')
self.tic()
self.assertEqual(o.getTitle(), 'acb')
@for_each_activity
def testTryAfterTag(self, activity):
"""
Ensure the order of an execution by a tag
"""
o = self.getOrganisation()
o.setTitle('?')
self.assertEqual(o.getTitle(), '?')
self.tic()
o.activate(after_tag = 'toto', activity = activity).setTitle('b')
o.activate(tag = 'toto', activity = activity).setTitle('a')
self.tic()
self.assertEqual(o.getTitle(), 'b')
o.setDefaultActivateParameterDict({'tag': 'toto'})
def titi(self):
self.setCorporateName(self.getTitle() + 'd')
o.__class__.titi = titi
o.activate(after_tag_and_method_id=('toto', 'setTitle'), activity = activity).titi()
o.activate(activity = activity).setTitle('c')
self.tic()
self.assertEqual(o.getCorporateName(), 'cd')
@for_each_activity
def testTryFlushActivityWithAfterTag(self, activity):
"""
Ensure the order of an execution by a tag
"""
o = self.getOrganisation()
o.setTitle('?')
o.setDescription('?')
self.assertEqual(o.getTitle(), '?')
self.assertEqual(o.getDescription(), '?')
self.tic()
o.activate(after_tag = 'toto', activity = activity).setDescription('b')
o.activate(tag = 'toto', activity = activity).setTitle('a')
self.commit()
tool = self.getActivityTool()
self.assertRaises(ActivityFlushError,tool.manageInvoke,o.getPath(),'setDescription')
tool.manageInvoke(o.getPath(),'setTitle')
self.commit()
self.assertEqual(o.getTitle(), 'a')
self.assertEqual(o.getDescription(), '?')
self.tic()
self.assertEqual(o.getTitle(), 'a')
self.assertEqual(o.getDescription(), 'b')
@for_each_activity
def testScheduling(self, activity):
"""
Check if active objects with different after parameters are executed in a correct order
"""
o = self.getOrganisation()
o.setTitle('?')
self.assertEqual(o.getTitle(), '?')
self.tic()
def toto(self, s):
self.setTitle(self.getTitle() + s)
o.__class__.toto = toto
o.activate(tag = 'toto', activity = activity).toto('a')
self.commit()
o.activate(after_tag = 'titi', activity = activity).toto('b')
self.commit()
o.activate(tag = 'titi', after_tag = 'toto', activity = activity).setTitle('c')
self.tic()
self.assertEqual(o.getTitle(), 'cb')
@for_each_activity
def testSchedulingAfterTagList(self, activity):
"""
Check if active objects with different after parameters are executed in a
correct order, when after_tag is passed as a list
"""
o = self.getOrganisation()
o.setTitle('')
self.tic()
def toto(self, s):
self.setTitle(self.getTitle() + s)
o.__class__.toto = toto
o.activate(tag='A', activity=activity).toto('a')
self.commit()
o.activate(tag='B', activity=activity).toto('b')
self.commit()
o.activate(after_tag=('A', 'B'), activity=activity).setTitle('last')
self.tic()
self.assertEqual(o.getTitle(), 'last')
@for_each_activity
def testCheckCountMessageWithTag(self, activity):
"""
Check countMessageWithTag function.
"""
activity_tool = self.portal.portal_activities
o = self.getOrganisation()
o.setTitle('?')
self.tic()
o.activate(tag = 'toto', activity = activity).setTitle('a')
self.commit()
self.assertEqual(o.getTitle(), '?')
self.assertEqual(activity_tool.countMessageWithTag('toto'), 1)
self.tic()
self.assertEqual(o.getTitle(), 'a')
self.assertEqual(activity_tool.countMessageWithTag('toto'), 0)
def testTryErrorsWhileFinishingCommitDB(self):
"""Try to execute active objects which may throw conflict errors
while validating, and check if they are still executed."""
activity_tool = self.portal.portal_activities
# Monkey patch Queue to induce conflict errors artificially.
def query(self, query_string,*args, **kw):
# Not so nice, this is specific to zsql method
if "REPLACE INTO" in query_string:
raise OperationalError
return self.original_query(query_string,*args, **kw)
# Test some range of conflict error occurences.
self.portal.organisation_module.reindexObject()
self.commit()
message, = activity_tool.getMessageList()
try:
DB.original_query = DB.query
DB.query = query
activity_tool.distribute()
activity_tool.tic()
self.commit()
finally:
DB.query = DB.original_query
del DB.original_query
self.deleteMessageList('SQLDict', [message])
@for_each_activity
def testIsMessageRegisteredMethod(self, activity):
dedup = activity != 'SQLQueue'
activity_tool = self.portal.portal_activities
object_b = self.getOrganisation()
object_a = object_b.getParentValue()
def check(count):
self.commit()
self.assertEqual(len(activity_tool.getMessageList()), count)
self.tic()
# First case: creating the same activity twice must only register one
# for queues with deduplication.
object_a.activate(activity=activity).getId()
object_a.activate(activity=activity).getId()
check(1 if dedup else 2)
# Second case: creating activity with same tag must only register one,
# for queues with deduplication.
# This behaviour is actually the same as the no-tag behaviour.
object_a.activate(activity=activity, tag='foo').getId()
object_a.activate(activity=activity, tag='foo').getId()
check(1 if dedup else 2)
# Third case: creating activities with different tags must register both.
object_a.activate(activity=activity, tag='foo').getId()
object_a.activate(activity=activity, tag='bar').getId()
check(2)
# Fourth case: creating activities on different objects must register
# both.
object_a.activate(activity=activity).getId()
object_b.activate(activity=activity).getId()
check(2)
# Fifth case: creating activities with different method must register
# both.
object_a.activate(activity=activity).getId()
object_a.activate(activity=activity).getTitle()
check(2)
def test_33_TryActivateFlushActivateTicWithSQLDict(self):
# Test if we call methods only once
self.TryActivateFlushActivateTic('SQLDict')
def test_34_TryActivateFlushActivateTicWithSQLQueue(self):
# Test if we call methods only once
self.TryActivateFlushActivateTic('SQLQueue')
def test_37_TryActivateFlushActivateTicWithMultipleActivities(self):
# Test if we call methods only once
self.TryActivateFlushActivateTic('SQLQueue',second='SQLDict')
self.TryActivateFlushActivateTic('SQLDict',second='SQLQueue')
def test_38_TryCommitSubTransactionWithSQLDict(self):
# Test if we call methods only once
self.TryActivateFlushActivateTic('SQLDict',commit_sub=1)
def test_39_TryCommitSubTransactionWithSQLQueue(self):
# Test if we call methods only once
self.TryActivateFlushActivateTic('SQLQueue',commit_sub=1)
def test_46_TryActiveProcessWithSQLDict(self):
# Test if we call methods only once
self.TryActiveProcess('SQLDict')
def test_47_TryActiveProcessWithSQLQueue(self):
# Test if we call methods only once
self.TryActiveProcess('SQLQueue')
def test_48_TryActiveProcessWithSQLJoblib(self):
# Test if we call methods only once
self.TryActiveProcessWithResultDict('SQLJoblib')
def test_57_TryCallActivityWithRightUser(self):
# Test if me execute methods with the right user
# This should be independant of the activity used
# We are first logged as seb
activity_tool = self.portal.portal_activities
organisation = self.getOrganisation()
# Add new user toto
uf = self.portal.acl_users
uf._doAddUser('toto', '', ['Manager'], [])
user = uf.getUserById('toto').__of__(uf)
newSecurityManager(None, user)
# Execute something as toto
organisation.activate().newContent(portal_type='Email',id='email')
# Then execute activities as seb
user = uf.getUserById('seb').__of__(uf)
newSecurityManager(None, user)
self.tic()
email = organisation.get('email')
# Check if what we did was executed as toto
self.assertEqual(email.getOwnerInfo()['id'],'toto')
def flushAllActivities(self, silent=0, loop_size=1000):
"""Executes all messages until the queue only contains failed
messages.
"""
activity_tool = self.portal.portal_activities
for _ in xrange(loop_size):
activity_tool.distribute(node_count=1)
activity_tool.tic(processing_node=1)
finished = all(message.processing_node == INVOKE_ERROR_STATE
for message in activity_tool.getMessageList())
activity_tool.timeShift(3 * VALIDATION_ERROR_DELAY)
self.commit()
if finished:
return
if not silent:
self.fail('flushAllActivities maximum loop count reached')
def test_68_TestMessageValidationAndFailedActivities(self):
"""after_method_id and failed activities.
Tests that if we have an active method scheduled by
after_method_id and a failed activity with this method id, the
method is NOT executed.
Note: earlier version of this test checked exactly the contrary, but it
was eventually agreed that this was a bug. If an activity fails, all the
activities that depend on it should be block until the first one is
resolved."""
activity_tool = self.portal.portal_activities
original_title = 'something'
obj = self.portal.organisation_module.newContent(
portal_type='Organisation',
title=original_title)
# Monkey patch Organisation to add a failing method
def failingMethod(self):
raise ValueError('This method always fail')
Organisation.failingMethod = failingMethod
for activity in ActivityTool.activity_dict:
# reset
activity_tool.manageClearActivities()
obj.setTitle(original_title)
self.commit()
# activate failing message and flush
for fail_activity in ActivityTool.activity_dict:
obj.activate(activity = fail_activity).failingMethod()
self.commit()
self.flushAllActivities(silent=1, loop_size=100)
full_message_list = activity_tool.getMessageList()
remaining_messages = [a for a in full_message_list if a.method_id !=
'failingMethod']
if len(full_message_list) != 3:
self.fail('failingMethod should not have been flushed')
if len(remaining_messages) != 0:
self.fail('Activity tool should have no other remaining messages')
# activate our message
new_title = 'nothing'
obj.activate(after_method_id = ['failingMethod'],
activity = activity ).setTitle(new_title)
self.commit()
self.flushAllActivities(silent=1, loop_size=100)
full_message_list = activity_tool.getMessageList()
remaining_messages = [a for a in full_message_list if a.method_id !=
'failingMethod']
self.assertEqual(len(full_message_list), 4,
'failingMethod should not have been flushed')
self.assertEqual(len(remaining_messages), 1,
'Activity tool should have one blocked setTitle activity')
self.assertEqual(remaining_messages[0].activity_kw['after_method_id'],
['failingMethod'])
self.assertEqual(obj.getTitle(), original_title)
activity_tool.manageClearActivities()
self.commit()
def test_70_TestCancelFailedActiveObject(self):
"""Cancel an active object to make sure that it does not refer to
a persistent object.
"""
activity_tool = self.portal.portal_activities
original_title = 'something'
obj = self.portal.organisation_module.newContent(
portal_type='Organisation',
title=original_title)
# Monkey patch Organisation to add a failing method
def failingMethod(self):
raise ValueError('This method always fail')
Organisation.failingMethod = failingMethod
# First, index the object.
self.commit()
self.flushAllActivities(silent=1, loop_size=100)
self.assertEqual(len(activity_tool.getMessageList()), 0)
# Insert a failing active object.
obj.activate().failingMethod()
self.commit()
self.assertEqual(len(activity_tool.getMessageList()), 1)
# Just wait for the active object to be abandoned.
self.flushAllActivities(silent=1, loop_size=100)
self.assertEqual(len(activity_tool.getMessageList()), 1)
self.assertEqual(activity_tool.getMessageList()[0].processing_node,
INVOKE_ERROR_STATE)
# Make sure that persistent objects are not present in the connection
# cache to emulate a restart of Zope. So all volatile attributes will
# be flushed, and persistent objects will be reloaded.
activity_tool._p_jar._resetCache()
# Cancel it via the management interface.
message = activity_tool.getMessageList()[0]
activity_tool.manageCancel(message.object_path, message.method_id)
self.commit()
def test_71_RetryMessageExecution(self):
activity_tool = self.portal.portal_activities
exec_count = [0]
# priority does not matter anymore
priority = random.Random().randint
def doSomething(self, retry_list):
i = exec_count[0]
exec_count[0] = i + 1
conflict, edit_kw = retry_list[i]
if edit_kw:
self.getActivityRuntimeEnvironment().edit(**edit_kw)
if conflict is not None:
raise ConflictError if conflict else Exception
def check(retry_list, **activate_kw):
fail = retry_list[-1][0] is not None and 1 or 0
for activity in ActivityTool.activity_dict:
exec_count[0] = 0
activity_tool.activate(activity=activity, priority=priority(1,6),
**activate_kw).doSomething(retry_list)
self.commit()
self.flushAllActivities(silent=1)
self.assertEqual(len(retry_list), exec_count[0])
self.assertEqual(fail, len(activity_tool.getMessageList()))
activity_tool.manageCancel(
activity_tool.getPhysicalPath(), 'doSomething')
self.commit()
activity_tool.__class__.doSomething = doSomething
try:
## Default behaviour
# Usual successful case: activity is run only once
check([(None, None)])
# Usual error case: activity is run 6 times before being frozen
check([(False, None)] * 6)
# On ConflictError, activity is reexecuted without increasing retry count
check([(True, None)] * 10 + [(None, None)])
check([(True, None), (False, None)] * 6)
## Customized behaviour
# Do not retry
check([(False, {'max_retry': 0})])
# ... even in case of ConflictError
check([(True, {'max_retry': 0}),
(True, {'max_retry': 0, 'conflict_retry': 0})])
check([(True, None)] * 6, conflict_retry=False)
# Customized number of retries
for n in 3, 9:
check([(False, {'max_retry': n})] * n + [(None, None)])
check([(False, {'max_retry': n})] * (n + 1))
# Infinite retry
for n in 3, 9:
check([(False, {'max_retry': None})] * n + [(None, None)])
check([(False, {'max_retry': None})] * n + [(False, {'max_retry': 0})])
check([(False, {'max_retry': None})] * 9 + [(False, None)])
finally:
del activity_tool.__class__.doSomething
def test_79_ActivateKwForNewContent(self):
o1 = self.getOrganisationModule().newContent(
activate_kw=dict(tag='The Tag'))
self.commit()
m, = self.getActivityTool().getMessageList(path=o1.getPath())
self.assertEqual(m.activity_kw.get('tag'), 'The Tag')
self.tic()
def test_80_FlushAfterMultipleActivate(self):
orga_module = self.getOrganisationModule()
p = orga_module.newContent(portal_type='Organisation')
self.tic()
self.assertEqual(p.getDescription(), "")
activity_tool = self.portal.portal_activities
def updateDesc(self):
d =self.getDescription()
self.setDescription(d+'a')
Organisation.updateDesc = updateDesc
# First check dequeue read same message only once
for i in xrange(10):
p.activate(activity="SQLDict").updateDesc()
self.commit()
self.assertEqual(len(activity_tool.getMessageList()), 10)
self.tic()
self.assertEqual(p.getDescription(), "a")
# Check if there is pending activity after deleting an object
for i in xrange(10):
p.activate(activity="SQLDict").updateDesc()
self.commit()
self.assertEqual(len(activity_tool.getMessageList()), 10)
activity_tool.flush(p, invoke=0)
self.commit()
@for_each_activity
def testCallWithGroupIdParamater(self, activity):
dedup = activity != 'SQLQueue'
activity_tool = self.portal.portal_activities
organisation = self.getOrganisation()
# Defined a group method
foobar_list = []
def setFoobar(self, object_list):
foobar_list.append(len(object_list))
for m in object_list:
obj = m.object
obj.foobar += m.kw.get('number', 1)
m.result = None
from Products.ERP5Type.Core.Folder import Folder
Folder.setFoobar = setFoobar
Organisation.getFoobar = lambda self: self.foobar
organisation.foobar = 0
self.assertEqual(0,organisation.getFoobar())
# Test group_method_id is working without group_id
for x in xrange(5):
organisation.activate(activity=activity, group_method_id="organisation_module/setFoobar").reindexObject(number=1)
self.commit()
message_list = activity_tool.getMessageList()
self.assertEqual(len(message_list),5)
activity_tool.tic()
expected = 1 if dedup else 5
self.assertEqual(expected, organisation.getFoobar())
# Test group_method_id is working with one group_id defined
for x in xrange(5):
organisation.activate(activity=activity, group_method_id="organisation_module/setFoobar", group_id="1").reindexObject(number=1)
self.commit()
message_list = activity_tool.getMessageList()
self.assertEqual(len(message_list),5)
activity_tool.tic()
self.assertEqual(expected * 2, organisation.getFoobar())
self.assertEqual([expected, expected], foobar_list)
del foobar_list[:]
# Test group_method_id is working with many group_id defined
for x in xrange(5):
organisation.activate(activity=activity, group_method_id="organisation_module/setFoobar", group_id="1").reindexObject(number=1)
self.commit()
organisation.activate(activity=activity, group_method_id="organisation_module/setFoobar", group_id="2").reindexObject(number=3)
self.commit()
organisation.activate(activity=activity, group_method_id="organisation_module/setFoobar", group_id="1").reindexObject(number=1)
self.commit()
organisation.activate(activity=activity, group_method_id="organisation_module/setFoobar", group_id="3").reindexObject(number=5)
self.commit()
message_list = activity_tool.getMessageList()
self.assertEqual(len(message_list),20)
activity_tool.tic()
self.assertEqual(11 if dedup else 60,
organisation.getFoobar())
self.assertEqual([1, 1, 1] if dedup else [5, 5, 10],
sorted(foobar_list))
def test_84_ActivateKwForWorkflowTransition(self):
"""
Test call of a workflow transition with activate_kw parameter propagate them
"""
o1 = self.getOrganisationModule().newContent()
self.tic()
o1.validate(activate_kw=dict(tag='The Tag'))
self.commit()
m, = self.getActivityTool().getMessageList(path=o1.getPath())
self.assertEqual(m.activity_kw.get('tag'), 'The Tag')
self.tic()
def test_85_LossOfVolatileAttribute(self):
"""
Test that the loss of volatile attribute doesn't loose activities
"""
activity_tool = self.getActivityTool()
def delete_volatiles():
for property_id in activity_tool.__dict__.keys():
if property_id.startswith('_v_'):
delattr(activity_tool, property_id)
organisation_module = self.getOrganisationModule()
active_organisation_module = organisation_module.activate()
delete_volatiles()
# Cause a message to be created
# If the buffer cannot be created, this will raise
active_organisation_module.getTitle()
delete_volatiles()
# Another activity to check that first one did not get lost even if volatile disapears
active_organisation_module.getId()
self.commit()
message_list = activity_tool.getMessageList()
self.assertEqual(len(message_list), 2)
self.tic()
def test_88_ProcessingMultipleMessagesMustRevertIndividualMessagesOnError(self):
"""
Check that, on queues which support it, processing a batch of multiple
messages doesn't cause failed ones to becommited along with succesful
ones.
Queues supporting message batch processing:
- SQLQueue
"""
activity_tool = self.getActivityTool()
obj = self.portal.organisation_module.newContent(portal_type='Organisation')
active_obj = obj.activate(activity='SQLQueue')
def appendToTitle(self, to_append, fail=False):
self.setTitle(self.getTitle() + to_append)
if fail:
raise ValueError('This method always fail')
try:
Organisation.appendToTitle = appendToTitle
obj.setTitle('a')
active_obj.appendToTitle('b')
active_obj.appendToTitle('c', fail=True)
active_obj.appendToTitle('d')
object_id = obj.getId()
self.commit()
self.assertEqual(obj.getTitle(), 'a')
self.assertEqual(activity_tool.countMessage(method_id='appendToTitle'), 3)
self.flushAllActivities(silent=1, loop_size=100)
self.assertEqual(sorted(obj.getTitle()), ['a', 'b', 'd'])
message, = self.getMessageList('SQLQueue', method_id='appendToTitle')
self.deleteMessageList('SQLQueue', [message])
finally:
del Organisation.appendToTitle
def test_89_RequestIsolationInsideSameTic(self):
"""
Check that request information do not leak from one activity to another
inside the same TIC invocation.
This only apply to queues supporting batch processing:
- SQLQueue
"""
obj = self.portal.organisation_module.newContent(portal_type='Organisation', title='Pending')
marker_id = 'marker_%i' % (random.randint(1, 10), )
def putMarkerValue(self, marker_id):
self.REQUEST.set(marker_id, 1)
def checkMarkerValue(self, marker_id):
if self.REQUEST.get(marker_id) is not None:
self.setTitle('Failed')
else:
self.setTitle('Success')
try:
Organisation.putMarkerValue = putMarkerValue
Organisation.checkMarkerValue = checkMarkerValue
obj.activate(activity='SQLQueue', tag='set_first').putMarkerValue(marker_id=marker_id)
obj.activate(activity='SQLQueue', after_tag='set_first').checkMarkerValue(marker_id=marker_id)
self.assertEqual(obj.getTitle(), 'Pending')
self.tic()
self.assertEqual(obj.getTitle(), 'Success')
finally:
del Organisation.putMarkerValue
del Organisation.checkMarkerValue
def test_globalrequest(self):
"""zope.globalrequest.getRequest (also known as Products.Global.get_request)
should be same as app.REQUEST, also when executing activities.
"""
from zope.globalrequest import getRequest
get_request_before = getRequest()
def checkRequest(active_self):
self.assertIs(getRequest(), active_self.REQUEST)
obj = self.portal.organisation_module.newContent(portal_type='Organisation')
Organisation.checkRequest = checkRequest
try:
obj.activate(activity='SQLQueue').checkRequest()
obj.activate(activity='SQLDict').checkRequest()
self.tic()
finally:
del Organisation.checkRequest
self.assertIs(getRequest(), get_request_before)
@for_each_activity
def testTryUserNotificationOnActivityFailure(self, activity):
message_list = self.portal.MailHost._message_list
del message_list[:]
portal_activities = self.portal.portal_activities
countMessage = portal_activities.countMessage
obj = self.portal.organisation_module.newContent(portal_type='Organisation')
self.tic()
def failingMethod(self): raise ValueError('This method always fails')
Organisation.failingMethod = failingMethod
try:
portal_activities.activity_failure_mail_notification = True
# MESSAGE_NOT_EXECUTED
obj.activate(activity=activity).failingMethod()
self.commit()
self.assertFalse(message_list)
self.flushAllActivities(silent=1, loop_size=100)
# Check there is a traceback in the email notification
sender, recipients, mail = message_list.pop()
self.assertIn("Module %s, line %s, in failingMethod" % (
__name__, inspect.getsourcelines(failingMethod)[1]), mail)
self.assertIn("ValueError:", mail)
portal_activities.manageClearActivities()
# MESSAGE_NOT_EXECUTABLE
obj_path = obj.getPath()
obj.activate(activity=activity).failingMethod()
self.commit()
obj.getParentValue()._delObject(obj.getId())
self.commit()
self.assertGreater(countMessage(path=obj_path), 0)
self.tic()
self.assertEqual(countMessage(path=obj_path), 0)
self.assertFalse(message_list)
finally:
self.portal.portal_activities.activity_failure_mail_notification = True
del Organisation.failingMethod
@for_each_activity
def testTryUserNotificationDisabledOnActivityFailure(self, activity):
message_list = self.portal.MailHost._message_list
del message_list[:]
portal_activities = self.portal.portal_activities
countMessage = portal_activities.countMessage
obj = self.portal.organisation_module.newContent(portal_type='Organisation')
self.tic()
def failingMethod(self): raise ValueError('This method always fails')
Organisation.failingMethod = failingMethod
try:
portal_activities.activity_failure_mail_notification = False
# MESSAGE_NOT_EXECUTED
obj.activate(activity=activity).failingMethod()
self.commit()
self.assertFalse(message_list)
self.flushAllActivities(silent=1, loop_size=100)
# Check there is a traceback in the email notification
self.assertFalse(message_list)
portal_activities.manageClearActivities()
# MESSAGE_NOT_EXECUTABLE
obj_path = obj.getPath()
obj.activate(activity=activity).failingMethod()
self.commit()
obj.getParentValue()._delObject(obj.getId())
self.commit()
self.assertGreater(countMessage(path=obj_path), 0)
self.tic()
self.assertEqual(countMessage(path=obj_path), 0)
self.assertFalse(message_list)
finally:
portal_activities.activity_failure_mail_notification = True
del Organisation.failingMethod
def test_93_tryUserNotificationRaise(self):
activity_tool = self.portal.portal_activities
obj = self.portal.organisation_module.newContent(portal_type='Organisation')
self.tic()
original_notifyUser = Message.notifyUser
def failingMethod(self, *args, **kw):
raise ValueError('This method always fail')
Message.notifyUser = failingMethod
Organisation.failingMethod = failingMethod
try:
for activity in ActivityTool.activity_dict:
obj.activate(activity=activity, priority=6).failingMethod()
self.commit()
self.flushAllActivities(silent=1, loop_size=100)
message, = activity_tool.getMessageList(
activity=activity, method_id='failingMethod')
self.assertEqual(message.processing_node, INVOKE_ERROR_STATE)
self.assertTrue(message.retry)
activity_tool.manageDelete(message.uid, activity)
self.commit()
finally:
Message.notifyUser = original_notifyUser
del Organisation.failingMethod
@for_each_activity
def testTryActivityRaiseInCommitDoesNotStallActivityConection(self, activity):
"""
Check that an activity which commit raises (as would a regular conflict
error be raised in tpc_vote) does not cause activity connection to
stall.
"""
try:
Organisation.registerFailingTransactionManager = registerFailingTransactionManager
obj = self.portal.organisation_module.newContent(portal_type='Organisation')
self.tic()
now = DateTime()
obj.activate(activity=activity).registerFailingTransactionManager()
self.commit()
self.flushAllActivities(silent=1, loop_size=100)
self.commit()
# Check that cmf_activity SQL connection still works
connection_da = self.portal.cmf_activity_sql_connection()
self.assertFalse(connection_da._registered)
connection_da.query('select 1')
self.assertTrue(connection_da._registered)
self.commit()
self.assertFalse(connection_da._registered)
message, = self.getMessageList(activity)
self.deleteMessageList(activity, [message])
finally:
del Organisation.registerFailingTransactionManager
@for_each_activity
def testTryActivityRaiseInCommitDoesNotLoseMessages(self, activity):
"""
"""
try:
Organisation.registerFailingTransactionManager = registerFailingTransactionManager
obj = self.portal.organisation_module.newContent(portal_type='Organisation')
self.tic()
now = DateTime()
obj.activate(activity=activity).registerFailingTransactionManager()
self.commit()
self.flushAllActivities(silent=1, loop_size=100)
self.commit()
message, = self.getMessageList(activity,
method_id='registerFailingTransactionManager')
self.deleteMessageList(activity, [message])
finally:
del Organisation.registerFailingTransactionManager
@for_each_activity
def testTryChangeSkinInActivity(self, activity):
activity_tool = self.getActivityTool()
def changeSkinToNone(self):
self.getPortalObject().changeSkin(None)
Organisation.changeSkinToNone = changeSkinToNone
try:
organisation = self.portal.organisation_module.newContent(portal_type='Organisation')
self.tic()
organisation.activate(activity=activity).changeSkinToNone()
self.commit()
self.assertEqual(len(activity_tool.getMessageList()), 1)
self.flushAllActivities(silent=1, loop_size=100)
finally:
del Organisation.changeSkinToNone
@for_each_activity
def testDeduplicatingQueuesDoNotDeleteSimilaritiesBeforeExecution(self,
activity):
"""
Test that SQLDict does not delete similar messages which have the same
method_id and path but a different tag before execution.
"""
if activity == 'SQLQueue':
return
activity_tool = self.getActivityTool()
marker = []
def doSomething(self, other_tag):
marker.append(self.countMessage(tag=other_tag))
activity_tool.__class__.doSomething = doSomething
try:
# Adds two similar but not the same activities.
activity_tool.activate(activity=activity, after_tag='foo',
tag='a').doSomething(other_tag='b')
activity_tool.activate(activity=activity, after_tag='bar',
tag='b').doSomething(other_tag='a')
self.commit()
activity_tool.tic() # make sure distribution phase was not skipped
activity_tool.distribute()
# after distribute, similarities are still there.
self.assertEqual(len(self.getMessageList(activity)), 2)
activity_tool.tic()
self.assertEqual(marker, [1])
finally:
del activity_tool.__class__.doSomething
@for_each_activity
def testDeduplicatingQueuesDoNotDeleteDuplicatesBeforeExecution(self,
activity):
"""
Test that SQLDict does not delete messages before execution
even if messages have the same method_id and path and tag.
There could be other things which differ (ex: serialization_tag) and may
not all be cheap to check during validation. Validation node is the only
non-paralelisable Zope-side task around activities, so it should be kept
simple.
Deduplication is cheap:
- inside the transaction which spawned duplicate activities, because it
has to have created activities around anyway, and can keep track
- inside the CMFActivity-level processing surrounding activity execution
because it has to load the activities to process them anyway
"""
if activity == 'SQLQueue':
return
activity_tool = self.getActivityTool()
# Adds two same activities.
activity_tool.activate(activity=activity, after_tag='foo', priority=2,
tag='a').getId()
self.commit()
uid1, = [x.uid for x in self.getMessageList(activity)]
activity_tool.activate(activity=activity, after_tag='bar', priority=1,
tag='a').getId()
self.commit()
uid2, = [x.uid for x in self.getMessageList(activity) if x.uid != uid1]
self.assertEqual(len(activity_tool.getMessageList()), 2)
activity_tool.distribute()
# After distribute, duplicate is still present.
self.assertItemsEqual([uid1, uid2],
[x.uid for x in self.getMessageList(activity)])
activity_tool.tic()
def testCheckSQLDictDistributeWithSerializationTagAndGroupMethodId(self):
"""
Distribuation was at some point buggy with this scenario when there was
activate with the same serialization_tag and one time with a group_method
id and one without group_method_id :
foo.activate(serialization_tag='a', group_method_id='x').getTitle()
foo.activate(serialization_tag='a').getId()
"""
def getMessageList():
return [
(x.activity_kw['serialization_tag'], x.processing_node)
for x in activity_tool.getMessageList()
]
def activate(serialization_tag='a'):
organisation.activate(
serialization_tag=serialization_tag,
group_method_id='portal_catalog/catalogObjectList',
).getTitle()
organisation = self.portal.organisation_module.newContent(portal_type='Organisation')
self.tic()
activity_tool = self.getActivityTool()
activate('a')
self.commit()
activate('a')
self.commit()
# Both activities are queued
self.assertItemsEqual(
getMessageList(),
[
('a', -1),
('a', -1),
],
)
activity_tool.distribute()
# Both activities are validated at the same time.
# Note: this specific test implmeentation relies on the absence of
# validation-time deduplication which is not strictly related to
# serialization_tag behaviour.
self.assertItemsEqual(
getMessageList(),
[
('a', 0),
('a', 0),
],
)
activate('a')
self.commit()
activate('b')
self.commit()
# 3rd & 4th activities queued
self.assertItemsEqual(
getMessageList(),
[
('a', 0),
('a', 0),
('a', -1),
('b', -1),
],
)
activity_tool.distribute()
# 3rd activity does not get validated, 4th is validated
self.assertItemsEqual(
getMessageList(),
[
('a', 0),
('a', 0),
('a', -1),
('b', 0),
],
)
# 1st, 2nd and 4th are executed, then 3rd gets validated an executed,
# and the queue ends empty.
self.tic()
def test_104_interQueuePriorities(self):
"""
Important note: there is no way to really reliably check that this
feature is correctly implemented, as activity execution order is
non-deterministic.
The best which can be done is to check that under certain circumstances
the activity exeicution order match expectations.
"""
organisation = self.portal.organisation_module.newContent(portal_type='Organisation')
self.tic()
activity_tool = self.getActivityTool()
check_result_dict = {}
def runAndCheck():
check_result_dict.clear()
self.commit()
self.assertEqual(len(check_result_dict), 0)
self.tic()
self.assertEqual(len(check_result_dict), 2)
self.assertTrue(check_result_dict['before_ran'])
self.assertTrue(check_result_dict['after_ran'])
def mustRunBefore(self):
check_result_dict['before_ran'] = 'after_ran' not in check_result_dict
def mustRunAfter(self):
check_result_dict['after_ran'] = 'before_ran' in check_result_dict
Organisation.mustRunBefore = mustRunBefore
Organisation.mustRunAfter = mustRunAfter
try:
# Check that ordering looks good (SQLQueue first)
organisation.activate(activity='SQLQueue', priority=1).mustRunBefore()
organisation.activate(activity='SQLDict', priority=2).mustRunAfter()
runAndCheck()
# Check that ordering looks good (SQLDict first)
organisation.activate(activity='SQLDict', priority=1).mustRunBefore()
organisation.activate(activity='SQLQueue', priority=2).mustRunAfter()
runAndCheck()
# Check that tag takes precedence over priority (SQLQueue first by priority)
organisation.activate(activity='SQLQueue', priority=1, after_tag='a').mustRunAfter()
organisation.activate(activity='SQLDict', priority=2, tag='a').mustRunBefore()
runAndCheck()
# Check that tag takes precedence over priority (SQLDict first by priority)
organisation.activate(activity='SQLDict', priority=1, after_tag='a').mustRunAfter()
organisation.activate(activity='SQLQueue', priority=2, tag='a').mustRunBefore()
runAndCheck()
finally:
del Organisation.mustRunBefore
del Organisation.mustRunAfter
@for_each_activity
def testCheckActivityRuntimeEnvironment(self, activity):
document = self.portal.organisation_module
activity_result = []
def extractActivityRuntimeEnvironment(self):
activity_result.append(self.getActivityRuntimeEnvironment())
document.__class__.doSomething = extractActivityRuntimeEnvironment
try:
document.activate(activity=activity).doSomething()
self.commit()
# Check that getActivityRuntimeEnvironment raises outside of activities
self.assertRaises(KeyError, document.getActivityRuntimeEnvironment)
# Check Runtime isolation
self.tic()
# Check that it still raises outside of activities
self.assertRaises(KeyError, document.getActivityRuntimeEnvironment)
# Check activity runtime environment instance
env = activity_result.pop()
self.assertFalse(activity_result)
message = env._message
self.assertEqual(message.line.priority, 1)
self.assertEqual(message.object_path, document.getPhysicalPath())
self.assertTrue(message.conflict_retry) # default value
env.edit(max_retry=0, conflict_retry=False)
self.assertFalse(message.conflict_retry) # edited value
self.assertRaises(AttributeError, env.edit, foo='bar')
finally:
del document.__class__.doSomething
@for_each_activity
def testSerializationTag(self, activity):
organisation = self.portal.organisation_module.newContent(portal_type='Organisation')
self.tic()
activity_tool = self.getActivityTool()
# First scenario: activate, distribute, activate, distribute
# Create first activity and distribute: it must be distributed
organisation.activate(activity=activity, serialization_tag='1').getTitle()
self.commit()
result = activity_tool.getMessageList()
self.assertEqual(len(result), 1)
activity_tool.distribute()
result = activity_tool.getMessageList()
self.assertEqual(len([x for x in result if x.processing_node == 0]), 1)
# Create second activity and distribute: it must *NOT* be distributed
organisation.activate(activity=activity, serialization_tag='1').getTitle()
self.commit()
result = activity_tool.getMessageList()
self.assertEqual(len(result), 2)
activity_tool.distribute()
result = activity_tool.getMessageList()
self.assertEqual(len([x for x in result if x.processing_node == 0]), 1) # Distributed message list len is still 1
self.tic()
# Second scenario: activate, activate, distribute
# Both messages must be distributed (this is different from regular tags)
organisation.activate(activity=activity, serialization_tag='1', priority=2).getTitle()
# Use a different method just so that SQLDict doesn't merge both activities prior to insertion.
organisation.activate(activity=activity, serialization_tag='1', priority=1).getId()
self.commit()
result = activity_tool.getMessageList()
self.assertEqual(len(result), 2)
activity_tool.distribute()
result = activity_tool.getMessageList()
# at most 1 activity for a given serialization tag can be validated
message, = [x for x in result if x.processing_node == 0]
self.assertEqual(message.method_id, 'getId')
# the other one is still waiting for validation
message, = [x for x in result if x.processing_node == -1]
self.assertEqual(message.method_id, 'getTitle')
self.tic()
# Check that giving a None value to serialization_tag does not confuse
# CMFActivity
organisation.activate(activity=activity, serialization_tag=None).getTitle()
self.tic()
def test_110_testAbsoluteUrl(self):
# Tests that absolute_url works in activities. The URL generation is based
# on REQUEST information when the method was activated.
request = self.portal.REQUEST
request.setServerURL('http', 'test.erp5.org', '9080')
request.other['PARENTS'] = [self.portal.organisation_module]
request.setVirtualRoot('virtual_root')
calls = []
def checkAbsoluteUrl(self):
calls.append(self.absolute_url())
Organisation.checkAbsoluteUrl = checkAbsoluteUrl
try:
o = self.portal.organisation_module.newContent(
portal_type='Organisation', id='test_obj')
self.assertEqual(o.absolute_url(),
'http://test.erp5.org:9080/virtual_root/test_obj')
o.activate().checkAbsoluteUrl()
# Reset server URL and virtual root before executing messages.
# This simulates the case of activities beeing executed with different
# REQUEST, such as TimerServer.
request.setServerURL('https', 'anotherhost.erp5.org', '443')
request.other['PARENTS'] = [self.app]
request.setVirtualRoot('')
# obviously, the object url is different
self.assertEqual(o.absolute_url(),
'https://anotherhost.erp5.org/%s/organisation_module/test_obj'
% self.portal.getId())
# but activities are executed using the previous request information
self.flushAllActivities(loop_size=1000)
self.assertEqual(calls, ['http://test.erp5.org:9080/virtual_root/test_obj'])
finally:
del Organisation.checkAbsoluteUrl
def CheckLocalizerWorks(self, activity):
FROM_STRING = 'Foo'
TO_STRING = 'Bar'
LANGUAGE = 'xx'
def translationTest(context):
from Products.ERP5Type.Message import Message
context.setTitle(context.Base_translateString(FROM_STRING))
context.setDescription(str(Message('erp5_ui', FROM_STRING)))
portal = self.portal
portal.Localizer.erp5_ui.manage_addLanguage(LANGUAGE)
# Add FROM_STRING to the message catalog
portal.Localizer.erp5_ui.gettext(FROM_STRING)
# ...and translate it.
portal.Localizer.erp5_ui.message_edit(message=FROM_STRING,
language=LANGUAGE, translation=TO_STRING, note='')
organisation = portal.organisation_module.newContent(
portal_type='Organisation')
self.tic()
Organisation.translationTest = translationTest
try:
REQUEST = organisation.REQUEST
# Simulate what a browser would have sent to Zope
REQUEST.environ['HTTP_ACCEPT_LANGUAGE'] = LANGUAGE
organisation.activate(activity=activity).translationTest()
self.commit()
# Remove request parameter to check that it was saved at activate call
# and restored at message execution.
del REQUEST.environ['HTTP_ACCEPT_LANGUAGE']
self.tic()
finally:
del Organisation.translationTest
self.assertEqual(TO_STRING, organisation.getTitle())
self.assertEqual(TO_STRING, organisation.getDescription())
def test_112_checkLocalizerWorksSQLQueue(self):
self.CheckLocalizerWorks('SQLQueue')
def test_113_checkLocalizerWorksSQLDict(self):
self.CheckLocalizerWorks('SQLDict')
def test_114_checkSQLQueueActivitySucceedsAfterActivityChangingSkin(self):
portal = self.portal
activity_tool = self.getActivityTool()
# Check that a reference script can be reached
script_id = 'ERP5Site_reindexAll'
self.assertIsNot(getattr(portal, script_id), None)
# Create a new skin selection
skin_selection_name = 'test_114'
portal.portal_skins.manage_skinLayers(add_skin=1, skinpath=[''], skinname=skin_selection_name)
# Create a dummy document
organisation = portal.organisation_module.newContent(portal_type='Organisation')
self.tic()
# Set custom methods to call as activities.
def first(context):
context.changeSkin(skin_selection_name)
if getattr(context, script_id, None) is not None:
raise Exception('%s is not supposed to be found here.' % script_id)
def second(context):
# If the wrong skin is selected this will raise.
getattr(context, script_id)
Organisation.firstTest = first
Organisation.secondTest = second
try:
organisation.activate(tag='foo', activity='SQLQueue').firstTest()
organisation.activate(after_tag='foo', activity='SQLQueue').secondTest()
self.commit()
gc.disable()
self.tic()
gc.enable()
# Forcibly restore skin selection, otherwise getMessageList would only
# emit a log when retrieving the ZSQLMethod.
portal.changeSkin(None)
finally:
del Organisation.firstTest
del Organisation.secondTest
def test_115_checkProcessShutdown(self):
# Thread execution plan for this test:
# main ActivityThread ProcessShutdownThread
# start ActivityThread None None
# wait for rendez_vous_lock (run) None
# wait for rendez_vous_lock release rendez_vous_lock None
# start ProcessShutdownThread wait for activity_lock None
# release activity_lock wait for activity_lock internal wait
# wait for activity_thread (finish) internal wait
# wait for process_shutdown_thread None (finish)
#
# This test only checks that:
# - activity tool can exit between 2 processable activity batches
# - activity tool won't process activities after process_shutdown was called
# - process_shutdown returns before Activity.tic()
# This is not perfect though, since it would require to have access to
# the waiting queue of CMFActivity's internal lock (is_running_lock) to
# make sure that it's what is preventing process_shutdown from returning.
activity_tool = self.getActivityTool()
organisation = self.portal.organisation_module.newContent(
portal_type='Organisation')
self.tic()
activity_event = threading.Event()
rendez_vous_event = threading.Event()
def waitingActivity(context):
# Inform test that we arrived at rendez-vous.
rendez_vous_event.set()
# When this event is available, it means test has called process_shutdown.
assert activity_event.wait(10)
original_dequeue = SQLDict.dequeueMessage
queue_tic_test_dict = {}
def dequeueMessage(self, activity_tool, processing_node, node_family_id_set):
# This is a one-shot method, revert after execution
SQLDict.dequeueMessage = original_dequeue
result = self.dequeueMessage(activity_tool, processing_node, node_family_id_set)
queue_tic_test_dict['isAlive'] = process_shutdown_thread.isAlive()
return result
SQLDict.dequeueMessage = dequeueMessage
Organisation.waitingActivity = waitingActivity
try:
# Use SQLDict with no group method so that both activities won't be
# executed in the same batch, letting activity tool a chance to check
# if execution should stop processing activities.
organisation.activate(activity='SQLDict', tag='foo').waitingActivity()
organisation.activate(activity='SQLDict', after_tag='foo').getTitle()
self.commit()
self.assertEqual(len(activity_tool.getMessageList()), 2)
activity_tool.distribute()
self.commit()
# Start a tic in another thread, so they can meet at rendez-vous.
class ActivityThread(threading.Thread):
def run(self):
# Call changeskin, since skin selection depend on thread id, and we
# are in a new thread.
activity_tool.changeSkin(None)
activity_tool.tic()
activity_thread = ActivityThread()
# Do not try to outlive main thread.
activity_thread.setDaemon(True)
# Call process_shutdown in yet another thread because it will wait for
# running activity to complete before returning, and we need to unlock
# activity *after* calling process_shutdown to make sure the next
# activity won't be executed.
class ProcessShutdownThread(threading.Thread):
def run(self):
activity_tool.process_shutdown(3, 0)
process_shutdown_thread = ProcessShutdownThread()
# Do not try to outlive main thread.
process_shutdown_thread.setDaemon(True)
activity_thread.start()
# Wait at rendez-vous for activity to arrive.
assert rendez_vous_event.wait(10)
# Initiate shutdown
process_shutdown_thread.start()
try:
# Let waiting activity finish and wait for thread exit
activity_event.set()
activity_thread.join(10)
assert not activity_thread.is_alive()
process_shutdown_thread.join(10)
assert not process_shutdown_thread.is_alive()
# Check that there is still one activity pending
message_list = activity_tool.getMessageList()
self.assertEqual(len(message_list), 1)
self.assertEqual(message_list[0].method_id, 'getTitle')
# Check that process_shutdown_thread was still runing when Queue_tic returned.
self.assertTrue(queue_tic_test_dict.get('isAlive'), repr(queue_tic_test_dict))
# Call tic in foreground. This must not lead to activity execution.
activity_tool.tic()
self.assertEqual(len(activity_tool.getMessageList()), 1)
finally:
# Put activity tool back in a working state
try:
cancelProcessShutdown()
except StandardException:
# If something failed in process_shutdown, shutdown lock might not
# be taken in CMFActivity, leading to a new esception here hiding
# test error.
pass
finally:
del Organisation.waitingActivity
SQLDict.dequeueMessage = original_dequeue
self.tic()
def test_hasActivity(self):
active_object = self.portal.organisation_module.newContent(
portal_type='Organisation')
active_process = self.portal.portal_activities.newActiveProcess()
self.tic()
self.assertFalse(active_object.hasActivity())
self.assertFalse(active_process.hasActivity())
def test(obj, **kw):
for activity in ActivityTool.activity_dict:
active_object.activate(activity=activity, **kw).getTitle()
self.commit()
self.assertTrue(obj.hasActivity(), activity)
self.tic()
self.assertFalse(obj.hasActivity(), activity)
test(active_object)
test(active_process, active_process=active_process)
test(active_process, active_process=active_process.getPath())
@for_each_activity
def test_hasErrorActivity_error(self, activity):
# Monkey patch Organisation to add a failing method
def failingMethod(self):
raise ValueError('This method always fail')
Organisation.failingMethod = failingMethod
active_object = self.portal.organisation_module.newContent(
portal_type='Organisation')
active_process = self.portal.portal_activities.newActiveProcess()
self.tic()
self.assertFalse(active_object.hasErrorActivity())
self.assertFalse(active_process.hasErrorActivity())
active_object.activate(
activity=activity, active_process=active_process).failingMethod()
self.commit()
# assert that any activity is created
self.assertTrue(active_object.hasActivity())
self.assertTrue(active_process.hasActivity())
# assert that no error is reported
self.assertFalse(active_object.hasErrorActivity())
self.assertFalse(active_process.hasErrorActivity())
self.flushAllActivities()
# assert that any activity is created
self.assertTrue(active_object.hasActivity())
self.assertTrue(active_process.hasActivity())
# assert that an error has been seen
self.assertTrue(active_object.hasErrorActivity())
self.assertTrue(active_process.hasErrorActivity())
message, = self.getMessageList(activity)
self.deleteMessageList(activity, [message])
@for_each_activity
def test_hasErrorActivity(self, activity):
active_object = self.portal.organisation_module.newContent(
portal_type='Organisation')
active_process = self.portal.portal_activities.newActiveProcess()
self.tic()
self.assertFalse(active_object.hasErrorActivity())
self.assertFalse(active_process.hasErrorActivity())
active_object.activate(
activity=activity, active_process=active_process).getTitle()
self.commit()
# assert that any activity is created
self.assertTrue(active_object.hasActivity())
self.assertTrue(active_process.hasActivity())
# assert that no error is reported
self.assertFalse(active_object.hasErrorActivity())
self.assertFalse(active_process.hasErrorActivity())
self.flushAllActivities()
# assert that any activity is created
self.assertFalse(active_object.hasActivity())
self.assertFalse(active_process.hasActivity())
# assert that no error is reported
self.assertFalse(active_object.hasErrorActivity())
self.assertFalse(active_process.hasErrorActivity())
def test_active_object_hasActivity_does_not_catch_exceptions(self):
"""
Some time ago, hasActivity was doing a silent try/except, and this was
a possible disaster for some projects. Here we make sure that if the
SQL request fails, then the exception is not ignored
"""
active_object = self.portal.organisation_module.newContent(
portal_type='Organisation')
self.tic()
self.assertFalse(active_object.hasActivity())
# Monkey patch to induce any error artificially in the sql connection.
def query(self, query_string,*args, **kw):
raise ValueError
from Products.ZMySQLDA.db import DB
DB.original_query = DB.query
try:
active_object.activate().getTitle()
self.commit()
self.assertTrue(active_object.hasActivity())
# Make the sql request not working
DB.original_query = DB.query
DB.query = query
# Make sure then that hasActivity fails
self.assertRaises(ValueError, active_object.hasActivity)
finally:
DB.query = DB.original_query
del DB.original_query
self.tic()
def test_insert_max_payload(self):
activity_tool = self.portal.portal_activities
# XXX: For unknown reasons, this test runs faster after the tables are
# recreated. We could also make this test run before all others.
activity_tool.manageClearActivities()
self.commit()
max_allowed_packet = activity_tool.getSQLConnection().getMaxAllowedPacket()
insert_list = []
invoke_list = []
N = 100
class Skip(Exception):
"""
Speed up test by not interrupting the first transaction
as soon as we have the information we want.
"""
original_query = six.get_unbound_function(DB.query)
def query(self, query_string, *args, **kw):
if query_string.startswith('INSERT'):
insert_list.append(len(query_string))
if not n:
raise Skip
return original_query(self, query_string, *args, **kw)
def check():
for i in xrange(1, N):
activity_tool.activate(activity=activity, group_id=str(i)
).doSomething(arg)
activity_tool.activate(activity=activity, group_id='~'
).doSomething(' ' * n)
self.tic()
self.assertEqual(len(invoke_list), N)
invoke_list.remove(n)
self.assertEqual(set(invoke_list), {len(arg)})
del invoke_list[:]
activity_tool.__class__.doSomething = \
lambda self, arg: invoke_list.append(len(arg))
try:
DB.query = query
for activity in ActivityTool.activity_dict:
arg = ' ' * (max_allowed_packet // N)
# Find the size of the last message argument, such that all messages
# are inserted in a single query whose size is to the maximum allowed.
n = 0
self.assertRaises(Skip, check)
self.abort()
n = max_allowed_packet - insert_list.pop()
self.assertFalse(insert_list)
# Now check with the biggest insert query possible.
check()
self.assertEqual(max_allowed_packet, insert_list.pop())
self.assertFalse(insert_list)
# And check that the insert query is split
# in order not to exceed max_allowed_packet.
n += 1
check()
self.assertEqual(len(insert_list), 2)
del insert_list[:]
finally:
del activity_tool.__class__.doSomething
DB.query = original_query
def test_115_TestSerializationTagSQLDictPreventsParallelExecution(self):
"""
Test if there are multiple activities with the same serialization tag,
then serialization tag guarantees that only one of the same serialization
tagged activities can be processed at the same time.
"""
portal = self.portal
activity_tool = portal.portal_activities
# Add 6 activities
portal.organisation_module.activate(activity='SQLDict', tag='', serialization_tag='test_115').getId()
self.commit()
portal.organisation_module.activate(activity='SQLDict', serialization_tag='test_115').getTitle()
self.commit()
portal.organisation_module.activate(activity='SQLDict', tag='tag_1', serialization_tag='test_115').getId()
self.commit()
portal.person_module.activate(activity='SQLDict', serialization_tag='test_115').getId()
self.commit()
portal.person_module.activate(activity='SQLDict', tag='tag_2').getId()
self.commit()
portal.organisation_module.activate(activity='SQLDict', tag='', serialization_tag='test_115').getId()
self.commit()
# distribute and assign them to 3 nodes
activity_tool.distribute()
self.commit()
activity = ActivityTool.activity_dict['SQLDict']
activity.getProcessableMessageList(activity_tool, 1, ())
self.commit()
activity.getProcessableMessageList(activity_tool, 2, ())
self.commit()
activity.getProcessableMessageList(activity_tool, 3, ())
self.commit()
result = activity._getMessageList(activity_tool.getSQLConnection())
try:
self.assertEqual(len([message
for message in result
if (message.processing_node>0 and
message.serialization_tag=='test_115')]),
1)
self.assertEqual(len([message
for message in result
if (message.processing_node==-1 and
message.serialization_tag=='test_115')]),
4)
self.assertEqual(len([message
for message in result
if (message.processing_node>0 and
message.serialization_tag=='')]),
1)
finally:
# Clear activities from all nodes
self.deleteMessageList('SQLDict', result)
def test_116_RaiseInCommitBeforeMessageExecution(self):
"""
Test behaviour of CMFActivity when the commit just before message
execution fails. In particular, it should restart the messages it
selected (processing_node=current_node) instead of ignoring them forever.
"""
processed = []
activity_tool = self.portal.portal_activities
activity_tool.__class__.doSomething = processed.append
try:
for activity in ActivityTool.activity_dict:
activity_tool.activate(activity=activity).doSomething(activity)
self.commit()
# Make first commit in dequeueMessage raise
registerFailingTransactionManager()
self.assertRaises(CommitFailed, activity_tool.tic)
# Normally, the request stops here and Zope aborts the transaction
self.abort()
self.assertEqual(processed, [])
# Activity is already reserved for current node. Check tic reselects it.
activity_tool.tic()
self.assertEqual(processed, [activity])
del processed[:]
finally:
del activity_tool.__class__.doSomething
def test_117_PlacelessDefaultReindexParameters(self):
"""
Test behaviour of PlacelessDefaultReindexParameters.
"""
portal = self.portal
# Make a new Person object to make sure that the portal type
# is migrated to an instance of a portal type class, otherwise
# the portal type may generate an extra active object.
portal.person_module.newContent(portal_type='Person')
self.tic()
original_reindex_parameters = portal.getPlacelessDefaultReindexParameters()
if original_reindex_parameters is None:
original_reindex_parameters = {}
tag = 'SOME_RANDOM_TAG'
activate_kw = original_reindex_parameters.get('activate_kw', {}).copy()
activate_kw['tag'] = tag
portal.setPlacelessDefaultReindexParameters(activate_kw=activate_kw, \
**original_reindex_parameters)
current_default_reindex_parameters = portal.getPlacelessDefaultReindexParameters()
self.assertEqual({'activate_kw': {'tag': tag}}, \
current_default_reindex_parameters)
person = portal.person_module.newContent(portal_type='Person')
self.commit()
# as we specified it in setPlacelessDefaultReindexParameters we should have
# an activity for this tags
self.assertEqual(1, portal.portal_activities.countMessageWithTag(tag))
self.tic()
self.assertEqual(0, portal.portal_activities.countMessageWithTag(tag))
# restore originals ones
portal.setPlacelessDefaultReindexParameters(**original_reindex_parameters)
person = portal.person_module.newContent(portal_type='Person')
# .. now no messages with this tag should apper
self.assertEqual(0, portal.portal_activities.countMessageWithTag(tag))
@for_each_activity
def testTryNotificationSavedOnEventLogWhenNotifyUserRaises(self, activity):
obj = self.portal.organisation_module.newContent(portal_type='Organisation')
self.tic()
original_notifyUser = six.get_unbound_function(Message.notifyUser)
def failSendingEmail(self, *args, **kw):
raise MailHostError('Mail is not sent')
activity_unit_test_error = Exception()
def failingMethod(self):
raise activity_unit_test_error
try:
Message.notifyUser = failSendingEmail
Organisation.failingMethod = failingMethod
self._catch_log_errors()
obj.activate(activity=activity, priority=6).failingMethod()
self.commit()
self.flushAllActivities(silent=1, loop_size=100)
message, = self.getMessageList(activity)
self.commit()
for log_record in self.logged:
if log_record.name == 'ActivityTool' and log_record.levelname == 'WARNING':
type, value, trace = log_record.exc_info
self.commit()
self.assertIs(activity_unit_test_error, value)
self.deleteMessageList(activity, [message])
finally:
Message.notifyUser = original_notifyUser
del Organisation.failingMethod
self._ignore_log_errors()
@for_each_activity
def testNotificationFailureIsNotSavedOnEventLogWhenMailNotificationIsDisabled(self, activity):
obj = self.portal.organisation_module.newContent(portal_type='Organisation')
self.tic()
original_notifyUser = six.get_unbound_function(Message.notifyUser)
def failSendingEmail(self, *args, **kw):
raise MailHostError('Mail is not sent')
activity_unit_test_error = Exception()
def failingMethod(self):
raise activity_unit_test_error
try:
self.portal.portal_activities.activity_failure_mail_notification = False
Message.notifyUser = failSendingEmail
Organisation.failingMethod = failingMethod
self._catch_log_errors()
obj.activate(activity=activity, priority=6).failingMethod()
self.commit()
self.flushAllActivities(silent=1, loop_size=100)
message, = self.getMessageList(activity)
self.commit()
for log_record in self.logged:
if log_record.name == 'ActivityTool' and log_record.levelname == 'WARNING':
type, value, trace = log_record.exc_info
self.commit()
self.assertIs(activity_unit_test_error, value)
self.deleteMessageList(activity, [message])
finally:
self.portal.portal_activities.activity_failure_mail_notification = True
Message.notifyUser = original_notifyUser
del Organisation.failingMethod
self._ignore_log_errors()
@for_each_activity
def testTryUserMessageContainingNoTracebackIsStillSent(self, activity):
# With Message.__call__
# 1: activity context does not exist when activity is executed
obj = self.portal.organisation_module.newContent(portal_type='Organisation')
self.tic()
notification_done = []
def fake_notifyUser(self, *args, **kw):
notification_done.append(True)
self.traceback = None
original_notifyUser = Message.notifyUser
def failingMethod(self):
raise ValueError("This method always fail")
Message.notifyUser = fake_notifyUser
Organisation.failingMethod = failingMethod
try:
obj.activate(activity=activity).failingMethod()
self.commit()
self.flushAllActivities(silent=1, loop_size=100)
message, = self.getMessageList(activity)
self.assertEqual(len(notification_done), 1)
self.assertEqual(message.traceback, None)
message(self.getActivityTool())
self.deleteMessageList(activity, [message])
finally:
Message.notifyUser = original_notifyUser
del Organisation.failingMethod
@for_each_activity
def testTryNotificationSavedOnEventLogWhenSiteErrorLoggerRaises(self, activity):
# Make sure that no active object is installed.
o = self.getOrganisation()
class ActivityUnitTestError(Exception):
pass
activity_unit_test_error = ActivityUnitTestError()
def failingMethod(self):
raise activity_unit_test_error
from Products.SiteErrorLog.SiteErrorLog import SiteErrorLog
original_raising = six.get_unbound_function(SiteErrorLog.raising)
# Monkey patch Site Error to induce conflict errors artificially.
def raising(self, info):
raise AttributeError
try:
SiteErrorLog.raising = raising
Organisation.failingMethod = failingMethod
self._catch_log_errors()
o.activate(activity = activity).failingMethod()
self.commit()
message, = self.getMessageList(activity)
self.flushAllActivities(silent = 1)
SiteErrorLog.raising = original_raising
self.commit()
for log_record in self.logged:
if log_record.name == 'ActivityTool' and log_record.levelname == 'WARNING':
type, value, trace = log_record.exc_info
self.assertIs(activity_unit_test_error, value)
self.deleteMessageList(activity, [message])
finally:
SiteErrorLog.raising = original_raising
del Organisation.failingMethod
self._ignore_log_errors()
def test_128_CheckDistributeWithSerializationTagAndGroupMethodId(self):
activity_tool = self.portal.portal_activities
obj1 = activity_tool.newActiveProcess()
obj2 = activity_tool.newActiveProcess()
self.tic()
group_method_call_list = []
def doSomething(self, message_list):
r = []
for m in message_list:
m.result = r.append((m.object.getPath(), m.args, m.kw))
r.sort()
group_method_call_list.append(r)
activity_tool.__class__.doSomething = doSomething
try:
for activity in ActivityTool.activity_dict:
activity_kw = dict(activity=activity, serialization_tag=self.id(),
group_method_id='portal_activities/doSomething')
obj1.activate(**activity_kw).dummy(1, x=None)
obj2.activate(**activity_kw).dummy(2, y=None)
self.commit()
activity_tool.distribute()
activity_tool.tic()
self.assertEqual(group_method_call_list.pop(),
sorted([(obj1.getPath(), (1,), dict(x=None)),
(obj2.getPath(), (2,), dict(y=None))]))
self.assertFalse(group_method_call_list)
self.assertFalse(activity_tool.getMessageList())
obj1.activate(priority=2, **activity_kw).dummy1(1, x=None)
obj1.activate(priority=1, **activity_kw).dummy2(2, y=None)
message1 = obj1.getPath(), (1,), dict(x=None)
message2 = obj1.getPath(), (2,), dict(y=None)
self.commit()
activity_tool.distribute()
self.assertEqual(len(activity_tool.getMessageList()), 2)
activity_tool.tic()
self.assertEqual(group_method_call_list.pop(),
[message2] if activity != 'SQLQueue' else [message1, message2])
self.assertFalse(group_method_call_list)
finally:
del activity_tool.__class__.doSomething
def test_129_beforeCommitHook(self):
"""
Check it is possible to activate an object from a before commit hook
"""
def doSomething(person):
person.activate(activity='SQLDict')._setFirstName('John')
person.activate(activity='SQLQueue')._setLastName('Smith')
person = self.portal.person_module.newContent()
transaction.get().addBeforeCommitHook(doSomething, (person,))
self.tic()
self.assertEqual(person.getTitle(), 'John Smith')
def test_connection_migration(self):
"""
Make sure the cmf_activity_sql_connection is automatically migrated from
the ZMySQLDA Connection class to ActivityConnection
"""
# replace the activity connector with a standard ZMySQLDA one
portal = self.portal
activity_tool = portal.portal_activities
stdconn = portal.cmf_activity_sql_connection
portal._delObject('cmf_activity_sql_connection')
portal.manage_addProduct['ZMySQLDA'].manage_addZMySQLConnection(
stdconn.id,
stdconn.title,
stdconn.connection_string,
)
oldconn = portal.cmf_activity_sql_connection
self.assertEqual(oldconn.meta_type, 'Z MySQL Database Connection')
# force rebootstrap and check that migration of the connection happens
# automatically
from Products.ERP5Type.dynamic import portal_type_class
portal_type_class._bootstrapped.clear()
portal_type_class.synchronizeDynamicModules(activity_tool, True)
activity_tool.activate(activity='SQLQueue').getId()
self.tic()
newconn = portal.cmf_activity_sql_connection
self.assertEqual(newconn.meta_type, 'CMFActivity Database Connection')
def test_connection_installable(self):
"""
Test if the cmf_activity_sql_connector can be installed
"""
# delete the activity connection
portal = self.portal
stdconn = portal.cmf_activity_sql_connection
portal._delObject('cmf_activity_sql_connection')
# check the installation form can be rendered
portal.manage_addProduct['CMFActivity'].connectionAdd(
portal.REQUEST
)
# check it can be installed
portal.manage_addProduct['CMFActivity'].manage_addActivityConnection(
stdconn.id,
stdconn.title,
stdconn.connection_string
)
newconn = portal.cmf_activity_sql_connection
self.assertEqual(newconn.meta_type, 'CMFActivity Database Connection')
def test_connection_sortkey(self):
"""
Check that SQL connection has properly initialized sort key,
even when its container (ZODB connection) is reused by another thread.
"""
def sortKey():
app = ZopeTestCase.app()
try:
c = app[self.getPortalName()].cmf_activity_sql_connection()
return app._p_jar, c.sortKey()
finally:
ZopeTestCase.close(app)
jar, sort_key = sortKey()
self.assertNotEqual(1, sort_key)
result = []
t = threading.Thread(target=lambda: result.extend(sortKey()))
t.daemon = True
t.start()
t.join()
self.assertIs(result[0], jar)
self.assertEqual(result[1], sort_key)
def test_onErrorCallback(self):
activity_tool = self.portal.portal_activities
obj = activity_tool.newActiveProcess()
self.tic()
def _raise(exception): # I wish exceptions are callable raising themselves
raise exception
def doSomething(self, conflict_error, cancel):
self.activity_count += 1
error = ConflictError() if conflict_error else Exception()
def onError(exc_type, exc_value, traceback):
assert exc_value is error
env = self.getActivityRuntimeEnvironment()
weakref_list.extend(map(weakref.ref, (env, env._message)))
self.on_error_count += 1
return cancel
self.getActivityRuntimeEnvironment().edit(on_error_callback=onError)
if not self.on_error_count:
if not conflict_error:
raise error
transaction.get().addBeforeCommitHook(_raise, (error,))
obj.__class__.doSomething = doSomething
try:
for activity in ActivityTool.activity_dict:
for conflict_error in False, True:
weakref_list = []
obj.activity_count = obj.on_error_count = 0
obj.activate(activity=activity).doSomething(conflict_error, True)
self.tic()
self.assertEqual(obj.activity_count, 0)
self.assertEqual(obj.on_error_count, 1)
gc.collect()
self.assertEqual([x() for x in weakref_list], [None, None])
weakref_list = []
obj.activate(activity=activity).doSomething(conflict_error, False)
obj.on_error_count = 0
self.tic()
self.assertEqual(obj.activity_count, 1)
self.assertEqual(obj.on_error_count, 1)
gc.collect()
self.assertEqual([x() for x in weakref_list], [None, None])
finally:
del obj.__class__.doSomething
def test_duplicateGroupedMessage(self):
activity_tool = self.portal.portal_activities
obj = activity_tool.newActiveProcess()
obj.reindexObject(activate_kw={'tag': 'foo', 'after_tag': 'bar'})
self.commit()
# Check that both messages were inserted.
# Also serves as a sanity check on indexation activities group_method_id.
indexation_group_metdod_id = 'portal_catalog/catalogObjectList'
self.assertEqual(
len([
x
for x in activity_tool.getMessageList(path=obj.getPath())
if x.activity_kw.get('group_method_id') == indexation_group_metdod_id
]),
2,
)
invoked = []
def invokeGroup(self, method_id, message_list, *args):
# Ignore any other activity which may be spawned from these catalog
# indexations (ex: fulltext indexations).
if method_id == indexation_group_metdod_id:
invoked.append(len(message_list))
return ActivityTool_invokeGroup(self, method_id, message_list, *args)
ActivityTool_invokeGroup = activity_tool.__class__.invokeGroup
try:
activity_tool.__class__.invokeGroup = invokeGroup
self.tic()
finally:
activity_tool.__class__.invokeGroup = ActivityTool_invokeGroup
self.assertEqual(invoked, [1])
def test_mergeParent(self):
category_tool = self.portal.portal_categories
# Test data: c0
# / \
# c1 c2
# / \ |
# c3 c4 c5
c = [category_tool.newContent()]
for i in xrange(5):
c.append(c[i//2].newContent())
self.tic()
def activate(i, priority=1, **kw):
kw.setdefault('merge_parent', c[0].getPath())
c[i].activate(priority=priority, **kw).doSomething()
def check(*expected):
self.tic()
self.assertEqual(tuple(invoked), expected)
del invoked[:]
invoked = []
def doSomething(self):
invoked.append(c.index(self))
Base.doSomething = doSomething
try:
for t in (0, 1), (0, 4, 2), (1, 0, 5), (3, 2, 0):
for p, i in enumerate(t):
activate(i, p)
check(0)
activate(1, 0); activate(5, 1); check(1, 5)
activate(3, 0); activate(1, 1); check(1)
activate(2, 0); activate(1, 1); activate(4, 2); check(2, 1)
activate(4, 0); activate(5, 1); activate(3, 2); check(4, 5, 3)
activate(3, 0, merge_parent=c[1].getPath()); activate(0, 1); check(3, 0)
# Following test shows that a child can be merged with a parent even if
# 'merge_parent' is not specified. This can't be avoided without loading
# all found duplicates, which would be bad for performance.
activate(0, 0); activate(4, 1, merge_parent=None); check(0)
finally:
del Base.doSomething
def activate(i, priority=1, **kw):
c[i].activate(group_method_id='portal_categories/invokeGroup',
merge_parent=c[(i-1)//2 or i].getPath(),
priority=priority, **kw).doSomething()
def invokeGroup(self, message_list):
r = []
for m in message_list:
m.result = r.append(c.index(m.object))
r.sort()
invoked.append(r)
category_tool.__class__.invokeGroup = invokeGroup
try:
activate(5, 0); activate(1, 1); check([1, 5])
activate(4, 0); activate(1, 1); activate(2, 0); check([1, 2])
activate(1, 0); activate(5, 0); activate(3, 1); check([1, 5])
for p, i in enumerate((5, 3, 2, 1, 4)):
activate(i, p, group_id=str(2 != i != 5))
check([2], [1])
for cost in 0.3, 0.1:
activate(2, 0, group_method_cost=cost)
activate(3, 1); activate(4, 2); activate(1, 3)
check([1, 2])
finally:
del category_tool.__class__.invokeGroup
category_tool._delObject(c[0].getId())
self.tic()
def test_getMessageList(self):
activity_tool = self.portal.portal_activities
module = self.portal.person_module
module.activate(after_tag="foo").getUid()
module.activate(activity='SQLQueue', tag="foo").getId()
activity_tool.activate(priority=-1).getId()
def check(expected, **kw):
self.assertEqual(expected, len(activity_tool.getMessageList(**kw)))
def test(check=lambda _, **kw: check(0, **kw)):
check(2, path=module.getPath())
check(3, method_id=("getId", "getUid"))
check(1, tag="foo")
check(0, tag="foo", method_id="getUid")
check(1, processing_node=-1)
check(3, processing_node=range(-5,5))
test()
self.commit()
test(check)
self.tic()
test()
def test_MessageNonExecutable(self):
message_list = self.portal.MailHost._message_list
del message_list[:]
activity_tool = self.portal.portal_activities
kw = {}
self._catch_log_errors()
try:
activity_tool.activity_failure_mail_notification = True
for kw['activity'] in ActivityTool.activity_dict:
for kw['group_method_id'] in '', None:
obj = activity_tool.newActiveProcess()
self.tic()
obj.activate(**kw).getId()
activity_tool._delOb(obj.getId())
obj = activity_tool.newActiveProcess(id=obj.getId(),
is_indexable=False)
self.commit()
self.assertEqual(1, activity_tool.countMessage())
self.flushAllActivities()
sender, recipients, mail = message_list.pop()
self.assertIn('UID mismatch', mail)
m, = activity_tool.getMessageList()
self.assertEqual(m.processing_node, INVOKE_ERROR_STATE)
obj.flushActivity()
obj.activate(**kw).getId()
activity_tool._delOb(obj.getId())
self.commit()
self.assertEqual(1, activity_tool.countMessage())
activity_tool.tic()
cmf_activty_log, = [log for log in self.logged if 'CMFActivity' in log.name]
self.logged = []
self.assertIn('no object found', cmf_activty_log.getMessage())
finally:
self._ignore_log_errors()
self.assertFalse(message_list)
def test_activateByPath(self):
organisation = self.getOrganisation()
self.portal.portal_activities.activateObject(
organisation.getPath(),
activity='SQLDict',
active_process=None
).getTitle()
self.tic()
def test_activateOnZsqlBrain(self):
organisation, = self.getOrganisationModule().searchFolder(
id=self.company_id)
organisation.activate().getTitle()
self.tic()
def test_flushActivitiesOnDelete(self):
organisation = self.getOrganisation()
organisation.getParentValue()._delObject(organisation.getId())
organisation.activate().getTitle()
self.tic()
def test_flushActivitiesOnDeleteWithAcquierableObject(self):
# Create an object with the same ID that can be acquired
self.portal._setObject(self.company_id, Organisation(self.company_id))
organisation = self.getOrganisation()
organisation.getParentValue()._delObject(organisation.getId())
organisation.reindexObject()
self.tic()
def test_failingGroupMethod(self):
activity_tool = self.portal.portal_activities
obj = activity_tool.newActiveProcess()
self.tic()
obj.x = 1
def doSomething(self):
self.x %= self.x
obj.__class__.doSomething = doSomething
try:
activity_kw = dict(activity="SQLQueue", group_method_id=None)
obj.activate(**activity_kw).doSomething()
obj.activate(**activity_kw).doSomething()
obj.activate(**activity_kw).doSomething()
self.commit()
self.assertEqual(3, len(activity_tool.getMessageList()))
activity_tool.tic()
self.assertEqual(obj.x, 0)
skipped, failed = activity_tool.getMessageList()
self.assertEqual(0, skipped.retry)
self.assertEqual(1, failed.retry)
obj.x = 1
self.commit()
activity_tool.timeShift(VALIDATION_ERROR_DELAY)
activity_tool.tic()
m, = activity_tool.getMessageList()
self.assertEqual(1, failed.retry)
obj.x = 1
self.commit()
activity_tool.timeShift(VALIDATION_ERROR_DELAY)
activity_tool.tic()
finally:
del obj.__class__.doSomething
def test_restrictedGroupMethod(self):
skin = self.portal.portal_skins.custom
script_id = self.id()
script = createZODBPythonScript(skin, script_id, "message_list", """if 1:
for m in message_list:
m.result = m.object.getProperty(*m.args, **m.kw)
""")
script.manage_proxy(("Manager",))
obj = self.portal.portal_activities.newActiveProcess(causality_value_list=(
self.portal.person_module, self.portal.organisation_module))
obj.manage_permission('Access contents information', ['Manager'])
self.logout()
foo = obj.activate(activity='SQLQueue',
group_method_id=script_id,
active_process=obj.getPath()).foo
foo('causality', portal_type='Organisation Module')
foo('stop_date', 'bar')
self.tic()
self.assertEqual(sorted(x.getResult() for x in obj.getResultList()),
['bar', 'organisation_module'])
skin.manage_delObjects([script_id])
self.tic()
def test_getCurrentNode(self):
current_node = getattr(getConfiguration(), 'product_config', {}) \
.get('cmfactivity', {}).get('node-id')
if not current_node:
current_node = getServerAddress()
node = getCurrentNode()
self.assertEqual(node, current_node)
activity_node = self.portal.portal_activities.getCurrentNode()
self.assertEqual(activity_node, current_node)
def test_getServerAddress(self):
host, port = self.startZServer()
ip = socket.gethostbyname(host)
server_address = '%s:%s' % (ip, port)
address = getServerAddress()
self.assertEqual(address, server_address)
activity_address = self.portal.portal_activities.getServerAddress()
self.assertEqual(activity_address, server_address)
def test_nodePreference(self):
"""
Test node preference, i.e. 'node' parameter of activate()
An object is activated by 2 different nodes and the 2 messages are
processed by the node that created the newest one:
- without node preference: they're ordered by date
- with node preference: they're executed in reverse order (the
processing node executes its message first even if it's newer)
Correct ordering of queues is also checked, by including scenarios
in which one message is in SQLDict and the other in SQLQueue.
"""
activity_tool = self.portal.portal_activities
o = self.getOrganisation()
node_dict = dict(activity_tool.getNodeDict())
assert len(node_dict) == 1 and '' not in node_dict, node_dict
before = DateTime() - 1
activities = 'SQLDict', 'SQLQueue'
for activities in product(activities, activities):
for node, expected in (None, '12'), ('', '21'), ('same', '12'):
o._setTitle('0')
# The dance around getNodeDict is to simulate the creation of
# activities from 2 different nodes. We also change title in 2
# different ways, so that SQLDict does not merge them.
o.activate(activity=activities[0], node=node)._setTitle('1')
activity_tool.getNodeDict = lambda: node_dict
node_dict[''] = ActivityTool.ROLE_PROCESSING
o.activate(activity=activities[1], node=node, at_date=before
)._setProperty('title', '2')
del node_dict['']
activity_tool._p_invalidate()
self.commit()
for title in expected:
self.ticOnce()
self.assertEqual(o.getTitle(), title, (activities, expected))
self.assertFalse(activity_tool.getMessageList())
def test_nodeFamilies(self):
"""
Test node families, i.e. 'node' parameter of activate() beyond "", "same"
and None.
"""
activity_tool = self.portal.portal_activities
node_id, = activity_tool.getNodeDict()
other = 'boo'
member = 'foo'
non_member = 'bar'
does_not_exist = 'baz'
# Family declaration API
self.assertItemsEqual(activity_tool.getFamilyNameList(), [])
self.assertRaises(
ValueError,
activity_tool.createFamily, 'same', # Reserved name
)
self.assertRaises(
TypeError,
activity_tool.createFamily, -5, # Not a string
)
activity_tool.createFamily(other)
self.assertRaises(
ValueError,
activity_tool.createFamily, other, # Exists
)
activity_tool.createFamily(member)
self.assertRaises(
ValueError,
activity_tool.renameFamily, other, member, # New name exists
)
self.assertRaises(
ValueError,
activity_tool.renameFamily, does_not_exist, member, # Old name does not exist
)
self.assertRaises(
TypeError,
activity_tool.renameFamily, other, -4, # New name not a string
)
activity_tool.deleteFamily(member)
# Silent success
activity_tool.deleteFamily(member)
activity_tool.createFamily(non_member)
self.assertItemsEqual(activity_tool.getFamilyNameList(), [other, non_member])
# API for node a-/di-ssociation with/from families
self.assertItemsEqual(activity_tool.getCurrentNodeFamilyNameSet(), [])
activity_tool.addNodeToFamily(node_id, other)
self.assertItemsEqual(activity_tool.getCurrentNodeFamilyNameSet(), [other])
# Silent success
activity_tool.addNodeToFamily(node_id, other)
self.assertItemsEqual(activity_tool.getCurrentNodeFamilyNameSet(), [other])
activity_tool.addNodeToFamily(node_id, non_member)
self.assertItemsEqual(activity_tool.getCurrentNodeFamilyNameSet(), [other, non_member])
activity_tool.removeNodeFromFamily(node_id, non_member)
self.assertItemsEqual(activity_tool.getCurrentNodeFamilyNameSet(), [other])
# Silent success
activity_tool.removeNodeFromFamily(node_id, non_member)
self.assertItemsEqual(activity_tool.getCurrentNodeFamilyNameSet(), [other])
activity_tool.createFamily(does_not_exist)
activity_tool.addNodeToFamily(node_id, does_not_exist)
self.assertItemsEqual(activity_tool.getCurrentNodeFamilyNameSet(), [other, does_not_exist])
activity_tool.deleteFamily(does_not_exist)
self.assertItemsEqual(activity_tool.getCurrentNodeFamilyNameSet(), [other])
self.assertItemsEqual(activity_tool.getFamilyNameList(), [other, non_member])
activity_tool.renameFamily(other, member)
self.assertItemsEqual(activity_tool.getFamilyNameList(), [member, non_member])
self.assertItemsEqual(activity_tool.getCurrentNodeFamilyNameSet(), [member])
activity_tool.createFamily(other)
activity_tool.addNodeToFamily(node_id, other)
self.assertItemsEqual(activity_tool.getFamilyNameList(), [member, non_member, other])
self.assertItemsEqual(activity_tool.getCurrentNodeFamilyNameSet(), [member, other])
activity_tool.deleteFamily(other)
self.assertItemsEqual(activity_tool.getFamilyNameList(), [member, non_member])
self.assertItemsEqual(activity_tool.getCurrentNodeFamilyNameSet(), [member])
o = self.getOrganisation()
for activity in 'SQLDict', 'SQLQueue':
# Sanity check.
self.assertEqual(self.getMessageList(activity), [])
self.assertRaises(
ValueError,
o.activate, activity=activity, node=does_not_exist,
)
for node, expected in (member, '1'), (non_member, '0'), ('', '1'), ('same', '1'):
o._setTitle('0')
o.activate(activity=activity, node=node)._setTitle('1')
self.commit()
self.ticOnce()
self.assertEqual(
o.getTitle(),
expected,
(activity, o.getTitle(), expected),
)
if expected == '0':
# The activity must still exist, waiting for a node of the
# appropriate family.
result = self.getMessageList(activity)
self.assertEqual(len(result), 1)
self.deleteMessageList(activity, result)
def test_message_auto_validation(self):
"""
Test that messages without dependencies are directly spawned with
processing_node=0.
"""
organisation = self.portal.organisation_module.newContent(portal_type='Organisation')
self.tic()
activity_tool = self.getActivityTool()
organisation.activate(tag='1').getId()
organisation.activate(tag='2', after_tag=None).getId()
organisation.activate(tag='3', after_tag='foo').getId()
self.commit()
activity_tool.getMessageList()
self.assertItemsEqual(
[('1', 0), ('2', 0), ('3', -1)],
[
(x.activity_kw['tag'], x.processing_node)
for x in self.getActivityTool().getMessageList()
],
)
self.tic()
def test_activity_timeout(self):
slow_method_id = 'Base_getSlowObjectList'
createZODBPythonScript(
self.portal.portal_skins.custom,
slow_method_id,
'selection=None, **kw',
"""
from time import sleep
sleep(3)
return [x.getObject() for x in context.portal_catalog(limit=100)]
""")
# Set short enough activity timeout configuration
import Products.ERP5Type.Timeout
Products.ERP5Type.Timeout.activity_timeout = 2.0
self.portal.portal_templates.activate().Base_getSlowObjectList()
with self.assertRaises(RuntimeError):
self.tic()
message, = self.getMessageList('SQLDict')
self.assertEqual(message.retry, 0)
self.deleteMessageList(
'SQLDict',
[message],
)
def test_zmi_views(self):
# we can render ZMI view without errors or warnings
with warnings.catch_warnings(record=True) as catched_warnings:
self.portal.portal_activities.manage_overview()
self.portal.portal_activities.manageActivities()
self.portal.portal_activities.manageActivitiesAdvanced()
self.portal.portal_activities.manageLoadBalancing()
self.assertEqual(catched_warnings, [])
@for_each_activity
def testSpawnTimeUserGroupAndRoleUsedDuringExecution(self, activity):
obj = self.portal.organisation_module.newContent(portal_type='Organisation')
self.tic()
# This user cannot be created by userfolder API, validating that activity
# execution does not use it.
# Using a PropertiedUser because it is the lowest-level class which has a
# groups notion.
artificial_user = PropertiedUser(
id='this user does not exist',
login='does not matter',
).__of__(self.portal.acl_users)
artificial_user._addGroups(groups=('group 1', 'group 2'))
artificial_user._addRoles(roles=('role 1', 'role 2'))
initial_security_manager = getSecurityManager()
def checkUserGroupAndRole(organisation_self):
user = getSecurityManager().getUser()
self.assertIs(type(aq_base(user)), PropertiedUser)
self.assertEqual(aq_parent(user), aq_parent(artificial_user))
self.assertEqual(user.getId(), artificial_user.getId())
self.assertItemsEqual(user.getGroups(), artificial_user.getGroups())
self.assertItemsEqual(user.getRoles(), artificial_user.getRoles())
Organisation.checkUserGroupAndRole = checkUserGroupAndRole
try:
newSecurityManager(None, artificial_user)
obj.activate(activity=activity).checkUserGroupAndRole()
self.tic()
finally:
setSecurityManager(initial_security_manager)
del Organisation.checkUserGroupAndRole
@for_each_activity
def test_dummyGroupMethodUser(self, activity):
activity_tool = self.portal.portal_activities
user_folder = self.portal.acl_users
expected_user_list = [
PropertiedUser(id='user1', login='user1').__of__(user_folder),
PropertiedUser(id='user2', login='user2').__of__(user_folder),
]
for index, user in enumerate(expected_user_list):
user._addGroups(groups=['role %i' % index])
context_list = [
self.portal.organisation_module.newContent(portal_type='Organisation')
for _ in expected_user_list
]
self.tic()
user_list = [None for _ in expected_user_list]
def doSomething(self, index):
user_list[index] = getSecurityManager().getUser()
Organisation.doSomething = doSomething
try:
initial_security_manager = getSecurityManager()
try:
for index, (context, user) in enumerate(zip(
context_list,
expected_user_list,
)):
newSecurityManager(None, user)
context.activate(
activity=activity,
group_method_id=None,
).doSomething(index=index)
finally:
setSecurityManager(initial_security_manager)
self.tic()
finally:
del Organisation.doSomething
self.assertEqual(
[x.getRoles() for x in user_list],
[x.getRoles() for x in expected_user_list],
)