Commit d84c2d18 authored by Rafael Monnerat's avatar Rafael Monnerat

slapos_crm: Dont update allocation scope anymore

See merge request nexedi/slapos.core!472
parents bf22039d 47db4304
Pipeline #25889 failed with stage
in 0 seconds
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="Alarm" module="erp5.portal_type"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>active_sense_method_id</string> </key>
<value> <string>Alarm_checkAndUpdateComputeNodeAllocationScope</string> </value>
</item>
<item>
<key> <string>description</string> </key>
<value> <string>Automatically turn allocation scope of public and friend compute_node to personnal if the owner is not a service provider.</string> </value>
</item>
<item>
<key> <string>enabled</string> </key>
<value> <int>0</int> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>slapos_crm_check_update_allocation_scope</string> </value>
</item>
<item>
<key> <string>periodicity_hour</string> </key>
<value>
<tuple/>
</value>
</item>
<item>
<key> <string>periodicity_minute</string> </key>
<value>
<tuple>
<int>5</int>
</tuple>
</value>
</item>
<item>
<key> <string>periodicity_minute_frequency</string> </key>
<value> <int>1</int> </value>
</item>
<item>
<key> <string>periodicity_month</string> </key>
<value>
<tuple/>
</value>
</item>
<item>
<key> <string>periodicity_month_day</string> </key>
<value>
<tuple/>
</value>
</item>
<item>
<key> <string>periodicity_start_date</string> </key>
<value>
<object>
<klass>
<global name="DateTime" module="DateTime.DateTime"/>
</klass>
<tuple>
<none/>
</tuple>
<state>
<tuple>
<float>1399420800.0</float>
<string>GMT</string>
</tuple>
</state>
</object>
</value>
</item>
<item>
<key> <string>periodicity_week</string> </key>
<value>
<tuple/>
</value>
</item>
<item>
<key> <string>portal_type</string> </key>
<value> <string>Alarm</string> </value>
</item>
<item>
<key> <string>sense_method_id</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>title</string> </key>
<value> <string>Check and Update Compute Node Allocation Scope</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="Alarm" module="erp5.portal_type"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>active_sense_method_id</string> </key>
<value> <string>Alarm_checkAndUpdatePersonalComputeNodeAllocationScope</string> </value>
</item>
<item>
<key> <string>description</string> </key>
<value> <string>Automatically turn allocation scope of personal compute_node to closed if the compute_node were never used.</string> </value>
</item>
<item>
<key> <string>enabled</string> </key>
<value> <int>0</int> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>slapos_crm_check_update_personal_allocation_scope</string> </value>
</item>
<item>
<key> <string>periodicity_hour</string> </key>
<value>
<tuple>
<int>22</int>
</tuple>
</value>
</item>
<item>
<key> <string>periodicity_minute</string> </key>
<value>
<tuple>
<int>30</int>
</tuple>
</value>
</item>
<item>
<key> <string>periodicity_minute_frequency</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>periodicity_month</string> </key>
<value>
<tuple/>
</value>
</item>
<item>
<key> <string>periodicity_month_day</string> </key>
<value>
<tuple/>
</value>
</item>
<item>
<key> <string>periodicity_start_date</string> </key>
<value>
<object>
<klass>
<global name="DateTime" module="DateTime.DateTime"/>
</klass>
<tuple>
<none/>
</tuple>
<state>
<tuple>
<float>1399420800.0</float>
<string>GMT</string>
</tuple>
</state>
</object>
</value>
</item>
<item>
<key> <string>periodicity_week</string> </key>
<value>
<tuple/>
</value>
</item>
<item>
<key> <string>portal_type</string> </key>
<value> <string>Alarm</string> </value>
</item>
<item>
<key> <string>sense_method_id</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>title</string> </key>
<value> <string>Check and Update Personal Compute Node Allocation Scope</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
portal = context.getPortalObject()
category_public = portal.restrictedTraverse("portal_categories/allocation_scope/open/public", None)
category_subscription = portal.restrictedTraverse("portal_categories/allocation_scope/open/subscription", None)
category_friend = portal.restrictedTraverse("portal_categories/allocation_scope/open/friend", None)
if category_public is not None:
portal.portal_catalog.searchAndActivate(
portal_type='Compute Node',
default_allocation_scope_uid=[category_public.getUid(), category_friend.getUid(), category_subscription.getUid()],
validation_state="validated",
method_id='ComputeNode_checkAndUpdateAllocationScope',
activate_kw={'tag': tag}
)
context.activate(after_tag=tag).getId()
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PythonScript" module="Products.PythonScripts.PythonScript"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>Script_magic</string> </key>
<value> <int>3</int> </value>
</item>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
</klass>
<tuple/>
<state>
<dictionary>
<item>
<key> <string>_asgns</string> </key>
<value>
<dictionary>
<item>
<key> <string>name_container</string> </key>
<value> <string>container</string> </value>
</item>
<item>
<key> <string>name_context</string> </key>
<value> <string>context</string> </value>
</item>
<item>
<key> <string>name_m_self</string> </key>
<value> <string>script</string> </value>
</item>
<item>
<key> <string>name_subpath</string> </key>
<value> <string>traverse_subpath</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</state>
</object>
</value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string>tag, fixit, params</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>Alarm_checkAndUpdateComputeNodeAllocationScope</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
from DateTime import DateTime
# from erp5.component.module.DateUtils import addToDate
# from Products.ZSQLCatalog.SQLCatalog import Query
portal = context.getPortalObject()
category_personal = portal.restrictedTraverse("portal_categories/allocation_scope/open/personal", None)
if category_personal is not None:
portal.portal_catalog.searchAndActivate(
portal_type='Compute Node',
validation_state='validated',
# XXX - creation_date is not indexed for compute_node
# creation_date=Query(range="max", creation_date=addToDate(DateTime(), {'day': -30})),
default_allocation_scope_uid=category_personal.getUid(),
method_id='ComputeNode_checkAndUpdatePersonalAllocationScope',
activate_kw={'tag': tag})
context.activate(after_tag=tag).getId()
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PythonScript" module="Products.PythonScripts.PythonScript"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>Script_magic</string> </key>
<value> <int>3</int> </value>
</item>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
</klass>
<tuple/>
<state>
<dictionary>
<item>
<key> <string>_asgns</string> </key>
<value>
<dictionary>
<item>
<key> <string>name_container</string> </key>
<value> <string>container</string> </value>
</item>
<item>
<key> <string>name_context</string> </key>
<value> <string>context</string> </value>
</item>
<item>
<key> <string>name_m_self</string> </key>
<value> <string>script</string> </value>
</item>
<item>
<key> <string>name_subpath</string> </key>
<value> <string>traverse_subpath</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</state>
</object>
</value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string>tag, fixit, params</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>Alarm_checkAndUpdatePersonalComputeNodeAllocationScope</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
from DateTime import DateTime
compute_node = context
portal = context.getPortalObject()
allocation_scope = compute_node.getAllocationScope()
compute_node_reference = compute_node.getReference()
if allocation_scope not in ['open/public', 'open/friend', 'open/personal']:
return
if allocation_scope == target_allocation_scope:
# already changed
return
person = compute_node.getSourceAdministrationValue(portal_type="Person")
if not person:
return
if check_service_provider and person.Person_isServiceProvider():
return
edit_kw = {
'allocation_scope': target_allocation_scope,
}
# Create a ticket (or re-open it) for this issue!
request_title = 'Allocation scope of %s changed to %s' % (compute_node_reference,
target_allocation_scope)
request_description = 'Allocation scope has been changed to ' \
'%s for %s' % (target_allocation_scope, compute_node_reference)
support_request = person.Base_getSupportRequestInProgress(
title=request_title,
aggregate=context.getRelativeUrl()
)
if support_request is None:
person.notify(support_request_title=request_title,
support_request_description=request_description,
aggregate=context.getRelativeUrl())
support_request_relative_url = context.REQUEST.get("support_request_relative_url")
support_request = portal.restrictedTraverse(support_request_relative_url)
if support_request is not None:
if support_request.getSimulationState() != "validated":
support_request.validate()
# Send notification message
message = request_description
notification_message = portal.portal_notifications.getDocumentValue(
reference=notification_message_reference)
if notification_message is not None:
mapping_dict = {'compute_node_title':compute_node.getTitle(),
'compute_node_id':compute_node_reference,
'compute_node_url':compute_node.getRelativeUrl(),
'allocation_scope':allocation_scope}
message = notification_message.asText(
substitution_method_parameter_dict={'mapping_dict': mapping_dict})
event = support_request.SupportRequest_getLastEvent(request_title)
if event is None:
support_request.notify(message_title=request_title, message=message)
event = support_request.REQUEST.get("ticket_notified_item")
if event is not None:
# event added, suspend ticket
if portal.portal_workflow.isTransitionPossible(support_request, 'suspend'):
support_request.suspend()
elif not force:
return support_request
compute_node.edit(**edit_kw)
return support_request
elif force:
# Update compute_node event if ticket is not created
compute_node.edit(**edit_kw)
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PythonScript" module="Products.PythonScripts.PythonScript"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>Script_magic</string> </key>
<value> <int>3</int> </value>
</item>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
</klass>
<tuple/>
<state>
<dictionary>
<item>
<key> <string>_asgns</string> </key>
<value>
<dictionary>
<item>
<key> <string>name_container</string> </key>
<value> <string>container</string> </value>
</item>
<item>
<key> <string>name_context</string> </key>
<value> <string>context</string> </value>
</item>
<item>
<key> <string>name_m_self</string> </key>
<value> <string>script</string> </value>
</item>
<item>
<key> <string>name_subpath</string> </key>
<value> <string>traverse_subpath</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</state>
</object>
</value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string>target_allocation_scope=\'open/personal\', notification_message_reference=\'slapos-crm-compute_node_allocation_scope.notification\', check_service_provider=True, force=False</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>ComputeNode_checkAndUpdateAllocationScope</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
if context.ComputeNode_hasContactedRecently():
return
return context.ComputeNode_checkAndUpdateAllocationScope(
target_allocation_scope = 'close/outdated',
notification_message_reference='slapos-crm-compute-node-allocation-scope-closed.notification',
check_service_provider=False,
force=True)
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PythonScript" module="Products.PythonScripts.PythonScript"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>Script_magic</string> </key>
<value> <int>3</int> </value>
</item>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
</klass>
<tuple/>
<state>
<dictionary>
<item>
<key> <string>_asgns</string> </key>
<value>
<dictionary>
<item>
<key> <string>name_container</string> </key>
<value> <string>container</string> </value>
</item>
<item>
<key> <string>name_context</string> </key>
<value> <string>context</string> </value>
</item>
<item>
<key> <string>name_m_self</string> </key>
<value> <string>script</string> </value>
</item>
<item>
<key> <string>name_subpath</string> </key>
<value> <string>traverse_subpath</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</state>
</object>
</value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string>**kw</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>ComputeNode_checkAndUpdatePersonalAllocationScope</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
if REQUEST is not None:
from zExceptions import Unauthorized
raise Unauthorized
if role_id in context.getRoleList():
return True
for assignment in context.contentValues(portal_type="Assignment"):
if assignment.getValidationState() == "open" and \
assignment.getRole() == role_id:
return True
return False
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PythonScript" module="Products.PythonScripts.PythonScript"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>Script_magic</string> </key>
<value> <int>3</int> </value>
</item>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
</klass>
<tuple/>
<state>
<dictionary>
<item>
<key> <string>_asgns</string> </key>
<value>
<dictionary>
<item>
<key> <string>name_container</string> </key>
<value> <string>container</string> </value>
</item>
<item>
<key> <string>name_context</string> </key>
<value> <string>context</string> </value>
</item>
<item>
<key> <string>name_m_self</string> </key>
<value> <string>script</string> </value>
</item>
<item>
<key> <string>name_subpath</string> </key>
<value> <string>traverse_subpath</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</state>
</object>
</value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string>role_id, REQUEST=None</string> </value>
</item>
<item>
<key> <string>_proxy_roles</string> </key>
<value>
<tuple>
<string>Manager</string>
</tuple>
</value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>Person_hasRequiredRole</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PythonScript" module="Products.PythonScripts.PythonScript"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>Script_magic</string> </key>
<value> <int>3</int> </value>
</item>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
</klass>
<tuple/>
<state>
<dictionary>
<item>
<key> <string>_asgns</string> </key>
<value>
<dictionary>
<item>
<key> <string>name_container</string> </key>
<value> <string>container</string> </value>
</item>
<item>
<key> <string>name_context</string> </key>
<value> <string>context</string> </value>
</item>
<item>
<key> <string>name_m_self</string> </key>
<value> <string>script</string> </value>
</item>
<item>
<key> <string>name_subpath</string> </key>
<value> <string>traverse_subpath</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</state>
</object>
</value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>Person_isServiceProvider</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
......@@ -22,7 +22,6 @@
##############################################################################
from erp5.component.test.SlapOSTestCaseMixin import \
SlapOSTestCaseMixin, SlapOSTestCaseMixinWithAbort
from unittest import skip
from DateTime import DateTime
class TestSlapOSCRMCreateRegularisationRequest(SlapOSTestCaseMixin):
......@@ -543,34 +542,6 @@ class TestSlapOSCrmMonitoringCheckComputeNodeState(SlapOSTestCaseMixinWithAbort)
self._test_alarm_check_compute_node_state_not_selected(
allocation_scope='closed/termination')
class TestSlapOSCrmMonitoringCheckComputeNodeAllocationScope(SlapOSTestCaseMixinWithAbort):
def test_alarm_not_allowed_allocation_scope_OpenPublic(self):
self._makeComputeNode()
self.compute_node.edit(allocation_scope = 'open/public')
self.tic()
alarm = self.portal.portal_alarms.\
slapos_crm_check_update_allocation_scope
self._test_alarm(alarm, self.compute_node, "ComputeNode_checkAndUpdateAllocationScope")
def test_alarm_not_allowed_allocation_scope_OpenFriend(self):
self._makeComputeNode()
self.compute_node.edit(allocation_scope = 'open/friend')
self.tic()
alarm = self.portal.portal_alarms.\
slapos_crm_check_update_allocation_scope
self._test_alarm(alarm, self.compute_node, "ComputeNode_checkAndUpdateAllocationScope")
def test_alarm_not_allowed_allocationScope_open_personal(self):
self._makeComputeNode()
self.compute_node.edit(allocation_scope = 'open/personal')
self.tic()
alarm = self.portal.portal_alarms.\
slapos_crm_check_update_allocation_scope
self._test_alarm_not_visited(alarm, self.compute_node, "ComputeNode_checkAndUpdateAllocationScope")
class TestSlapOSCrmMonitoringCheckComputeNodeSoftwareInstallation(SlapOSTestCaseMixinWithAbort):
def test_alarm_run_on_open_public(self):
......@@ -650,70 +621,6 @@ class TestSlapOSCrmMonitoringCheckComputeNodeSoftwareInstallation(SlapOSTestCase
def test_alarm_not_run_on_close_termination(self):
self._test_alarm_not_run_on_close('close/termination')
class TestSlapOSCrmMonitoringCheckComputeNodePersonalAllocationScope(SlapOSTestCaseMixinWithAbort):
def test_alarm_allowed_allocation_scope_OpenPersonal_old_compute_node(self):
self._makeComputeNode()
def getCreationDate(self):
return DateTime() - 31
self.compute_node.edit(allocation_scope = 'open/personal')
from Products.ERP5Type.Base import Base
self._simulateScript("ComputeNode_checkAndUpdatePersonalAllocationScope")
original_get_creation = Base.getCreationDate
Base.getCreationDate = getCreationDate
self.tic()
try:
self.portal.portal_alarms.slapos_crm_check_update_personal_allocation_scope.activeSense()
self.tic()
finally:
Base.getCreationDate = original_get_creation
self._dropScript('ComputeNode_checkAndUpdatePersonalAllocationScope')
self.assertEqual('Visited by ComputeNode_checkAndUpdatePersonalAllocationScope',
self.compute_node.workflow_history['edit_workflow'][-1]['comment'])
@skip('compute node creation date is not indexed')
def test_alarm_allowed_allocation_scope_OpenPersonal_recent_compute_node(self):
self._makeComputeNode()
def getCreationDate(self):
return DateTime() - 28
self.compute_node.edit(allocation_scope = 'open/personal')
from Products.ERP5Type.Base import Base
self._simulateScript("ComputeNode_checkAndUpdatePersonalAllocationScope")
original_get_creation = Base.getCreationDate
Base.getCreationDate = getCreationDate
try:
self.portal.portal_alarms.slapos_crm_check_update_personal_allocation_scope.activeSense()
self.tic()
finally:
Base.getCreationDate = original_get_creation
self._dropScript('ComputeNode_checkAndUpdatePersonalAllocationScope')
self.assertNotEqual('Visited by ComputeNode_checkAndUpdatePersonalAllocationScope',
self.compute_node.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_allowed_allocation_scope_OpenPersonal_already_closed(self):
self._makeComputeNode()
self.compute_node.edit(allocation_scope = 'open/oudated')
self._simulateScript("ComputeNode_checkAndUpdatePersonalAllocationScope")
try:
self.portal.portal_alarms.slapos_crm_check_update_personal_allocation_scope.activeSense()
self.tic()
finally:
self._dropScript('ComputeNode_checkAndUpdatePersonalAllocationScope')
self.assertNotEqual('Visited by ComputeNode_checkAndUpdatePersonalAllocationScope',
self.compute_node.workflow_history['edit_workflow'][-1]['comment'])
class TestSlapOSCrmMonitoringCheckInstanceInError(SlapOSTestCaseMixinWithAbort):
def _makeInstanceTree(self):
......
......@@ -629,165 +629,6 @@ class TestSlapOSBase_getOpenRelatedTicketList(TestCRMSkinsMixin):
open_related_ticket_list = document.Base_getOpenRelatedTicketList()
self.assertEqual(len(open_related_ticket_list), 0)
class TestSlapOSComputeNode_notifyWrongAllocationScope(TestCRMSkinsMixin):
def afterSetUp(self):
TestCRMSkinsMixin.afterSetUp(self)
self._cancelTestSupportRequestList(title="%%TESTCOMPT-%")
def _getGeneratedSupportRequest(self, compute_node):
request_title = '%%We have changed allocation scope for %s' % \
compute_node.getReference()
support_request = self.portal.portal_catalog.getResultValue(
portal_type = 'Support Request',
title = request_title,
simulation_state = 'suspended',
default_aggregate_uid = compute_node.getUid()
)
return support_request
def _makeNotificationMessage(self, reference):
notification_message = self.portal.notification_message_module.newContent(
portal_type="Notification Message",
title='We have changed allocation scope for %s' % reference,
text_content='Test NM content<br/>%s<br/>' % reference,
content_type='text/html',
)
return notification_message.getRelativeUrl()
@simulate('ERP5Site_isSupportRequestCreationClosed', '*args, **kwargs','return 0')
@simulate('NotificationTool_getDocumentValue',
'reference=None',
'assert reference == "slapos-crm-compute_node_allocation_scope.notification"\n' \
'return context.restrictedTraverse(' \
'context.REQUEST["test_ComputeNodeNotAllowedAllocationScope_OpenPublic"])')
def test_ComputeNodeNotAllowedAllocationScope_OpenPublic(self):
compute_node = self._makeComputeNode(owner=self.makePerson(user=0))[0]
person = compute_node.getSourceAdministrationValue()
self.portal.REQUEST['test_ComputeNodeNotAllowedAllocationScope_OpenPublic'] = \
self._makeNotificationMessage(compute_node.getReference())
compute_node.edit(allocation_scope='open/public')
ticket = compute_node.ComputeNode_checkAndUpdateAllocationScope()
self.tic()
self.assertEqual(compute_node.getAllocationScope(), 'open/personal')
#ticket = self._getGeneratedSupportRequest(compute_node)
self.assertNotEqual(None, ticket)
self.assertEqual(ticket.getSimulationState(), 'suspended')
event_list = ticket.getFollowUpRelatedValueList()
self.assertEqual(len(event_list), 1)
event = event_list[0]
self.assertEqual(event.getTitle(),
'Allocation scope of %s changed to %s' % (compute_node.getReference(), 'open/personal'))
self.assertIn(compute_node.getReference(), event.getTextContent())
self.assertEqual(event.getSource(), person.getRelativeUrl())
self.assertEqual(event.getDestination(), ticket.getSourceSection())
@simulate('ERP5Site_isSupportRequestCreationClosed', '*args, **kwargs','return 0')
@simulate('NotificationTool_getDocumentValue',
'reference=None',
'assert reference == "slapos-crm-compute_node_allocation_scope.notification"\n' \
'return context.restrictedTraverse(' \
'context.REQUEST["test_ComputeNodeNotAllowedAllocationScope_OpenFriend"])')
def test_ComputeNodeNotAllowedAllocationScope_OpenFriend(self):
compute_node = self._makeComputeNode(owner=self.makePerson(user=0))[0]
person = compute_node.getSourceAdministrationValue()
self.portal.REQUEST['test_ComputeNodeNotAllowedAllocationScope_OpenFriend'] = \
self._makeNotificationMessage(compute_node.getReference())
friend_person = self.makePerson()
compute_node.edit(allocation_scope='open/friend',
destination_section=friend_person.getRelativeUrl())
ticket = compute_node.ComputeNode_checkAndUpdateAllocationScope()
self.tic()
self.assertEqual(compute_node.getAllocationScope(), 'open/personal')
self.assertEqual(ticket.getSimulationState(), 'suspended')
event_list = ticket.getFollowUpRelatedValueList()
self.assertEqual(len(event_list), 1)
event = event_list[0]
self.assertEqual(event.getTitle(),
'Allocation scope of %s changed to %s' % (compute_node.getReference(), 'open/personal'))
self.assertIn(compute_node.getReference(), event.getTextContent())
self.assertEqual(event.getSource(), person.getRelativeUrl())
self.assertEqual(event.getDestination(), ticket.getSourceSection())
@simulate('ERP5Site_isSupportRequestCreationClosed', '*args, **kwargs','return 0')
@simulate('ComputeNode_hasContactedRecently', '*args, **kwargs','return False')
@simulate('NotificationTool_getDocumentValue',
'reference=None',
'assert reference == "slapos-crm-compute-node-allocation-scope-closed.notification"\n' \
'return context.restrictedTraverse(' \
'context.REQUEST["test_ComputeNodeToCloseAllocationScope_OpenPersonal"])')
def test_ComputeNodeToCloseAllocationScope_OpenPersonal(self):
compute_node = self._makeComputeNode(owner=self.makePerson(user=0))[0]
person = compute_node.getSourceAdministrationValue()
target_allocation_scope = 'close/outdated'
self.portal.REQUEST['test_ComputeNodeToCloseAllocationScope_OpenPersonal'] = \
self._makeNotificationMessage(compute_node.getReference())
compute_node.edit(allocation_scope='open/personal')
support_request = compute_node.ComputeNode_checkAndUpdatePersonalAllocationScope()
self.tic()
self.assertEqual('suspended', support_request.getSimulationState())
self.assertEqual(compute_node.getAllocationScope(), target_allocation_scope)
event_list = support_request.getFollowUpRelatedValueList()
self.assertEqual(len(event_list), 1)
event = event_list[0]
self.assertEqual(event.getTitle(),
'Allocation scope of %s changed to %s' % \
(compute_node.getReference(), target_allocation_scope))
self.assertIn(compute_node.getReference(), event.getTextContent())
self.assertEqual(event.getSource(), person.getRelativeUrl())
self.assertEqual(event.getDestination(), support_request.getSourceSection())
def test_ComputeNodeNormalAllocationScope_OpenPersonal(self):
compute_node = self._makeComputeNode(owner=self.makePerson(user=0))[0]
person = compute_node.getSourceAdministrationValue()
self._updatePersonAssignment(person, 'role/service_provider')
compute_node.edit(allocation_scope='open/personal')
compute_node.ComputeNode_checkAndUpdateAllocationScope()
self.tic()
self.assertEqual(compute_node.getAllocationScope(), 'open/personal')
def test_ComputeNodeAllowedAllocationScope_OpenPublic(self):
compute_node = self._makeComputeNode(owner=self.makePerson(user=0))[0]
person = compute_node.getSourceAdministrationValue()
self._updatePersonAssignment(person, 'role/service_provider')
compute_node.edit(allocation_scope='open/public')
compute_node.ComputeNode_checkAndUpdateAllocationScope()
self.tic()
self.assertEqual(compute_node.getAllocationScope(), 'open/public')
def test_ComputeNodeAllowedAllocationScope_OpenFriend(self):
compute_node = self._makeComputeNode(owner=self.makePerson(user=0))[0]
friend_person = self.makePerson()
person = compute_node.getSourceAdministrationValue()
self._updatePersonAssignment(person, 'role/service_provider')
compute_node.edit(allocation_scope='open/friend',
destination_section=friend_person.getRelativeUrl())
compute_node.ComputeNode_checkAndUpdateAllocationScope()
self.tic()
self.assertEqual(compute_node.getAllocationScope(), 'open/friend')
class TestComputeNode_hasContactedRecently(SlapOSTestCaseMixinWithAbort):
def createSPL(self, compute_node):
......@@ -871,32 +712,6 @@ class TestComputeNode_hasContactedRecently(SlapOSTestCaseMixinWithAbort):
has_contacted = compute_node.ComputeNode_hasContactedRecently()
self.assertFalse(has_contacted)
class TestSlapOSPerson_isServiceProvider(SlapOSTestCaseMixin):
abort_transaction = 1
def test_Person_isServiceProvider(self):
person = self.portal.person_module.template_member\
.Base_createCloneDocument(batch_mode=1)
person.edit(reference='TESTPERSON-%s' % (self.generateNewId(), ))
self.assertFalse(person.Person_isServiceProvider())
person.setRole("service_provider")
self.assertTrue(person.Person_isServiceProvider())
def test_Person_isServiceProvider_assignment(self):
person = self.portal.person_module.template_member\
.Base_createCloneDocument(batch_mode=1)
person.edit(reference='TESTPERSON-%s' % (self.generateNewId(), ))
self.assertFalse(person.Person_isServiceProvider())
assignment = person.newContent(portal_type="Assignment",
role="service_provider")
self.assertFalse(person.Person_isServiceProvider())
assignment.open()
self.assertTrue(person.Person_isServiceProvider())
class TestSlapOSisSupportRequestCreationClosed(TestCRMSkinsMixin):
def afterSetUp(self):
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment