Commit c9598767 authored by Alain Takoudjou's avatar Alain Takoudjou

slapos_crm: add/fix test for create alarm to close personal computer

parent 11ba14cf
from DateTime import DateTime
from Products.ERP5Type.DateUtils import addToDate
from Products.ZSQLCatalog.SQLCatalog import Query
portal = context.getPortalObject()
category_personal = portal.restrictedTraverse("portal_categories/allocation_scope/open/personal", None)
......@@ -6,6 +10,7 @@ if category_personal is not None:
portal.portal_catalog.searchAndActivate(
portal_type='Computer',
validation_state='validated',
creation_date=Query(range="max", creation_date=addToDate(DateTime(), {'day': -30})),
default_allocation_scope_uid=category_personal.getUid(),
method_id='Computer_checkAndUpdatePersonalAllocationScope',
activate_kw={'tag': tag})
......
import json
portal = context.getPortalObject()
computer = context
maximum_days = 31
now_date = DateTime()
if (now_date - computer.getCreationDate()) < maximum_days:
......
......@@ -50,7 +50,7 @@
</item>
<item>
<key> <string>_params</string> </key>
<value> <string></string> </value>
<value> <string>maximum_days=30</string> </value>
</item>
<item>
<key> <string>id</string> </key>
......
......@@ -969,44 +969,58 @@ portal_workflow.doActionFor(context, action='edit_action', comment='Visited by C
def test_alarm_allowed_allocation_scope_OpenPersonal_old_computer(self):
self._makeComputer()
def getCreationDate(self):
return DateTime() - 31
self.computer.edit(allocation_scope = 'open/personal')
def getModificationDate(self):
return DateTime() - 50
from Products.ERP5Type.Base import Base
self._simulateComputer_checkAndUpdatePersonalAllocationScope()
original_get_modification = Base.getModificationDate
Base.getModificationDate = getModificationDate
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.getModificationDate = original_get_modification
Base.getCreationDate = original_get_creation
self._dropComputer_checkAndUpdatePersonalAllocationScope()
self.assertEqual('Visited by Computer_checkAndUpdatePersonalAllocationScope',
self.computer.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_allowed_allocation_scope_OpenPersonalWithSoftwareInstallation(self):
def test_alarm_allowed_allocation_scope_OpenPersonal_recent_computer(self):
self._makeComputer()
def getCreationDate(self):
return DateTime() - 28
self.computer.edit(allocation_scope = 'open/personal')
self._makeSoftwareInstallation()
def getModificationDate(self):
return DateTime() - 50
from Products.ERP5Type.Base import Base
self._simulateComputer_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._dropComputer_checkAndUpdatePersonalAllocationScope()
self.assertNotEqual('Visited by Computer_checkAndUpdatePersonalAllocationScope',
self.computer.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_allowed_allocation_scope_OpenPersonal_already_closed(self):
self._makeComputer()
self.computer.edit(allocation_scope = 'open/oudated')
self._simulateComputer_checkAndUpdatePersonalAllocationScope()
original_get_modification = Base.getModificationDate
Base.getModificationDate = getModificationDate
try:
self.portal.portal_alarms.slapos_crm_check_update_personal_allocation_scope.activeSense()
self.tic()
finally:
Base.getModificationDate = original_get_modification
self._dropComputer_checkAndUpdatePersonalAllocationScope()
self.assertNotEqual('Visited by Computer_checkAndUpdatePersonalAllocationScope',
......@@ -1093,5 +1107,6 @@ portal_workflow.doActionFor(context, action='edit_action', comment='Visited by H
self.assertNotEqual('Visited by HostingSubscription_checkSofwareInstanceState',
host_sub.workflow_history['edit_workflow'][-1]['comment'])
......@@ -2199,11 +2199,12 @@ class TestSlapOSComputer_notifyWrongAllocationScope(testSlapOSMixin):
self.assertEquals(computer.getAllocationScope(), 'open/personal')
#ticket = self._getGeneratedSupportRequest(computer)
self.assertNotEquals(None, ticket)
self.assertEquals(ticket.getSimulationState(), 'validated')
self.assertEquals(ticket.getSimulationState(), 'suspended')
self.assertEqual('Visited by SupportRequest_trySendNotificationMessage ' \
'%s %s %s' % \
('We have changed allocation scope for %s' % computer.getReference(),
('Allocation scope of %s changed to %s' % (computer.getReference(),
'open/personal'),
'Test NM content\n%s\n' % computer.getReference(), person.getRelativeUrl()),
ticket.workflow_history['edit_workflow'][-1]['comment'])
......@@ -2235,18 +2236,19 @@ class TestSlapOSComputer_notifyWrongAllocationScope(testSlapOSMixin):
ticket = computer.Computer_checkAndUpdateAllocationScope()
self.tic()
self.assertEquals(computer.getAllocationScope(), 'open/personal')
self.assertEquals(ticket.getSimulationState(), 'validated')
self.assertEquals(ticket.getSimulationState(), 'suspended')
self.assertEqual('Visited by SupportRequest_trySendNotificationMessage ' \
'%s %s %s' % \
('We have changed allocation scope for %s' % computer.getReference(),
('Allocation scope of %s changed to %s' % (computer.getReference(),
'open/personal'),
'Test NM content\n%s\n' % computer.getReference(), person.getRelativeUrl()),
ticket.workflow_history['edit_workflow'][-1]['comment'])
@simulate('ERP5Site_isSupportRequestCreationClosed', '*args, **kwargs','return 0')
@simulate('Computer_hasContactedRecently', '*args, **kwargs','return False')
@simulate('NotificationTool_getDocumentValue',
'reference=None',
'assert reference == "slapos-crm-computer_personal_allocation_scope.notification"\n' \
'assert reference == "slapos-crm-computer-allocation-scope-closed.notification"\n' \
'return context.restrictedTraverse(' \
'context.REQUEST["test_computerToCloseAllocationScope_OpenPersonal"])')
@simulate('SupportRequest_trySendNotificationMessage',
......@@ -2259,6 +2261,7 @@ class TestSlapOSComputer_notifyWrongAllocationScope(testSlapOSMixin):
def test_computerToCloseAllocationScope_OpenPersonal(self):
computer = self._makeComputer()
person = computer.getSourceAdministrationValue()
target_allocation_scope = 'close/outdated'
self.portal.REQUEST['test_computerToCloseAllocationScope_OpenPersonal'] = \
self._makeNotificationMessage(computer.getReference())
......@@ -2267,11 +2270,12 @@ class TestSlapOSComputer_notifyWrongAllocationScope(testSlapOSMixin):
support_request = computer.Computer_checkAndUpdatePersonalAllocationScope()
self.tic()
self.assertEquals('validated', support_request.getSimulationState())
self.assertEquals(computer.getAllocationScope(), 'close/termination')
self.assertEquals('suspended', support_request.getSimulationState())
self.assertEquals(computer.getAllocationScope(), target_allocation_scope)
self.assertEqual('Visited by SupportRequest_trySendNotificationMessage ' \
'%s %s %s' % \
('We have changed allocation scope for %s' % computer.getReference(),
('Allocation scope of %s changed to %s' % (computer.getReference(),
target_allocation_scope),
'Test NM content\n%s\n' % computer.getReference(), person.getRelativeUrl()),
support_request.workflow_history['edit_workflow'][-1]['comment'])
......@@ -2310,6 +2314,107 @@ class TestSlapOSComputer_notifyWrongAllocationScope(testSlapOSMixin):
self.assertEquals(computer.getAllocationScope(), 'open/friend')
class TestComputer_hasContactedRecently(testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def afterSetUp(self):
super(TestComputer_hasContactedRecently, self).afterSetUp()
def _makeComputer(self):
super(TestComputer_hasContactedRecently, self)._makeComputer()
return self.computer
def createSPL(self, computer):
delivery_template = self.portal.restrictedTraverse(
self.portal.portal_preferences.getPreferredInstanceDeliveryTemplate())
delivery = delivery_template.Base_createCloneDocument(batch_mode=1)
delivery.edit(
title="TEST SPL COMP %s" % computer.getReference(),
start_date=computer.getCreationDate(),
)
delivery.newContent(
portal_type="Sale Packing List Line",
title="SPL Line for %s" % computer.getReference(),
quantity=1,
aggregate_value_list=computer,
)
delivery.confirm(comment="Created from %s" % computer.getRelativeUrl())
delivery.start()
delivery.stop()
delivery.deliver()
return delivery
def test_Computer_hasContactedRecently_newly_created(self):
computer = self._makeComputer()
self.tic()
has_contacted = computer.Computer_hasContactedRecently()
self.assertTrue(has_contacted)
@simulate('Computer_getCreationDate', '*args, **kwargs','return DateTime() - 32')
def test_Computer_hasContactedRecently_no_data(self):
computer = self._makeComputer()
self.tic()
computer.getCreationDate = self.portal.Computer_getCreationDate
has_contacted = computer.Computer_hasContactedRecently()
self.assertFalse(has_contacted)
@simulate('Computer_getCreationDate', '*args, **kwargs','return DateTime() - 32')
def test_Computer_hasContactedRecently_memcached(self):
computer = self._makeComputer()
memcached_dict = self.portal.portal_memcached.getMemcachedDict(
key_prefix='slap_tool',
plugin_path='portal_memcached/default_memcached_plugin')
memcached_dict[computer.getReference()] = json.dumps({
"created_at": DateTime().strftime("%Y/%m/%d %H:%M")
})
self.tic()
computer.getCreationDate = self.portal.Computer_getCreationDate
has_contacted = computer.Computer_hasContactedRecently()
self.assertTrue(has_contacted)
@simulate('Computer_getCreationDate', '*args, **kwargs','return DateTime() - 32')
def test_Computer_hasContactedRecently_memcached_oudated_no_spl(self):
computer = self._makeComputer()
memcached_dict = self.portal.portal_memcached.getMemcachedDict(
key_prefix='slap_tool',
plugin_path='portal_memcached/default_memcached_plugin')
memcached_dict[computer.getReference()] = json.dumps({
"created_at": (DateTime() - 32).strftime("%Y/%m/%d %H:%M")
})
self.tic()
computer.getCreationDate = self.portal.Computer_getCreationDate
has_contacted = computer.Computer_hasContactedRecently()
self.assertFalse(has_contacted)
@simulate('Computer_getCreationDate', '*args, **kwargs','return DateTime() - 32')
def test_Computer_hasContactedRecently_memcached_oudated_with_spl(self):
computer = self._makeComputer()
memcached_dict = self.portal.portal_memcached.getMemcachedDict(
key_prefix='slap_tool',
plugin_path='portal_memcached/default_memcached_plugin')
memcached_dict[computer.getReference()] = json.dumps({
"created_at": (DateTime() - 32).strftime("%Y/%m/%d %H:%M")
})
self.createSPL(computer)
self.tic()
computer.getCreationDate = self.portal.Computer_getCreationDate
has_contacted = computer.Computer_hasContactedRecently()
self.assertFalse(has_contacted)
class TestSlapOSPerson_isServiceProvider(testSlapOSMixin):
def beforeTearDown(self):
......@@ -2686,10 +2791,14 @@ class TestSlapOSGenerateSupportRequestForSlapOS(testSlapOSMixin):
self.assertEqual(support_request, same_support_request)
def test_Base_generateSupportRequestForSlapOS_inprogress(self):
in_progress = self.portal.support_request_module.getRelativeUrl()
self._makeComputer()
title = "Test Support Request %s" % self.computer.getRelativeUrl()
support_request = self.computer.Base_generateSupportRequestForSlapOS(
title, title, self.computer.getRelativeUrl())
in_progress = support_request.getRelativeUrl()
self.portal.REQUEST.set("support_request_in_progress", in_progress)
self._makeComputer()
title = "Test Support Request %s" % self.computer.getRelativeUrl()
support_request = self.computer.Base_generateSupportRequestForSlapOS(
title, title, self.computer.getRelativeUrl())
......@@ -3332,7 +3441,7 @@ class TestSupportRequestTrySendNotificationMessage(testSlapOSMixin):
self.assertNotEquals(None, first_event.getStartDate())
self.assertEquals("service_module/slapos_crm_information",
first_event.getResource())
self.assertEquals(first_event.getSource(), person.getRelativeUrl())
self.assertEquals(first_event.getDestination(), person.getRelativeUrl())
self.assertEquals(first_event.getFollowUp(), support_request.getRelativeUrl())
event = support_request.SupportRequest_trySendNotificationMessage(
......@@ -3346,7 +3455,7 @@ class TestSupportRequestTrySendNotificationMessage(testSlapOSMixin):
self.assertNotEquals(None, event.getStartDate())
self.assertEquals("service_module/slapos_crm_information",
event.getResource())
self.assertEquals(event.getSource(), person.getRelativeUrl())
self.assertEquals(event.getDestination(), person.getRelativeUrl())
title += "__zz"
event = support_request.SupportRequest_trySendNotificationMessage(
......@@ -3359,7 +3468,7 @@ class TestSupportRequestTrySendNotificationMessage(testSlapOSMixin):
self.assertNotEquals(None, event.getStartDate())
self.assertEquals("service_module/slapos_crm_information",
event.getResource())
self.assertEquals(event.getSource(), person.getRelativeUrl())
self.assertEquals(event.getDestination(), person.getRelativeUrl())
another_support_request = self.portal.support_request_module.newContent(\
title=title, description=title,
......@@ -3384,7 +3493,7 @@ class TestSupportRequestTrySendNotificationMessage(testSlapOSMixin):
self.assertNotEquals(None, another_first_event.getStartDate())
self.assertEquals("service_module/slapos_crm_information",
another_first_event.getResource())
self.assertEquals(another_first_event.getSource(), person.getRelativeUrl())
self.assertEquals(another_first_event.getDestination(), person.getRelativeUrl())
self.assertEquals(another_first_event.getFollowUp(),
another_support_request.getRelativeUrl())
......
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