Commit 0f7c72df authored by Rafael Monnerat's avatar Rafael Monnerat

Reduce allocation scope usage

See merge request !486
parents 06a27d31 8955cfe7
Pipeline #26466 failed with stage
in 0 seconds
......@@ -1276,6 +1276,7 @@ class TestSlapOSUpdateOpenSaleOrderPeriod(SlapOSTestCaseMixin):
open_order = self.createOpenOrder()
open_order.OpenSaleOrder_updatePeriod()
@simulateByEditWorkflowMark('Person_storeOpenSaleOrderJournal')
def test_updatePeriod_validated(self):
open_order = self.createOpenOrder()
person = self.portal.person_module.template_member\
......@@ -1284,14 +1285,12 @@ class TestSlapOSUpdateOpenSaleOrderPeriod(SlapOSTestCaseMixin):
destination_decision_value=person,
)
script_name = "Person_storeOpenSaleOrderJournal"
self._simulateScript(script_name)
try:
open_order.OpenSaleOrder_updatePeriod()
finally:
self._dropScript(script_name)
self.assertScriptVisited(person, script_name)
open_order.OpenSaleOrder_updatePeriod()
self.assertEqual(
'Visited by Person_storeOpenSaleOrderJournal',
person.workflow_history['edit_workflow'][-1]['comment'])
@simulateByEditWorkflowMark('Person_storeOpenSaleOrderJournal')
def test_updatePeriod_invalidated(self):
open_order = self.createOpenOrder()
person = self.portal.person_module.template_member\
......@@ -1300,14 +1299,11 @@ class TestSlapOSUpdateOpenSaleOrderPeriod(SlapOSTestCaseMixin):
destination_decision_value=person,
)
open_order.invalidate()
open_order.OpenSaleOrder_updatePeriod()
script_name = "Person_storeOpenSaleOrderJournal"
self._simulateScript(script_name)
try:
open_order.OpenSaleOrder_updatePeriod()
finally:
self._dropScript(script_name)
self.assertScriptNotVisited(person, script_name)
self.assertNotEqual(
'Visited by Person_storeOpenSaleOrderJournal',
person.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm(self):
open_order = self.createOpenOrder()
......@@ -1351,39 +1347,46 @@ class TestSlapOSReindexOpenSaleOrder(SlapOSTestCaseMixin):
)
return open_order
def _simulateScript(self, script_name, fake_return="False"):
if script_name in self.portal.portal_skins.custom.objectIds():
raise ValueError('Precondition failed: %s exists in custom' % script_name)
createZODBPythonScript(self.portal.portal_skins.custom,
script_name,
'uid=None,*args, **kwargs',
'# Script body\n'
"""portal_workflow = context.portal_workflow
document = context.portal_catalog.getResultValue(uid=uid)
portal_workflow.doActionFor(document, action='edit_action', comment='Visited by %s') """ % script_name )
transaction.commit()
def test_alarm(self):
open_order = self.createOpenOrder()
self.tic()
# Jut wait a bit so the line has a different timestamp > 1 sec.
time.sleep(1)
open_order.newContent(portal_type="Open Sale Order Line")
open_order_line = open_order.newContent(portal_type="Open Sale Order Line")
self.tic()
script_name = "OpenSaleOrder_reindexIfIndexedBeforeLine"
alarm = self.portal.portal_alarms.slapos_reindex_open_sale_order
self._test_alarm(
alarm, open_order, script_name)
order = self.portal.portal_catalog(
uid=open_order.getUid(),
select_dict={'indexation_timestamp': None})[0]
line = self.portal.portal_catalog(
uid=open_order_line.getUid(),
select_dict={'indexation_timestamp': None})[0]
self.assertTrue(order.indexation_timestamp < line.indexation_timestamp)
self.portal.portal_alarms.slapos_reindex_open_sale_order.activeSense()
self.tic()
order = self.portal.portal_catalog(
uid=open_order.getUid(),
select_dict={'indexation_timestamp': None})[0]
line = self.portal.portal_catalog(
uid=open_order_line.getUid(),
select_dict={'indexation_timestamp': None})[0]
self.assertTrue(order.indexation_timestamp > line.indexation_timestamp,
"%s %s" % (order.indexation_timestamp, line.indexation_timestamp))
def test_alarm_no_line(self):
open_order = self.createOpenOrder()
self.tic()
script_name = "OpenSaleOrder_reindexIfIndexedBeforeLine"
alarm = self.portal.portal_alarms.slapos_reindex_open_sale_order
self._test_alarm_not_visited(
alarm, open_order, script_name)
# Rather them test the alarm with fake script, directly
# test the ERP5Site_zGetOpenOrderWithModifiedLineUid code.
open_order_with_modified_line_uid_list = [i.uid for i in \
self.portal.ERP5Site_zGetOpenOrderWithModifiedLineUid()]
self.assertNotIn(open_order.getUid(), open_order_with_modified_line_uid_list)
class TestSlapOSGeneratePackingListFromTioXML(SlapOSTestCaseMixin):
......
......@@ -147,6 +147,8 @@ class DefaultScenarioMixin(TestSlapOSSecurityMixin):
server.edit(
allocation_scope='open/public')
self.assertEqual('open/public', server.getAllocationScope())
# Called by alarm
server.ComputeNode_checkAndUpdateCapacityScope()
self.assertEqual('open', server.getCapacityScope())
self.tic()
......@@ -155,6 +157,8 @@ class DefaultScenarioMixin(TestSlapOSSecurityMixin):
server.edit(
allocation_scope='open/subscription')
self.assertEqual('open/subscription', server.getAllocationScope())
# Called by alarm
server.ComputeNode_checkAndUpdateCapacityScope()
self.assertEqual('open', server.getCapacityScope())
self.tic()
......@@ -163,6 +167,8 @@ class DefaultScenarioMixin(TestSlapOSSecurityMixin):
server.edit(
allocation_scope='open/personal', subject_list=[])
self.assertEqual('open/personal', server.getAllocationScope())
# Called by alarm
server.ComputeNode_checkAndUpdateCapacityScope()
self.assertEqual('open', server.getCapacityScope())
self.tic()
......@@ -173,6 +179,8 @@ class DefaultScenarioMixin(TestSlapOSSecurityMixin):
server.edit(
allocation_scope='open/friend', subject_list=friend_list)
self.assertEqual('open/friend', server.getAllocationScope())
# Called by alarm
server.ComputeNode_checkAndUpdateCapacityScope()
self.assertEqual('open', server.getCapacityScope())
self.assertSameSet(friend_list, server.getSubjectList())
self.tic()
......
......@@ -78,6 +78,40 @@ def withAbort(func):
self.abort()
return wrapped
class TemporaryAlarmScript(object):
"""
Context manager for temporary python scripts
"""
def __init__(self, portal, script_name, fake_return="", attribute=None):
self.script_name = script_name
self.portal = portal
self.fake_return = fake_return
self.attribute = attribute
def __enter__(self):
if self.script_name in self.portal.portal_skins.custom.objectIds():
raise ValueError('Precondition failed: %s exists in custom' % self.script_name)
if self.attribute is None:
content = """portal_workflow = context.portal_workflow
portal_workflow.doActionFor(context, action='edit_action', comment='Visited by %s')
return %s""" % (self.script_name, self.fake_return)
else:
content = """portal_workflow = context.portal_workflow
context.edit(%s='Visited by %s')
return %s""" % (self.attribute, self.script_name, self.fake_return)
createZODBPythonScript(self.portal.portal_skins.custom,
self.script_name,
'*args, **kwargs',
'# Script body\n' + content)
transaction.commit()
def __exit__(self, exc_type, exc_value, traceback):
if self.script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(self.script_name)
transaction.commit()
class SlapOSTestCaseMixin(testSlapOSMixin):
expected_html_payzen_redirect_page = None
......@@ -620,53 +654,31 @@ class SlapOSTestCaseMixin(testSlapOSMixin):
resource='foo/bar',
)
# Set of methods to help test alarms and to see if the script was called
def _simulateScript(self, script_name, fake_return=""):
if script_name in self.portal.portal_skins.custom.objectIds():
raise ValueError('Precondition failed: %s exists in custom' % script_name)
createZODBPythonScript(self.portal.portal_skins.custom,
script_name,
'*args, **kwargs',
'# Script body\n'
"""portal_workflow = context.portal_workflow
portal_workflow.doActionFor(context, action='edit_action', comment='Visited by %s')
return %s""" % (script_name, fake_return ))
transaction.commit()
def _dropScript(self, script_name):
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
transaction.commit()
def assertScriptVisited(self, document, script_name):
self.assertEqual(
'Visited by %s' % script_name,
document.workflow_history['edit_workflow'][-1]['comment'])
def assertScriptNotVisited(self, document, script_name):
self.assertNotEqual(
'Visited by %s' % script_name,
document.workflow_history['edit_workflow'][-1]['comment'])
def _test_alarm(self, alarm, document, script_name):
def _test_alarm(self, alarm, document, script_name, attribute=None):
self.tic()
self._simulateScript(script_name)
try:
with TemporaryAlarmScript(self.portal, script_name, attribute=attribute):
alarm.activeSense()
self.tic()
finally:
self._dropScript(script_name)
self.assertScriptVisited(document, script_name)
if attribute is None:
content = document.workflow_history['edit_workflow'][-1]['comment']
else:
content = document.getProperty(attribute)
self.assertEqual(
'Visited by %s' % script_name,
content)
def _test_alarm_not_visited(self, alarm, document, script_name):
def _test_alarm_not_visited(self, alarm, document, script_name, attribute=None):
self.tic()
self._simulateScript(script_name)
try:
with TemporaryAlarmScript(self.portal, script_name, attribute=attribute):
alarm.activeSense()
self.tic()
finally:
self._dropScript(script_name)
self.assertScriptNotVisited(document, script_name)
if attribute is None:
content = document.workflow_history['edit_workflow'][-1]['comment']
else:
content = document.getProperty(attribute)
self.assertNotEqual(
'Visited by %s' % script_name,
content)
def restoreAccountingTemplatesOnPreferences(self):
self.login()
......
......@@ -209,6 +209,10 @@ return True""" )
self.software_instance.getUrlString())
self.compute_node.edit(allocation_scope='open/personal',
source_administration=self.person_user.getRelativeUrl())
self.compute_node.setAccessStatus("#access ok")
self.tic()
self.compute_node.ComputeNode_checkAndUpdateCapacityScope()
self.assertEqual(self.compute_node.getCapacityScope(), 'open')
self.tic()
self.assertEqual(None, self.software_instance.getAggregateValue(
......@@ -226,6 +230,10 @@ return True""" )
self.partition)
self.compute_node.edit(allocation_scope='open/personal',
source_administration=self.person_user.getRelativeUrl())
self.compute_node.setAccessStatus("#access ok")
self.tic()
self.compute_node.ComputeNode_checkAndUpdateCapacityScope()
self.assertEqual(self.compute_node.getCapacityScope(), 'open')
self.tic()
self.assertEqual(None, self.software_instance.getAggregateValue(
......@@ -259,6 +267,10 @@ return True""" )
source_administration=person_user.getRelativeUrl(),
destination_section=self.person_user.getRelativeUrl(),
allocation_scope='open/friend')
self.compute_node.setAccessStatus("#access ok")
self.tic()
self.compute_node.ComputeNode_checkAndUpdateCapacityScope()
self.assertEqual(self.compute_node.getCapacityScope(), 'open')
self.tic()
self.assertEqual(None, self.software_instance.getAggregateValue(
......@@ -292,6 +304,10 @@ return True""" )
source_administration=person_user.getRelativeUrl(),
destination_section=self.person_user.getRelativeUrl(),
allocation_scope='open/friend')
self.compute_node.setAccessStatus("#access ok")
self.tic()
self.compute_node.ComputeNode_checkAndUpdateCapacityScope()
self.assertEqual(self.compute_node.getCapacityScope(), 'open')
self.tic()
self.assertEqual(None, self.software_instance.getAggregateValue(
......
......@@ -6,12 +6,6 @@
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>_recorded_property_dict</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent>
</value>
</item>
<item>
<key> <string>default_reference</string> </key>
<value> <string>testSlapOSCloudAllocationAlarm</string> </value>
......@@ -55,28 +49,13 @@
<item>
<key> <string>workflow_history</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAM=</string> </persistent>
<persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="2" aka="AAAAAAAAAAI=">
<pickle>
<global name="PersistentMapping" module="Persistence.mapping"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>data</string> </key>
<value>
<dictionary/>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="3" aka="AAAAAAAAAAM=">
<pickle>
<global name="PersistentMapping" module="Persistence.mapping"/>
</pickle>
......@@ -89,7 +68,7 @@
<item>
<key> <string>component_validation_workflow</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAQ=</string> </persistent>
<persistent> <string encoding="base64">AAAAAAAAAAM=</string> </persistent>
</value>
</item>
</dictionary>
......@@ -98,7 +77,7 @@
</dictionary>
</pickle>
</record>
<record id="4" aka="AAAAAAAAAAQ=">
<record id="3" aka="AAAAAAAAAAM=">
<pickle>
<global name="WorkflowHistoryList" module="Products.ERP5Type.Workflow"/>
</pickle>
......
......@@ -641,7 +641,7 @@ class TestSlapOSCorePersonRequestComputeNode(SlapOSTestCaseMixin):
self.assertEqual('COMP-%s' % (previous_id + 1), compute_node.getReference())
self.assertEqual('validated', compute_node.getValidationState())
self.assertEqual('open/personal', compute_node.getAllocationScope())
self.assertEqual('open', compute_node.getCapacityScope())
self.assertEqual('close', compute_node.getCapacityScope())
def test_request_notReindexedCompute(self):
person = self.portal.portal_membership.getAuthenticatedMember().getUserValue()
......@@ -682,7 +682,7 @@ class TestSlapOSCorePersonRequestComputeNode(SlapOSTestCaseMixin):
self.assertEqual('COMP-%s' % (previous_id + 1), compute_node.getReference())
self.assertEqual('validated', compute_node.getValidationState())
self.assertEqual('open/personal', compute_node.getAllocationScope())
self.assertEqual('open', compute_node.getCapacityScope())
self.assertEqual('close', compute_node.getCapacityScope())
self.tic()
......@@ -708,7 +708,7 @@ class TestSlapOSCorePersonRequestComputeNode(SlapOSTestCaseMixin):
self.assertEqual('COMP-%s' % (previous_id + 1), compute_node.getReference())
self.assertEqual('validated', compute_node.getValidationState())
self.assertEqual('open/personal', compute_node.getAllocationScope())
self.assertEqual('open', compute_node.getCapacityScope())
self.assertEqual('close', compute_node.getCapacityScope())
# and now another one
person.requestComputeNode(compute_node_title=compute_node_title2)
......@@ -735,7 +735,7 @@ class TestSlapOSCorePersonRequestComputeNode(SlapOSTestCaseMixin):
self.assertEqual('COMP-%s' % (previous_id + 2), compute_node2.getReference())
self.assertEqual('validated', compute_node2.getValidationState())
self.assertEqual('open/personal', compute_node2.getAllocationScope())
self.assertEqual('open', compute_node2.getCapacityScope())
self.assertEqual('close', compute_node2.getCapacityScope())
def test_request_duplicatedComputeNode(self):
person = self.portal.portal_membership.getAuthenticatedMember().getUserValue()
......
......@@ -6,12 +6,6 @@
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>_recorded_property_dict</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent>
</value>
</item>
<item>
<key> <string>default_reference</string> </key>
<value> <string>testSlapOSCloudPersonSlapInterfaceWorkflow</string> </value>
......@@ -61,28 +55,13 @@
<item>
<key> <string>workflow_history</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAM=</string> </persistent>
<persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="2" aka="AAAAAAAAAAI=">
<pickle>
<global name="PersistentMapping" module="Persistence.mapping"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>data</string> </key>
<value>
<dictionary/>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="3" aka="AAAAAAAAAAM=">
<pickle>
<global name="PersistentMapping" module="Persistence.mapping"/>
</pickle>
......@@ -95,7 +74,7 @@
<item>
<key> <string>component_validation_workflow</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAQ=</string> </persistent>
<persistent> <string encoding="base64">AAAAAAAAAAM=</string> </persistent>
</value>
</item>
</dictionary>
......@@ -104,7 +83,7 @@
</dictionary>
</pickle>
</record>
<record id="4" aka="AAAAAAAAAAQ=">
<record id="3" aka="AAAAAAAAAAM=">
<pickle>
<global name="WorkflowHistoryList" module="Products.ERP5Type.Workflow"/>
</pickle>
......
......@@ -7,10 +7,6 @@ if not person or \
portal.ERP5Site_isSupportRequestCreationClosed():
return
if context.getAllocationScope("open").startswith("close"):
context.setMonitorScope("disabled")
return
reference = context.getReference()
compute_node_title = context.getTitle()
......
#
# XXX This ticket contains dupplicated coded found arround SlapOS
# It is required to rewrite this in a generic way.
# See also: InstanceTree_checkSoftwareInstanceState
# See also: ComputeNode_checkState
#
from DateTime import DateTime
if context.getSimulationState() == "invalidated":
......@@ -103,17 +96,11 @@ if aggregate_portal_type == "Instance Tree":
if instance.getAggregate() is not None:
compute_node = instance.getAggregateValue().getParentValue()
if instance.getPortalType() == "Software Instance" and \
compute_node.getAllocationScope() in ["open/public", "open/friend", "open/subscription"] and \
instance.getSlapState() == "start_requested" and \
instance.SoftwareInstance_hasReportedError():
message_list.append("%s has error (%s, %s at %s scope %s)" % (instance.getReference(), instance.getTitle(),
instance.getUrlString(), compute_node.getReference(),
compute_node.getAllocationScope()))
if instance.getPortalType() == "Software Instance" and \
compute_node.getAllocationScope() in ["closed/outdated", "open/personal"] and \
instance.getSlapState() == "start_requested" and \
instance.SoftwareInstance_hasReportedError():
message_list.append("%s on a %s compute_node" % (instance.getReference(), compute_node.getAllocationScope()) )
else:
message_list.append("%s is not allocated" % instance.getReference())
return ",".join(message_list)
......
......@@ -502,6 +502,19 @@ class TestSlapOSCrmMonitoringCheckComputeNodeState(SlapOSTestCaseMixinWithAbort)
slapos_crm_check_compute_node_state
self._test_alarm(alarm, self.compute_node, "ComputeNode_checkState")
def _test_alarm_check_compute_node_state_selected(self, allocation_scope,
monitor_scope=None):
self._makeComputeNode()
self.compute_node.edit(allocation_scope=allocation_scope)
self.tic()
if monitor_scope is not None:
self.compute_node.edit(monitor_scope=monitor_scope)
self.tic()
alarm = self.portal.portal_alarms.\
slapos_crm_check_compute_node_state
self._test_alarm(alarm, self.compute_node, "ComputeNode_checkState")
def _test_alarm_check_compute_node_state_not_selected(self, allocation_scope,
monitor_scope=None):
self._makeComputeNode()
......@@ -532,15 +545,15 @@ class TestSlapOSCrmMonitoringCheckComputeNodeState(SlapOSTestCaseMixinWithAbort)
def test_alarm_check_compute_node_state_closed_forever_compute_node(self):
self._test_alarm_check_compute_node_state_not_selected(
allocation_scope='closed/forever')
allocation_scope='close/forever')
def test_alarm_check_compute_node_state_closed_mantainence_compute_node(self):
self._test_alarm_check_compute_node_state_not_selected(
allocation_scope='closed/maintenance')
self._test_alarm_check_compute_node_state_selected(
allocation_scope='close/maintenance')
def test_alarm_check_compute_node_state_closed_termination_compute_node(self):
self._test_alarm_check_compute_node_state_not_selected(
allocation_scope='closed/termination')
self._test_alarm_check_compute_node_state_selected(
allocation_scope='close/termination')
class TestSlapOSCrmMonitoringCheckComputeNodeSoftwareInstallation(SlapOSTestCaseMixinWithAbort):
......@@ -601,25 +614,48 @@ class TestSlapOSCrmMonitoringCheckComputeNodeSoftwareInstallation(SlapOSTestCase
slapos_crm_check_software_installation_state
self._test_alarm_not_visited(alarm, self.compute_node, "ComputeNode_checkSoftwareInstallationState")
def _test_alarm_not_run_on_close(self, allocation_scope):
def _test_alarm_not_run_on_close(self, allocation_scope, monitor_scope=None):
self._makeComputeNode()
self.compute_node.edit(allocation_scope=allocation_scope)
self.tic()
if monitor_scope is not None:
self.compute_node.edit(monitor_scope=monitor_scope)
self.tic()
alarm = self.portal.portal_alarms.\
slapos_crm_check_software_installation_state
self._test_alarm_not_visited(alarm, self.compute_node, "ComputeNode_checkSoftwareInstallationState")
def _test_alarm_run_on_close(self, allocation_scope,):
self._makeComputeNode()
self.compute_node.edit(allocation_scope=allocation_scope)
self.tic()
alarm = self.portal.portal_alarms.\
slapos_crm_check_software_installation_state
self._test_alarm(alarm, self.compute_node, "ComputeNode_checkSoftwareInstallationState")
def test_alarm_not_run_on_close_forever(self):
self._test_alarm_not_run_on_close('close/forever')
def test_alarm_not_run_on_close_maintainence(self):
self._test_alarm_not_run_on_close('close/maintenence')
self._test_alarm_not_run_on_close('close/maintenence', monitor_scope="disabled")
def test_alarm_not_run_on_close_outdated(self):
self._test_alarm_not_run_on_close('close/outdated')
self._test_alarm_not_run_on_close('close/outdated', monitor_scope="disabled")
def test_alarm_not_run_on_close_termination(self):
self._test_alarm_not_run_on_close('close/termination')
self._test_alarm_not_run_on_close('close/termination', monitor_scope="disabled")
def test_alarm_run_on_close_maintainence(self):
self._test_alarm_run_on_close('close/maintenence')
def test_alarm_run_on_close_outdated(self):
self._test_alarm_run_on_close('close/outdated')
def test_alarm_run_on_close_termination(self):
self._test_alarm_run_on_close('close/termination')
class TestSlapOSCrmMonitoringCheckInstanceInError(SlapOSTestCaseMixinWithAbort):
......
# Copyright (c) 2002-2012 Nexedi SA and Contributors. All Rights Reserved.
import transaction
from erp5.component.test.SlapOSTestCaseMixin import SlapOSTestCaseMixinWithAbort
from Products.ERP5Type.tests.utils import createZODBPythonScript
from DateTime import DateTime
from zExceptions import Unauthorized
class TestSlapOSERP5CleanupActiveProcess(SlapOSTestCaseMixinWithAbort):
def _simulateActiveProcess_deleteSelf(self):
script_name = 'ActiveProcess_deleteSelf'
if script_name in self.portal.portal_skins.custom.objectIds():
raise ValueError('Precondition failed: %s exists in custom' % script_name)
createZODBPythonScript(self.portal.portal_skins.custom,
script_name,
'*args, **kwargs',
'# Script body\n'
"""description = '%s\\nVisited by ActiveProcess_deleteSelf' % context.getDescription()
context.edit(description=description)""")
transaction.commit()
def _dropActiveProcess_deleteSelf(self):
script_name = 'ActiveProcess_deleteSelf'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
transaction.commit()
def check_cleanup_active_process_alarm(self, date, assert_method):
def check_cleanup_active_process_alarm(self, date, test_method):
def verify_getCreationDate_call(*args, **kwargs):
return date
ActiveProcessClass = self.portal.portal_types.getPortalTypeClass(
......@@ -44,6 +24,13 @@ context.edit(description=description)""")