Skip to content

S
slapos.core
Commit acbe99d3 authored by Alain Takoudjou's avatar Alain Takoudjou
Browse files
Options

After Migrade: Delete old tests scripts from fileSystem in slapos_crm

parent 0ba4d720
Hide whitespace changes
Inline Side-by-side
Showing with 0 additions and 3961 deletions
master/bt5/slapos_crm/TestTemplateItem/testSlapOSCRMAlarm.py deleted 100644 → 0
View file @ 0ba4d720
# Copyright (c) 2013 Nexedi SA and Contributors. All Rights Reserved.
import transaction
from Products.SlapOS.tests.testSlapOSMixin import \
testSlapOSMixin
from Products.ERP5Type.tests.utils import createZODBPythonScript
class TestSlapOSCRMCreateRegularisationRequest(testSlapOSMixin):
def _simulatePerson_checkToCreateRegularisationRequest(self):
script_name = 'Person_checkToCreateRegularisationRequest'
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 Person_checkToCreateRegularisationRequest') """ )
transaction.commit()
def _dropPerson_checkToCreateRegularisationRequest(self):
script_name = 'Person_checkToCreateRegularisationRequest'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
transaction.commit()
def test_alarm_expected_person(self):
new_id = self.generateNewId()
person = self.portal.person_module.newContent(
portal_type='Person',
title="Test person %s" % new_id,
reference="TESTPERS_%s" % new_id,
default_email_text="%s@example.org" % new_id,
)
person.validate()
self.tic()
self._simulatePerson_checkToCreateRegularisationRequest()
try:
self.portal.portal_alarms.\
slapos_crm_create_regularisation_request.activeSense()
self.tic()
finally:
self._dropPerson_checkToCreateRegularisationRequest()
self.assertEqual(
'Visited by Person_checkToCreateRegularisationRequest',
person.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_no_email(self):
new_id = self.generateNewId()
person = self.portal.person_module.newContent(
portal_type='Person',
title="Test person %s" % new_id,
reference="TESTPERS_%s" % new_id,
)
person.validate()
self.tic()
self._simulatePerson_checkToCreateRegularisationRequest()
try:
self.portal.portal_alarms.\
slapos_crm_create_regularisation_request.activeSense()
self.tic()
finally:
self._dropPerson_checkToCreateRegularisationRequest()
self.assertNotEqual(
'Visited by Person_checkToCreateRegularisationRequest',
person.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_no_reference(self):
new_id = self.generateNewId()
person = self.portal.person_module.newContent(
portal_type='Person',
title="Test person %s" % new_id,
default_email_text="%s@example.org" % new_id,
)
person.validate()
self.tic()
self._simulatePerson_checkToCreateRegularisationRequest()
try:
self.portal.portal_alarms.\
slapos_crm_create_regularisation_request.activeSense()
self.tic()
finally:
self._dropPerson_checkToCreateRegularisationRequest()
self.assertNotEqual(
'Visited by Person_checkToCreateRegularisationRequest',
person.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_not_validated(self):
new_id = self.generateNewId()
person = self.portal.person_module.newContent(
portal_type='Person',
title="Test person %s" % new_id,
reference="TESTPERS_%s" % new_id,
default_email_text="%s@example.org" % new_id,
)
person.validate()
person.invalidate()
self.tic()
self._simulatePerson_checkToCreateRegularisationRequest()
try:
self.portal.portal_alarms.\
slapos_crm_create_regularisation_request.activeSense()
self.tic()
finally:
self._dropPerson_checkToCreateRegularisationRequest()
self.assertNotEqual(
'Visited by Person_checkToCreateRegularisationRequest',
person.workflow_history['edit_workflow'][-1]['comment'])
class TestSlapOSCrmInvalidateSuspendedRegularisationRequest(testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def createRegularisationRequest(self):
new_id = self.generateNewId()
return self.portal.regularisation_request_module.newContent(
portal_type='Regularisation Request',
title="Test Reg. Req.%s" % new_id,
reference="TESTREGREQ-%s" % new_id,
)
def _simulateRegularisationRequest_invalidateIfPersonBalanceIsOk(self):
script_name = 'RegularisationRequest_invalidateIfPersonBalanceIsOk'
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 RegularisationRequest_invalidateIfPersonBalanceIsOk') """ )
transaction.commit()
def _dropRegularisationRequest_invalidateIfPersonBalanceIsOk(self):
script_name = 'RegularisationRequest_invalidateIfPersonBalanceIsOk'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
transaction.commit()
def test_alarm_not_suspended_regularisation_request(self):
ticket = self.createRegularisationRequest()
ticket.validate()
self.tic()
self._simulateRegularisationRequest_invalidateIfPersonBalanceIsOk()
try:
self.portal.portal_alarms.\
slapos_crm_invalidate_suspended_regularisation_request.activeSense()
self.tic()
finally:
self._dropRegularisationRequest_invalidateIfPersonBalanceIsOk()
self.assertNotEqual(
'Visited by RegularisationRequest_invalidateIfPersonBalanceIsOk',
ticket.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_suspended_regularisation_request(self):
ticket = self.createRegularisationRequest()
ticket.validate()
ticket.suspend()
self.tic()
self._simulateRegularisationRequest_invalidateIfPersonBalanceIsOk()
try:
self.portal.portal_alarms.\
slapos_crm_invalidate_suspended_regularisation_request.activeSense()
self.tic()
finally:
self._dropRegularisationRequest_invalidateIfPersonBalanceIsOk()
self.assertEqual(
'Visited by RegularisationRequest_invalidateIfPersonBalanceIsOk',
ticket.workflow_history['edit_workflow'][-1]['comment'])
class TestSlapOSCrmCancelInvoiceRelatedToSuspendedRegularisationRequest(testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def createRegularisationRequest(self):
new_id = self.generateNewId()
return self.portal.regularisation_request_module.newContent(
portal_type='Regularisation Request',
title="Test Reg. Req.%s" % new_id,
reference="TESTREGREQ-%s" % new_id,
)
def _simulateRegularisationRequest_cancelInvoiceIfPersonOpenOrderIsEmpty(self):
script_name = 'RegularisationRequest_cancelInvoiceIfPersonOpenOrderIsEmpty'
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 RegularisationRequest_cancelInvoiceIfPersonOpenOrderIsEmpty') """ )
transaction.commit()
def _dropRegularisationRequest_cancelInvoiceIfPersonOpenOrderIsEmpty(self):
script_name = 'RegularisationRequest_cancelInvoiceIfPersonOpenOrderIsEmpty'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
transaction.commit()
def test_alarm_not_suspended_regularisation_request(self):
ticket = self.createRegularisationRequest()
ticket.validate()
self.tic()
self._simulateRegularisationRequest_cancelInvoiceIfPersonOpenOrderIsEmpty()
try:
self.portal.portal_alarms.\
slapos_crm_cancel_invoice.activeSense()
self.tic()
finally:
self._dropRegularisationRequest_cancelInvoiceIfPersonOpenOrderIsEmpty()
self.assertNotEqual(
'Visited by RegularisationRequest_cancelInvoiceIfPersonOpenOrderIsEmpty',
ticket.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_suspended_regularisation_request(self):
ticket = self.createRegularisationRequest()
ticket.validate()
ticket.suspend()
self.tic()
self._simulateRegularisationRequest_cancelInvoiceIfPersonOpenOrderIsEmpty()
try:
self.portal.portal_alarms.\
slapos_crm_cancel_invoice.activeSense()
self.tic()
finally:
self._dropRegularisationRequest_cancelInvoiceIfPersonOpenOrderIsEmpty()
self.assertEqual(
'Visited by RegularisationRequest_cancelInvoiceIfPersonOpenOrderIsEmpty',
ticket.workflow_history['edit_workflow'][-1]['comment'])
class TestSlapOSCrmTriggerEscalationOnAcknowledgmentRegularisationRequest(testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def createRegularisationRequest(self):
new_id = self.generateNewId()
return self.portal.regularisation_request_module.newContent(
portal_type='Regularisation Request',
title="Test Reg. Req.%s" % new_id,
reference="TESTREGREQ-%s" % new_id,
)
def _simulateRegularisationRequest_triggerAcknowledgmentEscalation(self):
script_name = 'RegularisationRequest_triggerAcknowledgmentEscalation'
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 RegularisationRequest_triggerAcknowledgmentEscalation') """ )
transaction.commit()
def _dropRegularisationRequest_triggerAcknowledgmentEscalation(self):
script_name = 'RegularisationRequest_triggerAcknowledgmentEscalation'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
transaction.commit()
def test_alarm_matching_regularisation_request(self):
ticket = self.createRegularisationRequest()
ticket.edit(resource='service_module/slapos_crm_acknowledgement')
ticket.validate()
ticket.suspend()
self.tic()
self._simulateRegularisationRequest_triggerAcknowledgmentEscalation()
try:
self.portal.portal_alarms.\
slapos_crm_trigger_acknowledgment_escalation.activeSense()
self.tic()
finally:
self._dropRegularisationRequest_triggerAcknowledgmentEscalation()
self.assertEqual(
'Visited by RegularisationRequest_triggerAcknowledgmentEscalation',
ticket.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_not_suspended(self):
ticket = self.createRegularisationRequest()
ticket.edit(resource='service_module/slapos_crm_acknowledgement')
ticket.validate()
self.tic()
self._simulateRegularisationRequest_triggerAcknowledgmentEscalation()
try:
self.portal.portal_alarms.\
slapos_crm_trigger_acknowledgment_escalation.activeSense()
self.tic()
finally:
self._dropRegularisationRequest_triggerAcknowledgmentEscalation()
self.assertNotEqual(
'Visited by RegularisationRequest_triggerAcknowledgmentEscalation',
ticket.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_not_expected_resource(self):
ticket = self.createRegularisationRequest()
ticket.validate()
ticket.suspend()
self.tic()
self._simulateRegularisationRequest_triggerAcknowledgmentEscalation()
try:
self.portal.portal_alarms.\
slapos_crm_trigger_acknowledgment_escalation.activeSense()
self.tic()
finally:
self._dropRegularisationRequest_triggerAcknowledgmentEscalation()
self.assertNotEqual(
'Visited by RegularisationRequest_triggerAcknowledgmentEscalation',
ticket.workflow_history['edit_workflow'][-1]['comment'])
class TestSlapOSCrmTriggerEscalationOnStopReminderRegularisationRequest(testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def createRegularisationRequest(self):
new_id = self.generateNewId()
return self.portal.regularisation_request_module.newContent(
portal_type='Regularisation Request',
title="Test Reg. Req.%s" % new_id,
reference="TESTREGREQ-%s" % new_id,
)
def _simulateRegularisationRequest_triggerStopReminderEscalation(self):
script_name = 'RegularisationRequest_triggerStopReminderEscalation'
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 RegularisationRequest_triggerStopReminderEscalation') """ )
transaction.commit()
def _dropRegularisationRequest_triggerStopReminderEscalation(self):
script_name = 'RegularisationRequest_triggerStopReminderEscalation'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
transaction.commit()
def test_alarm_matching_regularisation_request(self):
ticket = self.createRegularisationRequest()
ticket.edit(resource='service_module/slapos_crm_stop_reminder')
ticket.validate()
ticket.suspend()
self.tic()
self._simulateRegularisationRequest_triggerStopReminderEscalation()
try:
self.portal.portal_alarms.\
slapos_crm_trigger_stop_reminder_escalation.activeSense()
self.tic()
finally:
self._dropRegularisationRequest_triggerStopReminderEscalation()
self.assertEqual(
'Visited by RegularisationRequest_triggerStopReminderEscalation',
ticket.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_not_suspended(self):
ticket = self.createRegularisationRequest()
ticket.edit(resource='service_module/slapos_crm_stop_reminder')
ticket.validate()
self.tic()
self._simulateRegularisationRequest_triggerStopReminderEscalation()
try:
self.portal.portal_alarms.\
slapos_crm_trigger_stop_reminder_escalation.activeSense()
self.tic()
finally:
self._dropRegularisationRequest_triggerStopReminderEscalation()
self.assertNotEqual(
'Visited by RegularisationRequest_triggerStopReminderEscalation',
ticket.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_not_expected_resource(self):
ticket = self.createRegularisationRequest()
ticket.validate()
ticket.suspend()
self.tic()
self._simulateRegularisationRequest_triggerStopReminderEscalation()
try:
self.portal.portal_alarms.\
slapos_crm_trigger_stop_reminder_escalation.activeSense()
self.tic()
finally:
self._dropRegularisationRequest_triggerStopReminderEscalation()
self.assertNotEqual(
'Visited by RegularisationRequest_triggerStopReminderEscalation',
ticket.workflow_history['edit_workflow'][-1]['comment'])
class TestSlapOSCrmTriggerEscalationOnStopAcknowledgmentRegularisationRequest(testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def createRegularisationRequest(self):
new_id = self.generateNewId()
return self.portal.regularisation_request_module.newContent(
portal_type='Regularisation Request',
title="Test Reg. Req.%s" % new_id,
reference="TESTREGREQ-%s" % new_id,
)
def _simulateRegularisationRequest_triggerStopAcknowledgmentEscalation(self):
script_name = 'RegularisationRequest_triggerStopAcknowledgmentEscalation'
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 RegularisationRequest_triggerStopAcknowledgmentEscalation') """ )
transaction.commit()
def _dropRegularisationRequest_triggerStopAcknowledgmentEscalation(self):
script_name = 'RegularisationRequest_triggerStopAcknowledgmentEscalation'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
transaction.commit()
def test_alarm_matching_regularisation_request(self):
ticket = self.createRegularisationRequest()
ticket.edit(resource='service_module/slapos_crm_stop_acknowledgement')
ticket.validate()
ticket.suspend()
self.tic()
self._simulateRegularisationRequest_triggerStopAcknowledgmentEscalation()
try:
self.portal.portal_alarms.\
slapos_crm_trigger_stop_acknowledgment_escalation.activeSense()
self.tic()
finally:
self._dropRegularisationRequest_triggerStopAcknowledgmentEscalation()
self.assertEqual(
'Visited by RegularisationRequest_triggerStopAcknowledgmentEscalation',
ticket.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_not_suspended(self):
ticket = self.createRegularisationRequest()
ticket.edit(resource='service_module/slapos_crm_stop_acknowledgement')
ticket.validate()
self.tic()
self._simulateRegularisationRequest_triggerStopAcknowledgmentEscalation()
try:
self.portal.portal_alarms.\
slapos_crm_trigger_stop_acknowledgment_escalation.activeSense()
self.tic()
finally:
self._dropRegularisationRequest_triggerStopAcknowledgmentEscalation()
self.assertNotEqual(
'Visited by RegularisationRequest_triggerStopAcknowledgmentEscalation',
ticket.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_not_expected_resource(self):
ticket = self.createRegularisationRequest()
ticket.validate()
ticket.suspend()
self.tic()
self._simulateRegularisationRequest_triggerStopAcknowledgmentEscalation()
try:
self.portal.portal_alarms.\
slapos_crm_trigger_stop_acknowledgment_escalation.activeSense()
self.tic()
finally:
self._dropRegularisationRequest_triggerStopAcknowledgmentEscalation()
self.assertNotEqual(
'Visited by RegularisationRequest_triggerStopAcknowledgmentEscalation',
ticket.workflow_history['edit_workflow'][-1]['comment'])
class TestSlapOSCrmTriggerEscalationOnDeleteReminderRegularisationRequest(testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def createRegularisationRequest(self):
new_id = self.generateNewId()
return self.portal.regularisation_request_module.newContent(
portal_type='Regularisation Request',
title="Test Reg. Req.%s" % new_id,
reference="TESTREGREQ-%s" % new_id,
)
def _simulateRegularisationRequest_triggerDeleteReminderEscalation(self):
script_name = 'RegularisationRequest_triggerDeleteReminderEscalation'
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 RegularisationRequest_triggerDeleteReminderEscalation') """ )
transaction.commit()
def _dropRegularisationRequest_triggerDeleteReminderEscalation(self):
script_name = 'RegularisationRequest_triggerDeleteReminderEscalation'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
transaction.commit()
def test_alarm_matching_regularisation_request(self):
ticket = self.createRegularisationRequest()
ticket.edit(resource='service_module/slapos_crm_delete_reminder')
ticket.validate()
ticket.suspend()
self.tic()
self._simulateRegularisationRequest_triggerDeleteReminderEscalation()
try:
self.portal.portal_alarms.\
slapos_crm_trigger_delete_reminder_escalation.activeSense()
self.tic()
finally:
self._dropRegularisationRequest_triggerDeleteReminderEscalation()
self.assertEqual(
'Visited by RegularisationRequest_triggerDeleteReminderEscalation',
ticket.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_not_suspended(self):
ticket = self.createRegularisationRequest()
ticket.edit(resource='service_module/slapos_crm_delete_reminder')
ticket.validate()
self.tic()
self._simulateRegularisationRequest_triggerDeleteReminderEscalation()
try:
self.portal.portal_alarms.\
slapos_crm_trigger_delete_reminder_escalation.activeSense()
self.tic()
finally:
self._dropRegularisationRequest_triggerDeleteReminderEscalation()
self.assertNotEqual(
'Visited by RegularisationRequest_triggerDeleteReminderEscalation',
ticket.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_not_expected_resource(self):
ticket = self.createRegularisationRequest()
ticket.validate()
ticket.suspend()
self.tic()
self._simulateRegularisationRequest_triggerDeleteReminderEscalation()
try:
self.portal.portal_alarms.\
slapos_crm_trigger_delete_reminder_escalation.activeSense()
self.tic()
finally:
self._dropRegularisationRequest_triggerDeleteReminderEscalation()
self.assertNotEqual(
'Visited by RegularisationRequest_triggerDeleteReminderEscalation',
ticket.workflow_history['edit_workflow'][-1]['comment'])
class TestSlapOSCrmStopHostingSubscription(testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def createRegularisationRequest(self):
new_id = self.generateNewId()
return self.portal.regularisation_request_module.newContent(
portal_type='Regularisation Request',
title="Test Reg. Req.%s" % new_id,
reference="TESTREGREQ-%s" % new_id,
)
def _simulateRegularisationRequest_stopHostingSubscriptionList(self):
script_name = 'RegularisationRequest_stopHostingSubscriptionList'
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 RegularisationRequest_stopHostingSubscriptionList') """ )
transaction.commit()
def _dropRegularisationRequest_stopHostingSubscriptionList(self):
script_name = 'RegularisationRequest_stopHostingSubscriptionList'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
transaction.commit()
def test_alarm_matching_regularisation_request(self):
ticket = self.createRegularisationRequest()
ticket.edit(resource='service_module/slapos_crm_delete_reminder')
ticket.validate()
ticket.suspend()
self.tic()
self._simulateRegularisationRequest_stopHostingSubscriptionList()
try:
self.portal.portal_alarms.\
slapos_crm_stop_hosting_subscription.activeSense()
self.tic()
finally:
self._dropRegularisationRequest_stopHostingSubscriptionList()
self.assertEqual(
'Visited by RegularisationRequest_stopHostingSubscriptionList',
ticket.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_matching_regularisation_request_2(self):
ticket = self.createRegularisationRequest()
ticket.edit(resource='service_module/slapos_crm_stop_acknowledgement')
ticket.validate()
ticket.suspend()
self.tic()
self._simulateRegularisationRequest_stopHostingSubscriptionList()
try:
self.portal.portal_alarms.\
slapos_crm_stop_hosting_subscription.activeSense()
self.tic()
finally:
self._dropRegularisationRequest_stopHostingSubscriptionList()
self.assertEqual(
'Visited by RegularisationRequest_stopHostingSubscriptionList',
ticket.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_not_suspended(self):
ticket = self.createRegularisationRequest()
ticket.edit(resource='service_module/slapos_crm_stop_acknowledgement')
ticket.validate()
self.tic()
self._simulateRegularisationRequest_stopHostingSubscriptionList()
try:
self.portal.portal_alarms.\
slapos_crm_stop_hosting_subscription.activeSense()
self.tic()
finally:
self._dropRegularisationRequest_stopHostingSubscriptionList()
self.assertNotEqual(
'Visited by RegularisationRequest_stopHostingSubscriptionList',
ticket.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_other_resource(self):
ticket = self.createRegularisationRequest()
ticket.edit(resource='service_module/slapos_acknowledgement')
ticket.validate()
ticket.suspend()
self.tic()
self._simulateRegularisationRequest_stopHostingSubscriptionList()
try:
self.portal.portal_alarms.\
slapos_crm_stop_hosting_subscription.activeSense()
self.tic()
finally:
self._dropRegularisationRequest_stopHostingSubscriptionList()
self.assertNotEqual(
'Visited by RegularisationRequest_stopHostingSubscriptionList',
ticket.workflow_history['edit_workflow'][-1]['comment'])
class TestSlapOSCrmDeleteHostingSubscription(testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def createRegularisationRequest(self):
new_id = self.generateNewId()
return self.portal.regularisation_request_module.newContent(
portal_type='Regularisation Request',
title="Test Reg. Req.%s" % new_id,
reference="TESTREGREQ-%s" % new_id,
)
def _simulateRegularisationRequest_deleteHostingSubscriptionList(self):
script_name = 'RegularisationRequest_deleteHostingSubscriptionList'
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 RegularisationRequest_deleteHostingSubscriptionList') """ )
transaction.commit()
def _dropRegularisationRequest_deleteHostingSubscriptionList(self):
script_name = 'RegularisationRequest_deleteHostingSubscriptionList'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
transaction.commit()
def test_alarm_matching_regularisation_request(self):
ticket = self.createRegularisationRequest()
ticket.edit(resource='service_module/slapos_crm_delete_acknowledgement')
ticket.validate()
ticket.suspend()
self.tic()
self._simulateRegularisationRequest_deleteHostingSubscriptionList()
try:
self.portal.portal_alarms.\
slapos_crm_delete_hosting_subscription.activeSense()
self.tic()
finally:
self._dropRegularisationRequest_deleteHostingSubscriptionList()
self.assertEqual(
'Visited by RegularisationRequest_deleteHostingSubscriptionList',
ticket.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_not_suspended(self):
ticket = self.createRegularisationRequest()
ticket.edit(resource='service_module/slapos_crm_delete_acknowledgement')
ticket.validate()
self.tic()
self._simulateRegularisationRequest_deleteHostingSubscriptionList()
try:
self.portal.portal_alarms.\
slapos_crm_delete_hosting_subscription.activeSense()
self.tic()
finally:
self._dropRegularisationRequest_deleteHostingSubscriptionList()
self.assertNotEqual(
'Visited by RegularisationRequest_deleteHostingSubscriptionList',
ticket.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_other_resource(self):
ticket = self.createRegularisationRequest()
ticket.edit(resource='service_module/slapos_crm_delete_reminder')
ticket.validate()
ticket.suspend()
self.tic()
self._simulateRegularisationRequest_deleteHostingSubscriptionList()
try:
self.portal.portal_alarms.\
slapos_crm_delete_hosting_subscription.activeSense()
self.tic()
finally:
self._dropRegularisationRequest_deleteHostingSubscriptionList()
self.assertNotEqual(
'Visited by RegularisationRequest_deleteHostingSubscriptionList',
ticket.workflow_history['edit_workflow'][-1]['comment'])
master/bt5/slapos_crm/TestTemplateItem/testSlapOSCRMSkins.py deleted 100644 → 0
View file @ 0ba4d720
# Copyright (c) 2013 Nexedi SA and Contributors. All Rights Reserved.
import transaction
from Products.SlapOS.tests.testSlapOSMixin import \
testSlapOSMixin
from zExceptions import Unauthorized
from DateTime import DateTime
from functools import wraps
from Products.ERP5Type.tests.utils import createZODBPythonScript
import difflib
import json
def simulate(script_id, params_string, code_string):
def upperWrap(f):
@wraps(f)
def decorated(self, *args, **kw):
if script_id in self.portal.portal_skins.custom.objectIds():
raise ValueError('Precondition failed: %s exists in custom' % script_id)
createZODBPythonScript(self.portal.portal_skins.custom,
script_id, params_string, code_string)
try:
result = f(self, *args, **kw)
finally:
if script_id in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_id)
transaction.commit()
return result
return decorated
return upperWrap
class TestSlapOSPerson_checkToCreateRegularisationRequest(testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def createPerson(self):
new_id = self.generateNewId()
return self.portal.person_module.newContent(
portal_type='Person',
title="Person %s" % new_id,
reference="TESTPERS-%s" % new_id,
)
@simulate('Entity_statBalance', '*args, **kwargs', 'return "1"')
def test_addRegularisationRequest_payment_requested(self):
for preference in \
self.portal.portal_catalog(portal_type="System Preference"):
preference = preference.getObject()
if preference.getPreferenceState() == 'global':
preference.setPreferredSlaposWebSiteUrl('http://foobar.org/')
person = self.createPerson()
before_date = DateTime()
ticket, event = person.Person_checkToCreateRegularisationRequest()
after_date = DateTime()
self.assertEquals(ticket.getPortalType(), 'Regularisation Request')
self.assertEquals(ticket.getSimulationState(), 'suspended')
self.assertEquals(ticket.getSourceProject(), person.getRelativeUrl())
self.assertEquals(ticket.getResource(),
'service_module/slapos_crm_acknowledgement')
self.assertEquals(ticket.getTitle(),
'Account regularisation expected for "%s"' % person.getTitle())
self.assertEquals(ticket.getDestination(),
person.getRelativeUrl())
self.assertEquals(ticket.getDestinationDecision(),
person.getRelativeUrl())
self.assertEquals(event.getPortalType(), 'Mail Message')
self.assertEquals(event.getResource(),
'service_module/slapos_crm_acknowledgement')
self.assertTrue(event.getStartDate() >= before_date)
self.assertTrue(event.getStopDate() <= after_date)
self.assertEquals(event.getTitle(), "Invoice payment requested")
self.assertEquals(event.getDestination(),
person.getRelativeUrl())
self.assertEquals(event.getSource(),
ticket.getSource())
expected_text_content = """Dear user,
A new invoice has been generated.
You can access it in your invoice section at http://foobar.org/.
Do not hesitate to visit the web forum (http://community.slapos.org/forum) in case of question.
Regards,
The slapos team
"""
self.assertEquals(event.getTextContent(), expected_text_content,
'\n'.join([x for x in difflib.unified_diff(
event.getTextContent().splitlines(),
expected_text_content.splitlines())]))
self.assertEquals(event.getSimulationState(), 'delivered')
@simulate('NotificationTool_getDocumentValue',
'reference=None',
'assert reference == "slapos-crm.create.regularisation.request"\n' \
'return context.restrictedTraverse(' \
'context.REQUEST["test_addRegularisationRequest_notification_message"])')
@simulate('Entity_statBalance', '*args, **kwargs', 'return "1"')
def test_addRegularisationRequest_notification_message(self):
for preference in \
self.portal.portal_catalog(portal_type="System Preference"):
preference = preference.getObject()
if preference.getPreferenceState() == 'global':
preference.setPreferredSlaposWebSiteUrl('http://foobar.org/')
person = self.createPerson()
new_id = self.generateNewId()
notification_message = self.portal.notification_message_module.newContent(
portal_type="Notification Message",
title='Test NM title %s' % new_id,
text_content='Test NM content<br/>%s<br/>' % new_id,
content_type='text/html',
)
self.portal.REQUEST\
['test_addRegularisationRequest_notification_message'] = \
notification_message.getRelativeUrl()
before_date = DateTime()
ticket, event = person.Person_checkToCreateRegularisationRequest()
after_date = DateTime()
self.assertEquals(ticket.getPortalType(), 'Regularisation Request')
self.assertEquals(ticket.getSimulationState(), 'suspended')
self.assertEquals(ticket.getSourceProject(), person.getRelativeUrl())
self.assertEquals(ticket.getResource(),
'service_module/slapos_crm_acknowledgement')
self.assertEquals(ticket.getTitle(),
'Account regularisation expected for "%s"' % person.getTitle())
self.assertEquals(ticket.getDestination(),
person.getRelativeUrl())
self.assertEquals(ticket.getDestinationDecision(),
person.getRelativeUrl())
self.assertEquals(event.getPortalType(), 'Mail Message')
self.assertEquals(event.getResource(),
'service_module/slapos_crm_acknowledgement')
self.assertTrue(event.getStartDate() >= before_date)
self.assertTrue(event.getStopDate() <= after_date)
self.assertEquals(event.getTitle(),
'Test NM title %s' % new_id)
self.assertEquals(event.getDestination(),
person.getRelativeUrl())
self.assertEquals(event.getSource(),
ticket.getSource())
expected_text_content = 'Test NM content\n%s\n' % new_id
self.assertEquals(event.getTextContent(), expected_text_content,
'\n'.join([x for x in difflib.unified_diff(
event.getTextContent().splitlines(),
expected_text_content.splitlines())]))
self.assertEquals(event.getSimulationState(), 'delivered')
# def test_addRegularisationRequest_do_not_duplicate_ticket(self):
# person = self.createPerson()
# ticket = person.Person_checkToCreateRegularisationRequest()
# ticket2 = person.Person_checkToCreateRegularisationRequest()
# self.assertEquals(ticket.getRelativeUrl(), ticket2.getRelativeUrl())
@simulate('Entity_statBalance', '*args, **kwargs', 'return "1"')
def test_addRegularisationRequest_do_not_duplicate_ticket_if_not_reindexed(self):
person = self.createPerson()
ticket, event = person.Person_checkToCreateRegularisationRequest()
transaction.commit()
ticket2, event2 = person.Person_checkToCreateRegularisationRequest()
self.assertNotEquals(ticket, None)
self.assertNotEquals(event, None)
self.assertEquals(ticket2, None)
self.assertEquals(event2, None)
@simulate('Entity_statBalance', '*args, **kwargs', 'return "0"')
@simulate('RegularisationRequest_checkToSendUniqEvent',
'*args, **kwargs',
'raise NotImplementedError, "Should not have been called"')
def test_addRegularisationRequest_balance_ok(self):
person = self.createPerson()
ticket, event = person.Person_checkToCreateRegularisationRequest()
self.assertEquals(ticket, None)
self.assertEquals(event, None)
@simulate('Entity_statBalance', '*args, **kwargs', 'return "1"')
def test_addRegularisationRequest_existing_suspended_ticket(self):
person = self.createPerson()
ticket, event = person.Person_checkToCreateRegularisationRequest()
transaction.commit()
self.tic()
ticket2, event2 = person.Person_checkToCreateRegularisationRequest()
self.assertNotEquals(ticket, None)
self.assertNotEquals(event, None)
self.assertEquals(ticket2.getRelativeUrl(), ticket.getRelativeUrl())
self.assertEquals(event2, None)
@simulate('Entity_statBalance', '*args, **kwargs', 'return "1"')
def test_addRegularisationRequest_existing_validated_ticket(self):
person = self.createPerson()
ticket, event = person.Person_checkToCreateRegularisationRequest()
ticket.validate()
transaction.commit()
self.tic()
ticket2, event2 = person.Person_checkToCreateRegularisationRequest()
self.assertNotEquals(ticket, None)
self.assertNotEquals(event, None)
self.assertEquals(ticket2.getRelativeUrl(), ticket.getRelativeUrl())
self.assertEquals(event2, None)
@simulate('Entity_statBalance', '*args, **kwargs', 'return "1"')
def test_addRegularisationRequest_existing_invalidated_ticket(self):
person = self.createPerson()
ticket, event = person.Person_checkToCreateRegularisationRequest()
ticket.invalidate()
transaction.commit()
self.tic()
ticket2, event2 = person.Person_checkToCreateRegularisationRequest()
self.assertNotEquals(ticket2.getRelativeUrl(), ticket.getRelativeUrl())
self.assertNotEquals(event2, None)
def test_addRegularisationRequest_REQUEST_disallowed(self):
date = DateTime()
person = self.createPerson()
self.assertRaises(
Unauthorized,
person.Person_checkToCreateRegularisationRequest,
REQUEST={})
class TestSlapOSRegularisationRequest_invalidateIfPersonBalanceIsOk(
testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def createPerson(self):
new_id = self.generateNewId()
return self.portal.person_module.newContent(
portal_type='Person',
title="Person %s" % new_id,
reference="TESTPERS-%s" % new_id,
)
def createRegularisationRequest(self):
new_id = self.generateNewId()
return self.portal.regularisation_request_module.newContent(
portal_type='Regularisation Request',
title="Test Reg. Req.%s" % new_id,
reference="TESTREGREQ-%s" % new_id,
)
def test_invalidateIfPersonBalanceIsOk_REQUEST_disallowed(self):
ticket = self.createRegularisationRequest()
self.assertRaises(
Unauthorized,
ticket.RegularisationRequest_invalidateIfPersonBalanceIsOk,
REQUEST={})
@simulate('Entity_statBalance', '*args, **kwargs', 'return "0"')
def test_invalidateIfPersonBalanceIsOk_matching_case(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
ticket.edit(source_project_value=person)
ticket.validate()
ticket.suspend()
ticket.RegularisationRequest_invalidateIfPersonBalanceIsOk()
self.assertEquals(ticket.getSimulationState(), 'invalidated')
@simulate('Entity_statBalance', '*args, **kwargs', 'return "0"')
def test_invalidateIfPersonBalanceIsOk_not_suspended(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
ticket.edit(source_project_value=person)
ticket.validate()
ticket.RegularisationRequest_invalidateIfPersonBalanceIsOk()
self.assertEquals(ticket.getSimulationState(), 'validated')
@simulate('Entity_statBalance', '*args, **kwargs', 'return "0"')
def test_invalidateIfPersonBalanceIsOk_no_person(self):
ticket = self.createRegularisationRequest()
ticket.validate()
ticket.suspend()
ticket.RegularisationRequest_invalidateIfPersonBalanceIsOk()
self.assertEquals(ticket.getSimulationState(), 'suspended')
@simulate('Entity_statBalance', '*args, **kwargs', 'return "1"')
def test_invalidateIfPersonBalanceIsOk_wrong_balance(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
ticket.edit(source_project_value=person)
ticket.validate()
ticket.suspend()
ticket.RegularisationRequest_invalidateIfPersonBalanceIsOk()
self.assertEquals(ticket.getSimulationState(), 'suspended')
class TestSlapOSRegularisationRequest_checkToSendUniqEvent(testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def createPerson(self):
new_id = self.generateNewId()
return self.portal.person_module.newContent(
portal_type='Person',
title="Person %s" % new_id,
reference="TESTPERS-%s" % new_id,
)
def createRegularisationRequest(self):
new_id = self.generateNewId()
return self.portal.regularisation_request_module.newContent(
portal_type='Regularisation Request',
title="Test Reg. Req.%s" % new_id,
reference="TESTREGREQ-%s" % new_id,
resource='foo/bar',
)
def test_checkToSendUniqEvent_no_event(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
ticket.edit(
source='organisation_module/slapos',
destination_value=person,
source_project_value=person)
ticket.validate()
ticket.suspend()
before_date = DateTime()
event = ticket.RegularisationRequest_checkToSendUniqEvent(
'service_module/slapos_crm_spam', 'foo title', 'foo content', 'foo comment')
after_date = DateTime()
self.assertEquals(ticket.getSimulationState(), 'suspended')
self.assertEquals(ticket.getResource(), 'service_module/slapos_crm_spam')
self.assertEquals(event.getPortalType(), 'Mail Message')
self.assertEquals(event.getSimulationState(), 'delivered')
self.assertTrue(event.getStartDate() >= before_date)
self.assertTrue(event.getStopDate() <= after_date)
self.assertEquals(event.getTitle(), "foo title")
self.assertEquals(event.getResource(), 'service_module/slapos_crm_spam')
self.assertEquals(event.getFollowUp(), ticket.getRelativeUrl())
self.assertEquals(event.getSource(), "organisation_module/slapos")
self.assertEquals(event.getDestination(), person.getRelativeUrl())
self.assertEquals(event.getTextContent(), 'foo content')
def test_checkToSendUniqEvent_service_required(self):
ticket = self.createRegularisationRequest()
self.assertRaises(
AssertionError,
ticket.RegularisationRequest_checkToSendUniqEvent,
ticket.getRelativeUrl(), '', '', ''
)
def test_checkToSendUniqEvent_call_twice_with_tic(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
ticket.edit(
source='organisation_module/slapos',
destination_value=person,
source_project_value=person)
ticket.validate()
ticket.suspend()
event = ticket.RegularisationRequest_checkToSendUniqEvent(
'service_module/slapos_crm_spam', 'foo title', 'foo content', 'foo comment')
self.tic()
event2 = ticket.RegularisationRequest_checkToSendUniqEvent(
'service_module/slapos_crm_spam', 'foo2 title', 'foo2 content', 'foo2 comment')
self.assertEquals(event.getTitle(), "foo title")
self.assertEquals(event.getTextContent(), 'foo content')
self.assertEquals(event.getRelativeUrl(), event2.getRelativeUrl())
def test_checkToSendUniqEvent_manual_event(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
ticket.edit(
source='organisation_module/slapos',
destination_value=person,
source_project_value=person)
ticket.validate()
ticket.suspend()
event = self.portal.event_module.newContent(
portal_type="Mail Message",
follow_up=ticket.getRelativeUrl(),
resource='service_module/slapos_crm_spam',
)
self.tic()
event2 = ticket.RegularisationRequest_checkToSendUniqEvent(
'service_module/slapos_crm_spam', 'foo2 title', 'foo2 content', 'foo2 comment')
self.assertEquals(ticket.getResource(), 'foo/bar')
self.assertNotEquals(event.getTitle(), 'foo2 title')
self.assertEquals(event.getTextContent(), None)
self.assertEquals(event.getSimulationState(), 'draft')
self.assertEquals(event.getRelativeUrl(), event2.getRelativeUrl())
def test_checkToSendUniqEvent_not_suspended(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
ticket.edit(
source='organisation_module/slapos',
destination_value=person,
source_project_value=person)
ticket.validate()
event = ticket.RegularisationRequest_checkToSendUniqEvent(
'service_module/slapos_crm_spam', 'foo2 title', 'foo2 content', 'foo2 comment')
self.assertEquals(event, None)
def test_checkToSendUniqEvent_event_not_reindexed(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
ticket.edit(
source='organisation_module/slapos',
destination_value=person,
source_project_value=person)
ticket.validate()
ticket.suspend()
event = ticket.RegularisationRequest_checkToSendUniqEvent(
'service_module/slapos_crm_spam', 'foo title', 'foo content', 'foo comment')
transaction.commit()
event2 = ticket.RegularisationRequest_checkToSendUniqEvent(
'service_module/slapos_crm_spam', 'foo2 title', 'foo2 content', 'foo2 comment')
self.assertNotEquals(event, event2)
self.assertEquals(event2, None)
def test_checkToSendUniqEvent_REQUEST_disallowed(self):
ticket = self.createRegularisationRequest()
self.assertRaises(
Unauthorized,
ticket.RegularisationRequest_checkToSendUniqEvent,
'', '', '', '',
REQUEST={})
class TestSlapOSRegularisationRequest_cancelInvoiceIfPersonOpenOrderIsEmpty(
testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def createPerson(self):
new_id = self.generateNewId()
return self.portal.person_module.newContent(
portal_type='Person',
title="Person %s" % new_id,
reference="TESTPERS-%s" % new_id,
)
def createRegularisationRequest(self):
new_id = self.generateNewId()
return self.portal.regularisation_request_module.newContent(
portal_type='Regularisation Request',
title="Test Reg. Req.%s" % new_id,
reference="TESTREGREQ-%s" % new_id,
resource='foo/bar',
)
def createOpenOrder(self):
new_id = self.generateNewId()
return self.portal.open_sale_order_module.newContent(
portal_type='Open Sale Order',
title="Test Open Order %s" % new_id,
reference="TESTOPENORDER-%s" % new_id,
)
def createSaleInvoiceTransaction(self):
new_id = self.generateNewId()
return self.portal.accounting_module.newContent(
portal_type='Sale Invoice Transaction',
title="Test Sale Invoice %s" % new_id,
reference="TESTSALEINVOICE-%s" % new_id,
)
def createPaymentTransaction(self):
new_id = self.generateNewId()
return self.portal.accounting_module.newContent(
portal_type='Payment Transaction',
title="Test Payment %s" % new_id,
reference="TESTPAYMENT-%s" % new_id,
)
def test_cancelInvoiceIfPersonOpenOrderIsEmpty_REQUEST_disallowed(self):
ticket = self.createRegularisationRequest()
self.assertRaises(
Unauthorized,
ticket.RegularisationRequest_cancelInvoiceIfPersonOpenOrderIsEmpty,
REQUEST={})
@simulate('SaleInvoiceTransaction_createReversalPayzenTransaction',
'*args, **kwargs',
'context.portal_workflow.doActionFor(' \
'context, action="edit_action", ' \
'comment="Visited by SaleInvoiceTransaction_createReversalPayzenTransaction")')
@simulate('RegularisationRequest_checkToSendUniqEvent',
'service_relative_url, title, text_content, comment, REQUEST=None',
'context.portal_workflow.doActionFor(' \
'context, action="edit_action", ' \
'comment="Visited by RegularisationRequest_checkToSendUniqEvent ' \
'%s %s %s %s" % (service_relative_url, title, text_content, comment))\n' \
'return "fooevent"')
def test_cancelInvoiceIfPersonOpenOrderIsEmpty_invoice_to_cancel(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
ticket.edit(
destination_value=person,
source_project_value=person)
ticket.validate()
ticket.suspend()
order = self.createOpenOrder()
order.edit(destination_decision_value=person)
self.portal.portal_workflow._jumpToStateFor(order, 'validated')
invoice = self.createSaleInvoiceTransaction()
invoice.edit(
payment_mode="payzen",
)
payment = self.createPaymentTransaction()
payment.edit(
payment_mode="payzen",
causality_value=invoice,
destination_section_value=person)
self.portal.portal_workflow._jumpToStateFor(payment, 'started')
self.tic()
event, invoice_list = \
ticket.RegularisationRequest_cancelInvoiceIfPersonOpenOrderIsEmpty()
expected_service = 'service_module/slapos_crm_invoice_cancellation'
expected_title = 'Cancellation of your bill'
expected_text= """Hello,
Thank you to have used our decentralized Cloud Computing service slapos.org.
We noticed that all your instances have been removed upon receiving your bill, so we conclude that the instances that you requested were not being used but probably ordered then forgotten.
To not to charge our first users a "non use" of our service, we have choosen to cancel your bill. That's mean: *You have nothing to pay us.*
We hope to see you using our services in the future.
Regards,
The slapos team
"""
expected_comment = 'Cancelled payment.'
self.assertEqual(
'Visited by RegularisationRequest_checkToSendUniqEvent %s %s %s %s' % \
(expected_service, expected_title, expected_text, expected_comment),
ticket.workflow_history['edit_workflow'][-1]['comment'])
self.assertEqual(
'Visited by SaleInvoiceTransaction_createReversalPayzenTransaction',
invoice.workflow_history['edit_workflow'][-1]['comment'])
self.assertEqual(event, "fooevent")
self.assertEqual(invoice_list, [invoice.getRelativeUrl()])
@simulate('SaleInvoiceTransaction_createReversalPayzenTransaction',
'*args, **kwargs',
'raise NotImplementedError, "Should not have been called"')
@simulate('RegularisationRequest_checkToSendUniqEvent',
'*args, **kwargs',
'raise NotImplementedError, "Should not have been called"')
def test_cancelInvoiceIfPersonOpenOrderIsEmpty_not_suspended_ticket(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
ticket.edit(
destination_value=person,
source_project_value=person)
ticket.validate()
event, invoice_list = \
ticket.RegularisationRequest_cancelInvoiceIfPersonOpenOrderIsEmpty()
self.assertEqual(event, None)
self.assertEqual(invoice_list, [])
@simulate('SaleInvoiceTransaction_createReversalPayzenTransaction',
'*args, **kwargs',
'raise NotImplementedError, "Should not have been called"')
@simulate('RegularisationRequest_checkToSendUniqEvent',
'*args, **kwargs',
'raise NotImplementedError, "Should not have been called"')
def test_cancelInvoiceIfPersonOpenOrderIsEmpty_no_person_related(self):
ticket = self.createRegularisationRequest()
ticket.validate()
ticket.suspend()
event, invoice_list = \
ticket.RegularisationRequest_cancelInvoiceIfPersonOpenOrderIsEmpty()
self.assertEqual(event, None)
self.assertEqual(invoice_list, [])
@simulate('SaleInvoiceTransaction_createReversalPayzenTransaction',
'*args, **kwargs',
'raise NotImplementedError, "Should not have been called"')
@simulate('RegularisationRequest_checkToSendUniqEvent',
'*args, **kwargs',
'raise NotImplementedError, "Should not have been called"')
def test_cancelInvoiceIfPersonOpenOrderIsEmpty_no_open_order(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
ticket.edit(
destination_value=person,
source_project_value=person)
ticket.validate()
ticket.suspend()
event, invoice_list = \
ticket.RegularisationRequest_cancelInvoiceIfPersonOpenOrderIsEmpty()
self.assertEqual(event, None)
self.assertEqual(invoice_list, [])
@simulate('SaleInvoiceTransaction_createReversalPayzenTransaction',
'*args, **kwargs',
'raise NotImplementedError, "Should not have been called"')
@simulate('RegularisationRequest_checkToSendUniqEvent',
'*args, **kwargs',
'raise NotImplementedError, "Should not have been called"')
def test_cancelInvoiceIfPersonOpenOrderIsEmpty_with_open_order_line(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
ticket.edit(
destination_value=person,
source_project_value=person)
ticket.validate()
ticket.suspend()
order = self.createOpenOrder()
order.edit(destination_decision_value=person)
order.newContent(portal_type="Open Sale Order Line")
self.portal.portal_workflow._jumpToStateFor(order, 'validated')
self.tic()
event, invoice_list = \
ticket.RegularisationRequest_cancelInvoiceIfPersonOpenOrderIsEmpty()
self.assertEqual(event, None)
self.assertEqual(invoice_list, [])
@simulate('SaleInvoiceTransaction_createReversalPayzenTransaction',
'*args, **kwargs',
'raise NotImplementedError, "Should not have been called"')
@simulate('RegularisationRequest_checkToSendUniqEvent',
'service_relative_url, title, text_content, comment, REQUEST=None',
'context.portal_workflow.doActionFor(' \
'context, action="edit_action", ' \
'comment="Visited by RegularisationRequest_checkToSendUniqEvent ' \
'%s %s %s %s" % (service_relative_url, title, text_content, comment))\n' \
'return "fooevent"')
def test_cancelInvoiceIfPersonOpenOrderIsEmpty_no_invoice(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
ticket.edit(
destination_value=person,
source_project_value=person)
ticket.validate()
ticket.suspend()
order = self.createOpenOrder()
order.edit(destination_decision_value=person)
self.portal.portal_workflow._jumpToStateFor(order, 'validated')
self.tic()
event, invoice_list = \
ticket.RegularisationRequest_cancelInvoiceIfPersonOpenOrderIsEmpty()
self.assertEqual(event, "fooevent")
self.assertEqual(invoice_list, [])
class TestSlapOSRegularisationRequest_checkToTriggerNextEscalationStep(
testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def createRegularisationRequest(self):
new_id = self.generateNewId()
return self.portal.regularisation_request_module.newContent(
portal_type='Regularisation Request',
title="Test Reg. Req.%s" % new_id,
reference="TESTREGREQ-%s" % new_id,
resource='foo/bar',
)
def test_checkToTriggerNextEscalationStep_service_required(self):
ticket = self.createRegularisationRequest()
self.assertRaises(
AssertionError,
ticket.RegularisationRequest_checkToTriggerNextEscalationStep,
0, ticket.getRelativeUrl(), '', '', '', ''
)
@simulate('RegularisationRequest_checkToSendUniqEvent',
'service_relative_url, title, text_content, comment, REQUEST=None',
'context.portal_workflow.doActionFor(' \
'context, action="edit_action", ' \
'comment="Visited by RegularisationRequest_checkToSendUniqEvent ' \
'%s %s %s %s" % (service_relative_url, title, text_content, comment))\n' \
'return "fooevent"')
def test_checkToTriggerNextEscalationStep_matching_event(self):
ticket = self.createRegularisationRequest()
ticket.edit(resource='service_module/slapos_crm_acknowledgement')
ticket.validate()
ticket.suspend()
event = self.portal.event_module.newContent(
portal_type="Mail Message",
follow_up=ticket.getRelativeUrl(),
resource='service_module/slapos_crm_acknowledgement',
start_date=DateTime() - 8,
)
self.portal.portal_workflow._jumpToStateFor(event, 'delivered')
self.tic()
event2 = ticket.RegularisationRequest_checkToTriggerNextEscalationStep(
7, 'service_module/slapos_crm_acknowledgement',
'service_module/slapos_crm_spam',
'foo2 title', 'foo2 content', 'foo2 comment')
self.assertEquals(event2, event.getRelativeUrl())
self.assertEqual(
'Visited by RegularisationRequest_checkToSendUniqEvent %s %s %s %s' % \
('service_module/slapos_crm_spam', 'foo2 title', 'foo2 content',
'foo2 comment'),
ticket.workflow_history['edit_workflow'][-1]['comment'])
@simulate('RegularisationRequest_checkToSendUniqEvent',
'*args, **kwargs',
'raise NotImplementedError, "Should not have been called"')
def test_checkToTriggerNextEscalationStep_recent_event(self):
ticket = self.createRegularisationRequest()
ticket.edit(resource='service_module/slapos_crm_acknowledgement')
ticket.validate()
ticket.suspend()
event = self.portal.event_module.newContent(
portal_type="Mail Message",
follow_up=ticket.getRelativeUrl(),
resource='service_module/slapos_crm_acknowledgement',
start_date=DateTime() - 6,
)
self.portal.portal_workflow._jumpToStateFor(event, 'delivered')
self.tic()
event2 = ticket.RegularisationRequest_checkToTriggerNextEscalationStep(
7, 'service_module/slapos_crm_acknowledgement',
'service_module/slapos_crm_spam',
'foo2 title', 'foo2 content', 'foo2 comment')
self.assertEquals(event2, None)
@simulate('RegularisationRequest_checkToSendUniqEvent',
'*args, **kwargs',
'raise NotImplementedError, "Should not have been called"')
def test_checkToTriggerNextEscalationStep_other_ticket_event(self):
ticket = self.createRegularisationRequest()
ticket.edit(resource='service_module/slapos_crm_acknowledgement')
ticket.validate()
ticket.suspend()
event = self.portal.event_module.newContent(
portal_type="Mail Message",
resource='service_module/slapos_crm_acknowledgement',
start_date=DateTime() - 2,
)
self.portal.portal_workflow._jumpToStateFor(event, 'delivered')
self.tic()
event2 = ticket.RegularisationRequest_checkToTriggerNextEscalationStep(
1, 'service_module/slapos_crm_acknowledgement',
'service_module/slapos_crm_spam',
'foo2 title', 'foo2 content', 'foo2 comment')
self.assertEquals(event2, None)
@simulate('RegularisationRequest_checkToSendUniqEvent',
'*args, **kwargs',
'raise NotImplementedError, "Should not have been called"')
def test_checkToTriggerNextEscalationStep_other_resource_event(self):
ticket = self.createRegularisationRequest()
ticket.edit(resource='service_module/slapos_crm_acknowledgement')
ticket.validate()
ticket.suspend()
event = self.portal.event_module.newContent(
portal_type="Mail Message",
follow_up=ticket.getRelativeUrl(),
resource='service_module/slapos_crm_spam',
start_date=DateTime() - 2,
)
self.portal.portal_workflow._jumpToStateFor(event, 'delivered')
self.tic()
event2 = ticket.RegularisationRequest_checkToTriggerNextEscalationStep(
1, 'service_module/slapos_crm_acknowledgement',
'service_module/slapos_crm_spam',
'foo2 title', 'foo2 content', 'foo2 comment')
self.assertEquals(event2, None)
@simulate('RegularisationRequest_checkToSendUniqEvent',
'*args, **kwargs',
'raise NotImplementedError, "Should not have been called"')
def test_checkToTriggerNextEscalationStep_no_current_event(self):
ticket = self.createRegularisationRequest()
ticket.edit(resource='service_module/slapos_crm_acknowledgement')
ticket.validate()
ticket.suspend()
self.tic()
event2 = ticket.RegularisationRequest_checkToTriggerNextEscalationStep(
1, 'service_module/slapos_crm_acknowledgement',
'service_module/slapos_crm_spam',
'foo2 title', 'foo2 content', 'foo2 comment')
self.assertEquals(event2, None)
@simulate('RegularisationRequest_checkToSendUniqEvent',
'*args, **kwargs',
'raise NotImplementedError, "Should not have been called"')
def test_checkToTriggerNextEscalationStep_no_ticket_resource(self):
ticket = self.createRegularisationRequest()
ticket.validate()
ticket.suspend()
event = self.portal.event_module.newContent(
portal_type="Mail Message",
follow_up=ticket.getRelativeUrl(),
resource='service_module/slapos_crm_acknowledgement',
start_date=DateTime() - 2,
)
self.portal.portal_workflow._jumpToStateFor(event, 'delivered')
self.tic()
event2 = ticket.RegularisationRequest_checkToTriggerNextEscalationStep(
1, 'service_module/slapos_crm_acknowledgement',
'service_module/slapos_crm_spam',
'foo2 title', 'foo2 content', 'foo2 comment')
self.assertEquals(event2, None)
@simulate('RegularisationRequest_checkToSendUniqEvent',
'*args, **kwargs',
'raise NotImplementedError, "Should not have been called"')
def test_checkToTriggerNextEscalationStep_not_suspended(self):
ticket = self.createRegularisationRequest()
ticket.edit(resource='service_module/slapos_crm_acknowledgement')
ticket.validate()
event = self.portal.event_module.newContent(
portal_type="Mail Message",
follow_up=ticket.getRelativeUrl(),
resource='service_module/slapos_crm_acknowledgement',
start_date=DateTime() - 2,
)
self.portal.portal_workflow._jumpToStateFor(event, 'delivered')
self.tic()
event2 = ticket.RegularisationRequest_checkToTriggerNextEscalationStep(
1, 'service_module/slapos_crm_acknowledgement',
'service_module/slapos_crm_spam',
'foo2 title', 'foo2 content', 'foo2 comment')
self.assertEquals(event2, None)
def test_checkToTriggerNextEscalationStep_REQUEST_disallowed(self):
ticket = self.createRegularisationRequest()
self.assertRaises(
Unauthorized,
ticket.RegularisationRequest_checkToTriggerNextEscalationStep,
'', '', '', '', '', '',
REQUEST={})
class TestSlapOSRegularisationRequest_triggerAcknowledgmentEscalation(
testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def createRegularisationRequest(self):
new_id = self.generateNewId()
return self.portal.regularisation_request_module.newContent(
portal_type='Regularisation Request',
title="Test Reg. Req.%s" % new_id,
reference="TESTREGREQ-%s" % new_id,
resource='foo/bar',
)
def test_triggerAcknowledgmentEscalation_REQUEST_disallowed(self):
ticket = self.createRegularisationRequest()
self.assertRaises(
Unauthorized,
ticket.RegularisationRequest_triggerAcknowledgmentEscalation,
REQUEST={})
@simulate('RegularisationRequest_checkToTriggerNextEscalationStep',
'day, current, next, title, text_content, comment, REQUEST=None',
'context.portal_workflow.doActionFor(' \
'context, action="edit_action", ' \
'comment="Visited by RegularisationRequest_checkToTriggerNextEscalationStep ' \
'%s %s %s %s %s %s" % (day, current, next, title, text_content, comment))')
def test_checkToTriggerNextEscalationStep_matching_event(self):
ticket = self.createRegularisationRequest()
ticket.RegularisationRequest_triggerAcknowledgmentEscalation()
self.assertEqual(
'Visited by RegularisationRequest_checkToTriggerNextEscalationStep ' \
'%s %s %s %s %s %s' % \
(38,
'service_module/slapos_crm_acknowledgement',
'service_module/slapos_crm_stop_reminder',
'Reminder: invoice payment requested',
"""Dear user,
We would like to remind you the unpaid invoice you have on %s.
If no payment is done during the coming days, we will stop all your current instances to free some hardware resources.
Do not hesitate to visit the web forum (http://community.slapos.org/forum) in case of question.
Regards,
The slapos team
""" % self.portal.portal_preferences.getPreferredSlaposWebSiteUrl(),
'Stopping reminder.'),
ticket.workflow_history['edit_workflow'][-1]['comment'])
@simulate('NotificationTool_getDocumentValue',
'reference=None',
'assert reference == "slapos-crm.acknowledgment.escalation"\n' \
'return context.restrictedTraverse(' \
'context.REQUEST["test_checkToTriggerNextEscalationStep_notification_message"])')
@simulate('RegularisationRequest_checkToTriggerNextEscalationStep',
'day, current, next, title, text_content, comment, REQUEST=None',
'context.portal_workflow.doActionFor(' \
'context, action="edit_action", ' \
'comment="Visited by RegularisationRequest_checkToTriggerNextEscalationStep ' \
'%s %s %s %s %s %s" % (day, current, next, title, text_content, comment))')
def test_checkToTriggerNextEscalationStep_notification_message(self):
ticket = self.createRegularisationRequest()
new_id = self.generateNewId()
notification_message = self.portal.notification_message_module.newContent(
portal_type="Notification Message",
title='Test NM title %s' % new_id,
text_content='Test NM content<br/>%s<br/>' % new_id,
content_type='text/html',
)
self.portal.REQUEST\
['test_checkToTriggerNextEscalationStep_notification_message'] = \
notification_message.getRelativeUrl()
ticket.RegularisationRequest_triggerAcknowledgmentEscalation()
self.assertEqual(
'Visited by RegularisationRequest_checkToTriggerNextEscalationStep ' \
'%s %s %s %s %s %s' % \
(38,
'service_module/slapos_crm_acknowledgement',
'service_module/slapos_crm_stop_reminder',
'Test NM title %s' % new_id,
'Test NM content\n%s\n' % new_id,
'Stopping reminder.'),
ticket.workflow_history['edit_workflow'][-1]['comment'])
class TestSlapOSRegularisationRequest_triggerStopReminderEscalation(
testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def createRegularisationRequest(self):
new_id = self.generateNewId()
return self.portal.regularisation_request_module.newContent(
portal_type='Regularisation Request',
title="Test Reg. Req.%s" % new_id,
reference="TESTREGREQ-%s" % new_id,
resource='foo/bar',
)
def test_triggerStopReminderEscalation_REQUEST_disallowed(self):
ticket = self.createRegularisationRequest()
self.assertRaises(
Unauthorized,
ticket.RegularisationRequest_triggerStopReminderEscalation,
REQUEST={})
@simulate('RegularisationRequest_checkToTriggerNextEscalationStep',
'day, current, next, title, text_content, comment, REQUEST=None',
'context.portal_workflow.doActionFor(' \
'context, action="edit_action", ' \
'comment="Visited by RegularisationRequest_checkToTriggerNextEscalationStep ' \
'%s %s %s %s %s %s" % (day, current, next, title, text_content, comment))')
def test_checkToTriggerNextEscalationStep_matching_event(self):
ticket = self.createRegularisationRequest()
ticket.RegularisationRequest_triggerStopReminderEscalation()
self.assertEqual(
'Visited by RegularisationRequest_checkToTriggerNextEscalationStep ' \
'%s %s %s %s %s %s' % \
(7,
'service_module/slapos_crm_stop_reminder',
'service_module/slapos_crm_stop_acknowledgement',
'Acknowledgment: instances stopped',
"""Dear user,
Despite our last reminder, you still have an unpaid invoice on %s.
We will now stop all your current instances to free some hardware resources.
Do not hesitate to visit the web forum (http://community.slapos.org/forum) in case of question.
Regards,
The slapos team
""" % self.portal.portal_preferences.getPreferredSlaposWebSiteUrl(),
'Stopping acknowledgment.'),
ticket.workflow_history['edit_workflow'][-1]['comment'])
@simulate('NotificationTool_getDocumentValue',
'reference=None',
'assert reference == "slapos-crm.stop.reminder.escalation"\n' \
'return context.restrictedTraverse(' \
'context.REQUEST["test_checkToTriggerNextEscalationStep_notification_message"])')
@simulate('RegularisationRequest_checkToTriggerNextEscalationStep',
'day, current, next, title, text_content, comment, REQUEST=None',
'context.portal_workflow.doActionFor(' \
'context, action="edit_action", ' \
'comment="Visited by RegularisationRequest_checkToTriggerNextEscalationStep ' \
'%s %s %s %s %s %s" % (day, current, next, title, text_content, comment))')
def test_checkToTriggerNextEscalationStep_notification_message(self):
ticket = self.createRegularisationRequest()
new_id = self.generateNewId()
notification_message = self.portal.notification_message_module.newContent(
portal_type="Notification Message",
title='Test NM title %s' % new_id,
text_content='Test NM content<br/>%s<br/>' % new_id,
content_type='text/html',
)
self.portal.REQUEST\
['test_checkToTriggerNextEscalationStep_notification_message'] = \
notification_message.getRelativeUrl()
ticket.RegularisationRequest_triggerStopReminderEscalation()
self.assertEqual(
'Visited by RegularisationRequest_checkToTriggerNextEscalationStep ' \
'%s %s %s %s %s %s' % \
(7,
'service_module/slapos_crm_stop_reminder',
'service_module/slapos_crm_stop_acknowledgement',
'Test NM title %s' % new_id,
'Test NM content\n%s\n' % new_id,
'Stopping acknowledgment.'),
ticket.workflow_history['edit_workflow'][-1]['comment'])
class TestSlapOSRegularisationRequest_triggerStopAcknowledgmentEscalation(
testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def createRegularisationRequest(self):
new_id = self.generateNewId()
return self.portal.regularisation_request_module.newContent(
portal_type='Regularisation Request',
title="Test Reg. Req.%s" % new_id,
reference="TESTREGREQ-%s" % new_id,
resource='foo/bar',
)
def test_triggerStopAcknowledgmentEscalation_REQUEST_disallowed(self):
ticket = self.createRegularisationRequest()
self.assertRaises(
Unauthorized,
ticket.RegularisationRequest_triggerStopAcknowledgmentEscalation,
REQUEST={})
@simulate('RegularisationRequest_checkToTriggerNextEscalationStep',
'day, current, next, title, text_content, comment, REQUEST=None',
'context.portal_workflow.doActionFor(' \
'context, action="edit_action", ' \
'comment="Visited by RegularisationRequest_checkToTriggerNextEscalationStep ' \
'%s %s %s %s %s %s" % (day, current, next, title, text_content, comment))')
def test_checkToTriggerNextEscalationStep_matching_event(self):
ticket = self.createRegularisationRequest()
ticket.RegularisationRequest_triggerStopAcknowledgmentEscalation()
self.assertEqual(
'Visited by RegularisationRequest_checkToTriggerNextEscalationStep ' \
'%s %s %s %s %s %s' % \
(13,
'service_module/slapos_crm_stop_acknowledgement',
'service_module/slapos_crm_delete_reminder',
'Last reminder: invoice payment requested',
"""Dear user,
We would like to remind you the unpaid invoice you have on %s.
If no payment is done during the coming days, we will delete all your instances.
Do not hesitate to visit the web forum (http://community.slapos.org/forum) in case of question.
Regards,
The slapos team
""" % self.portal.portal_preferences.getPreferredSlaposWebSiteUrl(),
'Deleting reminder.'),
ticket.workflow_history['edit_workflow'][-1]['comment'])
@simulate('NotificationTool_getDocumentValue',
'reference=None',
'assert reference == "slapos-crm.stop.acknowledgment.escalation"\n' \
'return context.restrictedTraverse(' \
'context.REQUEST["test_checkToTriggerNextEscalationStep_notification_message"])')
@simulate('RegularisationRequest_checkToTriggerNextEscalationStep',
'day, current, next, title, text_content, comment, REQUEST=None',
'context.portal_workflow.doActionFor(' \
'context, action="edit_action", ' \
'comment="Visited by RegularisationRequest_checkToTriggerNextEscalationStep ' \
'%s %s %s %s %s %s" % (day, current, next, title, text_content, comment))')
def test_checkToTriggerNextEscalationStep_notification_message(self):
ticket = self.createRegularisationRequest()
new_id = self.generateNewId()
notification_message = self.portal.notification_message_module.newContent(
portal_type="Notification Message",
title='Test NM title %s' % new_id,
text_content='Test NM content<br/>%s<br/>' % new_id,
content_type='text/html',
)
self.portal.REQUEST\
['test_checkToTriggerNextEscalationStep_notification_message'] = \
notification_message.getRelativeUrl()
ticket.RegularisationRequest_triggerStopAcknowledgmentEscalation()
self.assertEqual(
'Visited by RegularisationRequest_checkToTriggerNextEscalationStep ' \
'%s %s %s %s %s %s' % \
(13,
'service_module/slapos_crm_stop_acknowledgement',
'service_module/slapos_crm_delete_reminder',
'Test NM title %s' % new_id,
'Test NM content\n%s\n' % new_id,
'Deleting reminder.'),
ticket.workflow_history['edit_workflow'][-1]['comment'])
class TestSlapOSRegularisationRequest_triggerDeleteReminderEscalation(
testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def createRegularisationRequest(self):
new_id = self.generateNewId()
return self.portal.regularisation_request_module.newContent(
portal_type='Regularisation Request',
title="Test Reg. Req.%s" % new_id,
reference="TESTREGREQ-%s" % new_id,
resource='foo/bar',
)
def test_triggerDeleteReminderEscalation_REQUEST_disallowed(self):
ticket = self.createRegularisationRequest()
self.assertRaises(
Unauthorized,
ticket.RegularisationRequest_triggerDeleteReminderEscalation,
REQUEST={})
@simulate('RegularisationRequest_checkToTriggerNextEscalationStep',
'day, current, next, title, text_content, comment, REQUEST=None',
'context.portal_workflow.doActionFor(' \
'context, action="edit_action", ' \
'comment="Visited by RegularisationRequest_checkToTriggerNextEscalationStep ' \
'%s %s %s %s %s %s" % (day, current, next, title, text_content, comment))')
def test_checkToTriggerNextEscalationStep_matching_event(self):
ticket = self.createRegularisationRequest()
ticket.RegularisationRequest_triggerDeleteReminderEscalation()
self.assertEqual(
'Visited by RegularisationRequest_checkToTriggerNextEscalationStep ' \
'%s %s %s %s %s %s' % \
(2,
'service_module/slapos_crm_delete_reminder',
'service_module/slapos_crm_delete_acknowledgement',
'Acknowledgment: instances deleted',
"""Dear user,
Despite our last reminder, you still have an unpaid invoice on %s.
We will now delete all your instances.
Do not hesitate to visit the web forum (http://community.slapos.org/forum) in case of question.
Regards,
The slapos team
""" % self.portal.portal_preferences.getPreferredSlaposWebSiteUrl(),
'Deleting acknowledgment.'),
ticket.workflow_history['edit_workflow'][-1]['comment'])
@simulate('NotificationTool_getDocumentValue',
'reference=None',
'assert reference == "slapos-crm.delete.reminder.escalation"\n' \
'return context.restrictedTraverse(' \
'context.REQUEST["test_checkToTriggerNextEscalationStep_notification_message"])')
@simulate('RegularisationRequest_checkToTriggerNextEscalationStep',
'day, current, next, title, text_content, comment, REQUEST=None',
'context.portal_workflow.doActionFor(' \
'context, action="edit_action", ' \
'comment="Visited by RegularisationRequest_checkToTriggerNextEscalationStep ' \
'%s %s %s %s %s %s" % (day, current, next, title, text_content, comment))')
def test_checkToTriggerNextEscalationStep_notification_message(self):
ticket = self.createRegularisationRequest()
new_id = self.generateNewId()
notification_message = self.portal.notification_message_module.newContent(
portal_type="Notification Message",
title='Test NM title %s' % new_id,
text_content='Test NM content<br/>%s<br/>' % new_id,
content_type='text/html',
)
self.portal.REQUEST\
['test_checkToTriggerNextEscalationStep_notification_message'] = \
notification_message.getRelativeUrl()
ticket.RegularisationRequest_triggerDeleteReminderEscalation()
self.assertEqual(
'Visited by RegularisationRequest_checkToTriggerNextEscalationStep ' \
'%s %s %s %s %s %s' % \
(2,
'service_module/slapos_crm_delete_reminder',
'service_module/slapos_crm_delete_acknowledgement',
'Test NM title %s' % new_id,
'Test NM content\n%s\n' % new_id,
'Deleting acknowledgment.'),
ticket.workflow_history['edit_workflow'][-1]['comment'])
class TestSlapOSRegularisationRequest_stopHostingSubscriptionList(
testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def createRegularisationRequest(self):
new_id = self.generateNewId()
return self.portal.regularisation_request_module.newContent(
portal_type='Regularisation Request',
title="Test Reg. Req.%s" % new_id,
reference="TESTREGREQ-%s" % new_id,
resource='foo/bar',
)
def createPerson(self):
new_id = self.generateNewId()
person = self.portal.person_module.template_member.\
Base_createCloneDocument(batch_mode=1)
person.edit(
title="Person %s" % new_id,
reference="TESTPERS-%s" % new_id,
default_email_text="live_test_%s@example.org" % new_id,
)
person.manage_delObjects(
[x.getId() for x in person.contentValues(portal_type="Assignment")]
)
return person
def createHostingSubscription(self):
new_id = self.generateNewId()
hosting_subscription = self.portal.hosting_subscription_module\
.template_hosting_subscription.Base_createCloneDocument(batch_mode=1)
hosting_subscription.edit(
reference="TESTHS-%s" % new_id,
)
hosting_subscription.validate()
self.portal.portal_workflow._jumpToStateFor(
hosting_subscription, 'start_requested')
return hosting_subscription
def test_stopHostingSubscriptionList_REQUEST_disallowed(self):
ticket = self.createRegularisationRequest()
self.assertRaises(
Unauthorized,
ticket.RegularisationRequest_stopHostingSubscriptionList,
'footag',
REQUEST={})
@simulate('HostingSubscription_stopFromRegularisationRequest',
'person, REQUEST=None',
'context.portal_workflow.doActionFor(' \
'context, action="edit_action", ' \
'comment="Visited by HostingSubscription_stopFromRegularisationRequest ' \
'%s" % (person))')
def test_stopHostingSubscriptionList_matching_subscription(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
hosting_subscription = self.createHostingSubscription()
ticket.edit(
source_project_value=person,
resource='service_module/slapos_crm_stop_acknowledgement',
)
ticket.validate()
ticket.suspend()
hosting_subscription.edit(
destination_section=person.getRelativeUrl(),
)
self.tic()
result = ticket.\
RegularisationRequest_stopHostingSubscriptionList('footag')
self.assertTrue(result)
self.tic()
self.assertEqual(
'Visited by HostingSubscription_stopFromRegularisationRequest ' \
'%s' % person.getRelativeUrl(),
hosting_subscription.workflow_history['edit_workflow'][-1]['comment'])
@simulate('HostingSubscription_stopFromRegularisationRequest',
'person, REQUEST=None',
'context.portal_workflow.doActionFor(' \
'context, action="edit_action", ' \
'comment="Visited by HostingSubscription_stopFromRegularisationRequest ' \
'%s" % (person))')
def test_stopHostingSubscriptionList_matching_subscription_2(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
hosting_subscription = self.createHostingSubscription()
ticket.edit(
source_project_value=person,
resource='service_module/slapos_crm_delete_reminder',
)
ticket.validate()
ticket.suspend()
hosting_subscription.edit(
destination_section=person.getRelativeUrl(),
)
self.tic()
result = ticket.\
RegularisationRequest_stopHostingSubscriptionList('footag')
self.assertTrue(result)
self.tic()
self.assertEqual(
'Visited by HostingSubscription_stopFromRegularisationRequest ' \
'%s' % person.getRelativeUrl(),
hosting_subscription.workflow_history['edit_workflow'][-1]['comment'])
@simulate('HostingSubscription_stopFromRegularisationRequest',
'*args, **kwargs',
'raise NotImplementedError, "Should not have been called"')
def test_stopHostingSubscriptionList_other_subscription(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
hosting_subscription = self.createHostingSubscription()
ticket.edit(
source_project_value=person,
resource='service_module/slapos_crm_stop_acknowledgement',
)
ticket.validate()
ticket.suspend()
self.tic()
result = ticket.\
RegularisationRequest_stopHostingSubscriptionList('footag')
self.assertTrue(result)
self.tic()
@simulate('HostingSubscription_stopFromRegularisationRequest',
'*args, **kwargs',
'raise NotImplementedError, "Should not have been called"')
def test_stopHostingSubscriptionList_no_person(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
ticket.edit(
resource='service_module/slapos_crm_stop_acknowledgement',
)
ticket.validate()
ticket.suspend()
self.tic()
result = ticket.\
RegularisationRequest_stopHostingSubscriptionList('footag')
self.assertFalse(result)
self.tic()
@simulate('HostingSubscription_stopFromRegularisationRequest',
'*args, **kwargs',
'raise NotImplementedError, "Should not have been called"')
def test_stopHostingSubscriptionList_not_suspended(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
hosting_subscription = self.createHostingSubscription()
ticket.edit(
source_project_value=person,
resource='service_module/slapos_crm_stop_acknowledgement',
)
ticket.validate()
self.tic()
result = ticket.\
RegularisationRequest_stopHostingSubscriptionList('footag')
self.assertFalse(result)
self.tic()
@simulate('HostingSubscription_stopFromRegularisationRequest',
'*args, **kwargs',
'raise NotImplementedError, "Should not have been called"')
def test_stopHostingSubscriptionList_other_resource(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
hosting_subscription = self.createHostingSubscription()
ticket.edit(
source_project_value=person,
resource='service_module/slapos_crm_acknowledgement',
)
ticket.validate()
ticket.suspend()
self.tic()
result = ticket.\
RegularisationRequest_stopHostingSubscriptionList('footag')
self.assertFalse(result)
self.tic()
class TestSlapOSHostingSubscription_stopFromRegularisationRequest(
testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def createPerson(self):
new_id = self.generateNewId()
person = self.portal.person_module.template_member.\
Base_createCloneDocument(batch_mode=1)
person.edit(
title="Person %s" % new_id,
reference="TESTPERS-%s" % new_id,
default_email_text="live_test_%s@example.org" % new_id,
)
person.manage_delObjects(
[x.getId() for x in person.contentValues(portal_type="Assignment")]
)
return person
def createHostingSubscription(self):
new_id = self.generateNewId()
hosting_subscription = self.portal.hosting_subscription_module\
.template_hosting_subscription.Base_createCloneDocument(batch_mode=1)
hosting_subscription.edit(
reference="TESTHS-%s" % new_id,
)
hosting_subscription.validate()
self.portal.portal_workflow._jumpToStateFor(
hosting_subscription, 'start_requested')
return hosting_subscription
def test_stopFromRegularisationRequest_REQUEST_disallowed(self):
self.assertRaises(
Unauthorized,
self.portal.HostingSubscription_stopFromRegularisationRequest,
'',
REQUEST={})
def test_stopFromRegularisationRequest_matching_subscription(self):
person = self.createPerson()
hosting_subscription = self.createHostingSubscription()
hosting_subscription.edit(
destination_section=person.getRelativeUrl(),
)
self.tic()
software_release = hosting_subscription.getUrlString()
software_title = hosting_subscription.getTitle()
software_type = hosting_subscription.getSourceReference()
instance_xml = hosting_subscription.getTextContent()
sla_xml = hosting_subscription.getSlaXml()
shared = hosting_subscription.isRootSlave()
self.assertEquals(hosting_subscription.getSlapState(), "start_requested")
result = hosting_subscription.\
HostingSubscription_stopFromRegularisationRequest(person.getRelativeUrl())
self.assertEquals(result, True)
self.assertEquals(hosting_subscription.getUrlString(), software_release)
self.assertEquals(hosting_subscription.getTitle(), software_title)
self.assertEquals(hosting_subscription.getSourceReference(), software_type)
self.assertEquals(hosting_subscription.getTextContent(), instance_xml)
self.assertEquals(hosting_subscription.getSlaXml(), sla_xml)
self.assertEquals(hosting_subscription.isRootSlave(), shared)
self.assertEquals(hosting_subscription.getSlapState(), "stop_requested")
def test_stopFromRegularisationRequest_stopped_subscription(self):
person = self.createPerson()
hosting_subscription = self.createHostingSubscription()
hosting_subscription.edit(
destination_section=person.getRelativeUrl(),
)
self.portal.portal_workflow._jumpToStateFor(
hosting_subscription, 'stop_requested')
result = hosting_subscription.\
HostingSubscription_stopFromRegularisationRequest(person.getRelativeUrl())
self.assertEquals(result, False)
def test_stopFromRegularisationRequest_non_matching_person(self):
hosting_subscription = self.createHostingSubscription()
self.assertRaises(
AssertionError,
hosting_subscription.HostingSubscription_stopFromRegularisationRequest,
'foobar')
class TestSlapOSHostingSubscription_deleteFromRegularisationRequest(
testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def createPerson(self):
new_id = self.generateNewId()
person = self.portal.person_module.template_member.\
Base_createCloneDocument(batch_mode=1)
person.edit(
title="Person %s" % new_id,
reference="TESTPERS-%s" % new_id,
default_email_text="live_test_%s@example.org" % new_id,
)
person.manage_delObjects(
[x.getId() for x in person.contentValues(portal_type="Assignment")]
)
return person
def createHostingSubscription(self):
new_id = self.generateNewId()
hosting_subscription = self.portal.hosting_subscription_module\
.template_hosting_subscription.Base_createCloneDocument(batch_mode=1)
hosting_subscription.edit(
reference="TESTHS-%s" % new_id,
)
hosting_subscription.validate()
self.portal.portal_workflow._jumpToStateFor(
hosting_subscription, 'start_requested')
return hosting_subscription
def test_deleteFromRegularisationRequest_REQUEST_disallowed(self):
self.assertRaises(
Unauthorized,
self.portal.HostingSubscription_deleteFromRegularisationRequest,
'',
REQUEST={})
def test_deleteFromRegularisationRequest_started_subscription(self):
person = self.createPerson()
hosting_subscription = self.createHostingSubscription()
hosting_subscription.edit(
destination_section=person.getRelativeUrl(),
)
self.tic()
software_release = hosting_subscription.getUrlString()
software_title = hosting_subscription.getTitle()
software_type = hosting_subscription.getSourceReference()
instance_xml = hosting_subscription.getTextContent()
sla_xml = hosting_subscription.getSlaXml()
shared = hosting_subscription.isRootSlave()
self.assertEquals(hosting_subscription.getSlapState(), "start_requested")
result = hosting_subscription.\
HostingSubscription_deleteFromRegularisationRequest(person.getRelativeUrl())
self.assertEquals(result, True)
self.assertEquals(hosting_subscription.getUrlString(), software_release)
self.assertEquals(hosting_subscription.getTitle(), software_title)
self.assertEquals(hosting_subscription.getSourceReference(), software_type)
self.assertEquals(hosting_subscription.getTextContent(), instance_xml)
self.assertEquals(hosting_subscription.getSlaXml(), sla_xml)
self.assertEquals(hosting_subscription.isRootSlave(), shared)
self.assertEquals(hosting_subscription.getSlapState(), "destroy_requested")
def test_deleteFromRegularisationRequest_stopped_subscription(self):
person = self.createPerson()
hosting_subscription = self.createHostingSubscription()
hosting_subscription.edit(
destination_section=person.getRelativeUrl(),
)
self.portal.portal_workflow._jumpToStateFor(
hosting_subscription, 'stop_requested')
self.tic()
software_release = hosting_subscription.getUrlString()
software_title = hosting_subscription.getTitle()
software_type = hosting_subscription.getSourceReference()
instance_xml = hosting_subscription.getTextContent()
sla_xml = hosting_subscription.getSlaXml()
shared = hosting_subscription.isRootSlave()
self.assertEquals(hosting_subscription.getSlapState(), "stop_requested")
result = hosting_subscription.\
HostingSubscription_deleteFromRegularisationRequest(person.getRelativeUrl())
self.assertEquals(result, True)
self.assertEquals(hosting_subscription.getUrlString(), software_release)
self.assertEquals(hosting_subscription.getTitle(), software_title)
self.assertEquals(hosting_subscription.getSourceReference(), software_type)
self.assertEquals(hosting_subscription.getTextContent(), instance_xml)
self.assertEquals(hosting_subscription.getSlaXml(), sla_xml)
self.assertEquals(hosting_subscription.isRootSlave(), shared)
self.assertEquals(hosting_subscription.getSlapState(), "destroy_requested")
def test_deleteFromRegularisationRequest_destroyed_subscription(self):
person = self.createPerson()
hosting_subscription = self.createHostingSubscription()
hosting_subscription.edit(
destination_section=person.getRelativeUrl(),
)
self.portal.portal_workflow._jumpToStateFor(
hosting_subscription, 'destroy_requested')
result = hosting_subscription.\
HostingSubscription_deleteFromRegularisationRequest(person.getRelativeUrl())
self.assertEquals(result, False)
def test_deleteFromRegularisationRequest_non_matching_person(self):
hosting_subscription = self.createHostingSubscription()
self.assertRaises(
AssertionError,
hosting_subscription.HostingSubscription_deleteFromRegularisationRequest,
'foobar')
class TestSlapOSRegularisationRequest_deleteHostingSubscriptionList(
testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def createRegularisationRequest(self):
new_id = self.generateNewId()
return self.portal.regularisation_request_module.newContent(
portal_type='Regularisation Request',
title="Test Reg. Req.%s" % new_id,
reference="TESTREGREQ-%s" % new_id,
resource='foo/bar',
)
def createPerson(self):
new_id = self.generateNewId()
person = self.portal.person_module.template_member.\
Base_createCloneDocument(batch_mode=1)
person.edit(
title="Person %s" % new_id,
reference="TESTPERS-%s" % new_id,
default_email_text="live_test_%s@example.org" % new_id,
)
person.manage_delObjects(
[x.getId() for x in person.contentValues(portal_type="Assignment")]
)
return person
def createHostingSubscription(self):
new_id = self.generateNewId()
hosting_subscription = self.portal.hosting_subscription_module\
.template_hosting_subscription.Base_createCloneDocument(batch_mode=1)
hosting_subscription.edit(
reference="TESTHS-%s" % new_id,
)
hosting_subscription.validate()
self.portal.portal_workflow._jumpToStateFor(
hosting_subscription, 'start_requested')
return hosting_subscription
def test_deleteHostingSubscriptionList_REQUEST_disallowed(self):
ticket = self.createRegularisationRequest()
self.assertRaises(
Unauthorized,
ticket.RegularisationRequest_deleteHostingSubscriptionList,
'footag',
REQUEST={})
@simulate('HostingSubscription_deleteFromRegularisationRequest',
'person, REQUEST=None',
'context.portal_workflow.doActionFor(' \
'context, action="edit_action", ' \
'comment="Visited by HostingSubscription_deleteFromRegularisationRequest ' \
'%s" % (person))')
def test_deleteHostingSubscriptionList_matching_subscription(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
hosting_subscription = self.createHostingSubscription()
ticket.edit(
source_project_value=person,
resource='service_module/slapos_crm_delete_acknowledgement',
)
ticket.validate()
ticket.suspend()
hosting_subscription.edit(
destination_section=person.getRelativeUrl(),
)
self.tic()
result = ticket.\
RegularisationRequest_deleteHostingSubscriptionList('footag')
self.assertTrue(result)
self.tic()
self.assertEqual(
'Visited by HostingSubscription_deleteFromRegularisationRequest ' \
'%s' % person.getRelativeUrl(),
hosting_subscription.workflow_history['edit_workflow'][-1]['comment'])
@simulate('HostingSubscription_deleteFromRegularisationRequest',
'*args, **kwargs',
'raise NotImplementedError, "Should not have been called"')
def test_deleteHostingSubscriptionList_other_subscription(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
hosting_subscription = self.createHostingSubscription()
ticket.edit(
source_project_value=person,
resource='service_module/slapos_crm_delete_acknowledgement',
)
ticket.validate()
ticket.suspend()
self.tic()
result = ticket.\
RegularisationRequest_deleteHostingSubscriptionList('footag')
self.assertTrue(result)
self.tic()
@simulate('HostingSubscription_deleteFromRegularisationRequest',
'*args, **kwargs',
'raise NotImplementedError, "Should not have been called"')
def test_deleteHostingSubscriptionList_no_person(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
ticket.edit(
resource='service_module/slapos_crm_delete_acknowledgement',
)
ticket.validate()
ticket.suspend()
self.tic()
result = ticket.\
RegularisationRequest_deleteHostingSubscriptionList('footag')
self.assertFalse(result)
self.tic()
@simulate('HostingSubscription_deleteFromRegularisationRequest',
'*args, **kwargs',
'raise NotImplementedError, "Should not have been called"')
def test_deleteHostingSubscriptionList_not_suspended(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
hosting_subscription = self.createHostingSubscription()
ticket.edit(
source_project_value=person,
resource='service_module/slapos_crm_delete_acknowledgement',
)
ticket.validate()
self.tic()
result = ticket.\
RegularisationRequest_deleteHostingSubscriptionList('footag')
self.assertFalse(result)
self.tic()
@simulate('HostingSubscription_deleteFromRegularisationRequest',
'*args, **kwargs',
'raise NotImplementedError, "Should not have been called"')
def test_deleteHostingSubscriptionList_other_resource(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
hosting_subscription = self.createHostingSubscription()
ticket.edit(
source_project_value=person,
resource='service_module/slapos_crm_delete_reminder',
)
ticket.validate()
ticket.suspend()
self.tic()
result = ticket.\
RegularisationRequest_deleteHostingSubscriptionList('footag')
self.assertFalse(result)
self.tic()
class TestSlapOSComputer_notifyWrongAllocationScope(testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def afterSetUp(self):
super(TestSlapOSComputer_notifyWrongAllocationScope, self).afterSetUp()
self.new_id = self.generateNewId()
self._cancelTestSupportRequestList()
def _cancelTestSupportRequestList(self):
for support_request in self.portal.portal_catalog(
portal_type="Support Request",
title="[MONITORING] % TESTCOMPT-%",
simulation_state=["validated", "suspended"]):
support_request.invalidate()
self.tic()
def _makeComputer(self,new_id):
# Clone computer document
person = self.portal.person_module.template_member\
.Base_createCloneDocument(batch_mode=1)
computer = self.portal.computer_module\
.template_computer.Base_createCloneDocument(batch_mode=1)
computer.edit(
title="computer ticket %s" % (new_id, ),
reference="TESTCOMPT-%s" % (new_id, ),
source_administration_value=person
)
computer.validate()
return computer
def _makePerson(self, new_id):
# Clone computer document
person = self.portal.person_module.template_member\
.Base_createCloneDocument(batch_mode=1)
person.edit(reference='TESTPERSON-%s' % (new_id, ))
person.immediateReindexObject()
return person
def _updatePersonAssignment(self, person, role='role/member'):
for assignment in person.contentValues(portal_type="Assignment"):
assignment.cancel()
assignment = person.newContent(portal_type='Assignment')
assignment.setRole(role)
assignment.setStartDate(DateTime())
assignment.open()
return assignment
def _getGeneratedSupportRequest(self, computer):
request_title = '[MONITORING] We have changed allocation scope for %s' % \
computer.getReference()
support_request = self.portal.portal_catalog.getResultValue(
portal_type = 'Support Request',
title = request_title,
simulation_state = 'suspended',
source_project_uid = computer.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('NotificationTool_getDocumentValue',
'reference=None',
'assert reference == "slapos-crm-computer_allocation_scope.notification"\n' \
'return context.restrictedTraverse(' \
'context.REQUEST["test_computerNotAllowedAllocationScope_OpenPublic"])')
@simulate('SupportRequest_trySendNotificationMessage',
'message_title, message, source_relative_url, interval_of_day=1',
'context.portal_workflow.doActionFor(' \
'context, action="edit_action", ' \
'comment="Visited by SupportRequest_trySendNotificationMessage ' \
'%s %s %s %s" % (message_title, message, source_relative_url, interval_of_day))\n' \
'return 1')
def test_computerNotAllowedAllocationScope_OpenPublic(self):
computer = self._makeComputer(self.new_id)
person = computer.getSourceAdministrationValue()
self._updatePersonAssignment(person, 'role/member')
self.portal.REQUEST['test_computerNotAllowedAllocationScope_OpenPublic'] = \
self._makeNotificationMessage(computer.getReference())
computer.edit(allocation_scope='open/public')
computer.Computer_checkAndUpdateAllocationScope()
self.tic()
self.assertEquals(computer.getAllocationScope(), 'open/personal')
ticket = self._getGeneratedSupportRequest(computer)
self.assertEquals(ticket.getSimulationState(), 'suspended')
self.assertEqual('Visited by SupportRequest_trySendNotificationMessage ' \
'%s %s %s %s' % \
('We have changed allocation scope for %s' % computer.getReference(),
'Test NM content\n%s\n' % computer.getReference(), person.getRelativeUrl(), '1'),
ticket.workflow_history['edit_workflow'][-1]['comment'])
@simulate('NotificationTool_getDocumentValue',
'reference=None',
'assert reference == "slapos-crm-computer_allocation_scope.notification"\n' \
'return context.restrictedTraverse(' \
'context.REQUEST["test_computerNotAllowedAllocationScope_OpenFriend"])')
@simulate('SupportRequest_trySendNotificationMessage',
'message_title, message, source_relative_url, interval_of_day=1',
'context.portal_workflow.doActionFor(' \
'context, action="edit_action", ' \
'comment="Visited by SupportRequest_trySendNotificationMessage ' \
'%s %s %s %s" % (message_title, message, source_relative_url, interval_of_day))\n' \
'return 1')
def test_computerNotAllowedAllocationScope_OpenFriend(self):
computer = self._makeComputer(self.new_id)
person = computer.getSourceAdministrationValue()
self._updatePersonAssignment(person, 'role/member')
self.portal.REQUEST['test_computerNotAllowedAllocationScope_OpenFriend'] = \
self._makeNotificationMessage(computer.getReference())
friend_person = self._makePerson(self.generateNewId())
computer.edit(allocation_scope='open/friend',
destination_section=friend_person.getRelativeUrl())
computer.Computer_checkAndUpdateAllocationScope()
self.tic()
self.assertEquals(computer.getAllocationScope(), 'open/personal')
ticket = self._getGeneratedSupportRequest(computer)
self.assertEquals(ticket.getSimulationState(), 'suspended')
self.assertEqual('Visited by SupportRequest_trySendNotificationMessage ' \
'%s %s %s %s' % \
('We have changed allocation scope for %s' % computer.getReference(),
'Test NM content\n%s\n' % computer.getReference(), person.getRelativeUrl(), '1'),
ticket.workflow_history['edit_workflow'][-1]['comment'])
@simulate('NotificationTool_getDocumentValue',
'reference=None',
'assert reference == "slapos-crm-computer_personal_allocation_scope.notification"\n' \
'return context.restrictedTraverse(' \
'context.REQUEST["test_computerToCloseAllocationScope_OpenPersonal"])')
@simulate('SupportRequest_trySendNotificationMessage',
'message_title, message, source_relative_url, interval_of_day=1',
'context.portal_workflow.doActionFor(' \
'context, action="edit_action", ' \
'comment="Visited by SupportRequest_trySendNotificationMessage ' \
'%s %s %s %s" % (message_title, message, source_relative_url, interval_of_day))\n' \
'return 1')
def test_computerToCloseAllocationScope_OpenPersonal(self):
computer = self._makeComputer(self.new_id)
person = computer.getSourceAdministrationValue()
self.portal.REQUEST['test_computerToCloseAllocationScope_OpenPersonal'] = \
self._makeNotificationMessage(computer.getReference())
friend_person = self._makePerson(self.generateNewId())
computer.edit(allocation_scope='open/personal',
destination_section=friend_person.getRelativeUrl())
computer.Computer_checkAndUpdatePersonalAllocationScope()
self.tic()
self.assertEquals(computer.getAllocationScope(), 'close/termination')
ticket = self._getGeneratedSupportRequest(computer)
self.assertEquals(ticket.getSimulationState(), 'suspended')
self.assertEqual('Visited by SupportRequest_trySendNotificationMessage ' \
'%s %s %s %s' % \
('We have changed allocation scope for %s' % computer.getReference(),
'Test NM content\n%s\n' % computer.getReference(), person.getRelativeUrl(), '1'),
ticket.workflow_history['edit_workflow'][-1]['comment'])
def test_computerNormalAllocationScope_OpenPersonal(self):
computer = self._makeComputer(self.new_id)
person = computer.getSourceAdministrationValue()
self._updatePersonAssignment(person, 'role/member')
computer.edit(allocation_scope='open/personal')
computer.Computer_checkAndUpdateAllocationScope()
self.tic()
self.assertEquals(computer.getAllocationScope(), 'open/personal')
def test_computerAllowedAllocationScope_OpenPublic(self):
computer = self._makeComputer(self.new_id)
person = computer.getSourceAdministrationValue()
self._updatePersonAssignment(person, 'role/service_provider')
computer.edit(allocation_scope='open/public')
computer.Computer_checkAndUpdateAllocationScope()
self.tic()
self.assertEquals(computer.getAllocationScope(), 'open/public')
def test_computerAllowedAllocationScope_OpenFriend(self):
computer = self._makeComputer(self.new_id)
person = computer.getSourceAdministrationValue()
self._updatePersonAssignment(person, 'role/service_provider')
friend_person = self._makePerson(self.generateNewId())
computer.edit(allocation_scope='open/friend',
destination_section=friend_person.getRelativeUrl())
computer.Computer_checkAndUpdateAllocationScope()
self.tic()
self.assertEquals(computer.getAllocationScope(), 'open/friend')
class TestSlapOSComputer_CheckState(testSlapOSMixin):
def beforeTearDown(self):
self._cancelTestSupportRequestList()
transaction.abort()
def afterSetUp(self):
super(TestSlapOSComputer_CheckState, self).afterSetUp()
self.new_id = self.generateNewId()
self._cancelTestSupportRequestList()
def _cancelTestSupportRequestList(self):
for support_request in self.portal.portal_catalog(
portal_type="Support Request",
title="[MONITORING] % TESTCOMPT-%",
simulation_state=["validated", "suspended"]):
support_request.invalidate()
self.tic()
def _makeNotificationMessage(self, reference):
notification_message = self.portal.notification_message_module.newContent(
portal_type="Notification Message",
title='The Computer %s has not contacted the server for more than 24 hours' % reference,
text_content='Test NM content<br/>%s<br/>' % reference,
content_type='text/html',
)
return notification_message.getRelativeUrl()
def _getGeneratedSupportRequest(self, source_uid, request_title):
support_request = self.portal.portal_catalog.getResultValue(
portal_type = 'Support Request',
title = request_title,
simulation_state = 'validated',
source_project_uid = source_uid
)
return support_request
def _makeComputer(self,new_id):
# Clone computer document
person = self.portal.person_module.template_member\
.Base_createCloneDocument(batch_mode=1)
computer = self.portal.computer_module\
.template_computer.Base_createCloneDocument(batch_mode=1)
computer.edit(
title="computer ticket %s" % (new_id, ),
reference="TESTCOMPT-%s" % (new_id, ),
source_administration_value=person
)
computer.validate()
return computer
def _makeHostingSubscription(self, new_id):
person = self.portal.person_module.template_member\
.Base_createCloneDocument(batch_mode=1)
hosting_subscription = self.portal\
.hosting_subscription_module.template_hosting_subscription\
.Base_createCloneDocument(batch_mode=1)
hosting_subscription.validate()
hosting_subscription.edit(
title= "Test hosting sub ticket %s" % new_id,
reference="TESTHST-%s" % new_id,
destination_section_value=person
)
return hosting_subscription
def _makeSoftwareInstance(self, hosting_subscription, software_url):
kw = dict(
software_release=software_url,
software_type=self.generateNewSoftwareType(),
instance_xml=self.generateSafeXml(),
sla_xml=self.generateSafeXml(),
shared=False,
software_title=hosting_subscription.getTitle(),
state='started'
)
hosting_subscription.requestStart(**kw)
hosting_subscription.requestInstance(**kw)
def _simulateBase_generateSupportRequestForSlapOS(self):
script_name = 'Base_generateSupportRequestForSlapOS'
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, **kw',
'# Script body\n'
"""return 'Visited Base_generateSupportRequestForSlapOS'""")
transaction.commit()
def _dropBase_generateSupportRequestForSlapOS(self):
script_name = 'Base_generateSupportRequestForSlapOS'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
transaction.commit()
self.assertFalse(script_name in self.portal.portal_skins.custom.objectIds())
def test_Computer_checkState_call_support_request(self):
computer = self._makeComputer(self.new_id)
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":"%s" % (DateTime() - 1.1)}
)
self._simulateBase_generateSupportRequestForSlapOS()
try:
result = computer.Computer_checkState()
finally:
self._dropBase_generateSupportRequestForSlapOS()
self.assertEqual('Visited Base_generateSupportRequestForSlapOS',
result)
def test_Computer_checkState_empty_cache(self):
computer = self._makeComputer(self.new_id)
self._simulateBase_generateSupportRequestForSlapOS()
try:
result = computer.Computer_checkState()
finally:
self._dropBase_generateSupportRequestForSlapOS()
self.assertEqual('Visited Base_generateSupportRequestForSlapOS',
result)
@simulate('NotificationTool_getDocumentValue',
'reference=None',
'assert reference == "slapos-crm-computer_check_state.notification"\n' \
'return context.restrictedTraverse(' \
'context.REQUEST["test_Computer_checkState_notify"])')
@simulate('SupportRequest_trySendNotificationMessage',
'message_title, message, source_relative_url, interval_of_day=1',
'context.portal_workflow.doActionFor(' \
'context, action="edit_action", ' \
'comment="Visited by SupportRequest_trySendNotificationMessage ' \
'%s %s %s %s" % (message_title, message, source_relative_url, interval_of_day))')
def test_Computer_checkState_notify(self):
computer = self._makeComputer(self.new_id)
person = computer.getSourceAdministrationValue()
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":"%s" % (DateTime() - 1.1)}
)
message_interval_per_day = 5
self.portal.REQUEST['test_Computer_checkState_notify'] = \
self._makeNotificationMessage(computer.getReference())
computer.Computer_checkState()
self.tic()
ticket_title = "[MONITORING] Lost contact with computer %s" % computer.getReference()
ticket = self._getGeneratedSupportRequest(computer.getUid(), ticket_title)
self.assertNotEqual(ticket, None)
self.assertEqual('Visited by SupportRequest_trySendNotificationMessage ' \
'%s %s %s %d' % ( \
ticket_title.replace('[MONITORING] ', ''),
'Test NM content\n%s\n' % computer.getReference(),
person.getRelativeUrl(), message_interval_per_day),
ticket.workflow_history['edit_workflow'][-1]['comment'])
@simulate('NotificationTool_getDocumentValue',
'reference=None',
'assert reference == "slapos-crm-computer_check_state.notification"\n' \
'return context.restrictedTraverse(' \
'context.REQUEST["test_Computer_checkState_empty_cache_notify"])')
@simulate('SupportRequest_trySendNotificationMessage',
'message_title, message, source_relative_url, interval_of_day=1',
'context.portal_workflow.doActionFor(' \
'context, action="edit_action", ' \
'comment="Visited by SupportRequest_trySendNotificationMessage ' \
'%s %s %s %s" % (message_title, message, source_relative_url, interval_of_day))')
def test_Computer_checkState_empty_cache_notify(self):
computer = self._makeComputer(self.new_id)
person = computer.getSourceAdministrationValue()
message_interval_per_day = 5
self.portal.REQUEST['test_Computer_checkState_empty_cache_notify'] = \
self._makeNotificationMessage(computer.getReference())
computer.Computer_checkState()
self.tic()
ticket_title = "[MONITORING] No information about %s" % computer.getReference()
ticket = self._getGeneratedSupportRequest(computer.getUid(), ticket_title)
self.assertNotEqual(ticket, None)
self.assertEqual('Visited by SupportRequest_trySendNotificationMessage ' \
'%s %s %s %d' % ( \
ticket_title.replace('[MONITORING] ', ''),
'Test NM content\n%s\n' % computer.getReference(),
person.getRelativeUrl(), message_interval_per_day),
ticket.workflow_history['edit_workflow'][-1]['comment'])
@simulate('NotificationTool_getDocumentValue',
'reference=None',
'assert reference == "slapos-crm-hosting_subscription_state.notification"\n' \
'return context.restrictedTraverse(' \
'context.REQUEST["test_SoftwareInstance_checkState"])')
@simulate('SupportRequest_trySendNotificationMessage',
'message_title, message, source_relative_url, interval_of_day=1',
'context.portal_workflow.doActionFor(' \
'context, action="edit_action", ' \
'comment="Visited by SupportRequest_trySendNotificationMessage ' \
'%s %s %s %s" % (message_title, message, source_relative_url, interval_of_day))')
def test_SoftwareInstance_checkState(self):
host_sub = self._makeHostingSubscription(self.new_id)
self._makeSoftwareInstance(host_sub,self.generateNewSoftwareReleaseUrl())
instance = host_sub.getPredecessorValue()
person_url = host_sub.getDestinationSection()
instance.workflow_history['edit_workflow'] = [{
'comment':'edit',
'error_message': '',
'actor': 'ERP5TypeTestCase',
'state': 'current',
'time': DateTime('2012/11/30 11:11'),
'action': 'edit'
}]
memcached_dict = self.portal.portal_memcached.getMemcachedDict(
key_prefix='slap_tool',
plugin_path='portal_memcached/default_memcached_plugin')
memcached_dict[instance.getReference()] = json.dumps(
{"created_at":"%s" % DateTime(), "text":"#error "}
)
message_interval_per_day = 0
self.portal.REQUEST['test_SoftwareInstance_checkState'] = \
self._makeNotificationMessage(instance.getReference())
self.tic()
ticket_url = instance.SoftwareInstance_checkState()
self.tic()
self.assertNotEqual(ticket_url, None)
ticket_title = "[MONITORING] Service %s in error state" % host_sub.getReference()
ticket = self._getGeneratedSupportRequest(host_sub.getUid(), ticket_title)
self.assertNotEqual(ticket, None)
self.assertEqual('Visited by SupportRequest_trySendNotificationMessage ' \
'%s %s %s %s' % ( \
ticket_title.replace('[MONITORING] ', ''),
'Test NM content\n%s\n' % instance.getReference(),
person_url, message_interval_per_day),
ticket.workflow_history['edit_workflow'][-1]['comment'])
master/bt5/slapos_crm/TestTemplateItem/testSlapOSCRMSupportRequestGeneration.py deleted 100644 → 0
View file @ 0ba4d720
# -*- coding:utf-8 -*-
##############################################################################
#
# Copyright (c) 2002-2013 Nexedi SA and Contributors. All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility 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
# guarantees 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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
#
##############################################################################
import transaction
from Products.SlapOS.tests.testSlapOSMixin import \
testSlapOSMixin
from Products.ERP5Type.tests.utils import createZODBPythonScript
import json
from DateTime import DateTime
class TestSlapOSCloudSupportRequestGeneration(testSlapOSMixin):
def afterSetUp(self):
super(TestSlapOSCloudSupportRequestGeneration, self).afterSetUp()
self.new_id = self.generateNewId()
self._cancelTestSupportRequestList()
def _cancelTestSupportRequestList(self):
for support_request in self.portal.portal_catalog(
portal_type="Support Request",
title="%Test Support Request %",
simulation_state=["validated", "suspended"]):
support_request.invalidate()
self.tic()
def _updatePersonAssignment(self, person, role='role/member'):
for assignment in person.contentValues(portal_type="Assignment"):
assignment.cancel()
assignment = person.newContent(portal_type='Assignment')
assignment.setRole(role)
assignment.setStartDate(DateTime())
assignment.open()
return assignment
def _makePerson(self, new_id):
# Clone computer document
person = self.portal.person_module.template_member\
.Base_createCloneDocument(batch_mode=1)
person.edit(reference='TESTPERSON-%s' % (new_id, ))
person.immediateReindexObject()
return person
def _makeComputer(self,new_id):
# Clone computer document
person = self.portal.person_module.template_member\
.Base_createCloneDocument(batch_mode=1)
computer = self.portal.computer_module\
.template_computer.Base_createCloneDocument(batch_mode=1)
computer.edit(
title="computer ticket %s" % (new_id, ),
reference="TESTCOMPT-%s" % (new_id, ),
source_administration_value=person
)
computer.validate()
return computer
def _makeComputerPartitions(self,computer):
for i in range(1, 5):
id_ = 'partition%s' % (i, )
p = computer.newContent(portal_type='Computer Partition',
id=id_,
title=id_,
reference=id_,
default_network_address_ip_address='ip_address_%s' % i,
default_network_address_netmask='netmask_%s' % i)
p.markFree()
p.validate()
def _makeSoftwareRelease(self, new_id, software_release_url=None):
software_release = self.portal.software_release_module\
.template_software_release.Base_createCloneDocument(batch_mode=1)
software_release.edit(
url_string=software_release_url or self.generateNewSoftwareReleaseUrl(),
reference='TESTSOFTRELS-%s' % new_id,
title='Start requested for %s' % new_id
)
software_release.release()
return software_release
def _makeSoftwareInstallation(self, new_id, computer, software_release_url):
software_installation = self.portal\
.software_installation_module.template_software_installation\
.Base_createCloneDocument(batch_mode=1)
software_installation.edit(
url_string=software_release_url,
aggregate=computer.getRelativeUrl(),
reference='TESTSOFTINSTS-%s' % new_id,
title='Start requested for %s' % computer.getUid()
)
software_installation.validate()
software_installation.requestStart()
return software_installation
def _makeHostingSubscription(self, new_id):
person = self.portal.person_module.template_member\
.Base_createCloneDocument(batch_mode=1)
hosting_subscription = self.portal\
.hosting_subscription_module.template_hosting_subscription\
.Base_createCloneDocument(batch_mode=1)
hosting_subscription.validate()
hosting_subscription.edit(
title= "Test hosting sub ticket %s" % new_id,
reference="TESTHST-%s" % new_id,
destination_section_value=person
)
return hosting_subscription
def _makeSoftwareInstance(self, hosting_subscription, software_url):
kw = dict(
software_release=software_url,
software_type=self.generateNewSoftwareType(),
instance_xml=self.generateSafeXml(),
sla_xml=self.generateSafeXml(),
shared=False,
software_title=hosting_subscription.getTitle(),
state='started'
)
hosting_subscription.requestStart(**kw)
hosting_subscription.requestInstance(**kw)
def test_computer_Base_generateSupportRequestForSlapOS(self):
self._dropBase_generateSupportRequestForSlapOS()
title = "Test Support Request %s" % self.new_id
computer = self._makeComputer(self.new_id)
support_request = computer.Base_generateSupportRequestForSlapOS(
title, title, computer.getRelativeUrl()
)
self.tic()
self.assertNotEqual(support_request, None)
support_request = self.portal.restrictedTraverse(support_request)
# The support request is added to computer owner.
self.assertEquals(support_request.getDestinationDecision(),
computer.getSourceAdministration())
def test_software_instance_Base_generateSupportRequestForSlapOS(self):
host_sub = self._makeHostingSubscription(self.new_id)
self._makeSoftwareInstance(host_sub,self.generateNewSoftwareReleaseUrl())
instance = host_sub.getPredecessorValue()
title = "Test Support Request %s" % self.new_id
support_request = instance.Base_generateSupportRequestForSlapOS(
title, title, instance.getRelativeUrl()
)
self.tic()
self.assertNotEqual(support_request, None)
support_request = self.portal.restrictedTraverse(support_request)
# The support request is added to computer owner.
self.assertEquals(support_request.getDestinationDecision(),
host_sub.getDestinationSection())
def test_hosting_subscription_Base_generateSupportRequestForSlapOS(self):
host_sub = self._makeHostingSubscription(self.new_id)
title = "Test Support Request %s" % self.new_id
support_request = host_sub.Base_generateSupportRequestForSlapOS(
title, title, host_sub.getRelativeUrl()
)
self.tic()
self.assertNotEqual(support_request, None)
support_request = self.portal.restrictedTraverse(support_request)
# The support request is added to computer owner.
self.assertEquals(support_request.getDestinationDecision(),
host_sub.getDestinationSection())
def test_Base_generateSupportRequestForSlapOS_do_not_recreate_if_open(self):
title = "Test Support Request %s" % self.new_id
computer = self._makeComputer(self.new_id)
computer.Base_generateSupportRequestForSlapOS(
title, title, computer.getRelativeUrl()
)
self.tic()
support_request = computer.Base_generateSupportRequestForSlapOS(
title, title, computer.getRelativeUrl()
)
self.tic()
self.assertEqual(support_request, None)
def test_ERP5Site_isSupportRequestCreationClosed(self):
person = self._makePerson(self.new_id)
other_person = self._makePerson('other-%s' % self.new_id)
url = person.getRelativeUrl()
self.assertFalse(self.portal.ERP5Site_isSupportRequestCreationClosed(url))
self.assertFalse(self.portal.ERP5Site_isSupportRequestCreationClosed())
def newSupportRequest():
sr = self.portal.support_request_module.newContent(\
title="Test Support Request POIUY",
destination_decision=url)
sr.validate()
sr.immediateReindexObject()
newSupportRequest()
self.assertFalse(self.portal.ERP5Site_isSupportRequestCreationClosed(url))
newSupportRequest()
self.assertFalse(self.portal.ERP5Site_isSupportRequestCreationClosed(url))
newSupportRequest()
self.assertFalse(self.portal.ERP5Site_isSupportRequestCreationClosed(url))
newSupportRequest()
self.assertFalse(self.portal.ERP5Site_isSupportRequestCreationClosed(url))
newSupportRequest()
self.assertTrue(self.portal.ERP5Site_isSupportRequestCreationClosed(url))
self.assertTrue(self.portal.ERP5Site_isSupportRequestCreationClosed())
self.assertFalse(self.portal.ERP5Site_isSupportRequestCreationClosed(
other_person.getRelativeUrl()))
def test_ERP5Site_isSupportRequestCreationClosed_submitted_state(self):
person = self._makePerson(self.new_id)
url = person.getRelativeUrl()
self.assertFalse(self.portal.ERP5Site_isSupportRequestCreationClosed(url))
self.assertFalse(self.portal.ERP5Site_isSupportRequestCreationClosed())
def newSupportRequest():
sr = self.portal.support_request_module.newContent(\
title="Test Support Request POIUY",
destination_decision=url)
sr.validate()
sr.suspend()
sr.immediateReindexObject()
# Create five tickets, the limit of ticket creation
newSupportRequest()
newSupportRequest()
newSupportRequest()
newSupportRequest()
newSupportRequest()
self.assertFalse(self.portal.ERP5Site_isSupportRequestCreationClosed(url))
self.assertFalse(self.portal.ERP5Site_isSupportRequestCreationClosed())
def test_Base_generateSupportRequestForSlapOS_recreate_if_closed(self):
title = "Test Support Request %s" % self.new_id
computer = self._makeComputer(self.new_id)
support_request = computer.Base_generateSupportRequestForSlapOS(
title, title, computer.getRelativeUrl()
)
self.tic()
self.portal.restrictedTraverse(support_request).invalidate()
self.tic()
support_request = computer.Base_generateSupportRequestForSlapOS(
title, title, computer.getRelativeUrl()
)
self.tic()
self.assertNotEqual(support_request,None)
def _simulateBase_generateSupportRequestForSlapOS(self):
script_name = 'Base_generateSupportRequestForSlapOS'
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, **kw',
'# Script body\n'
"""return 'Visited Base_generateSupportRequestForSlapOS'""")
transaction.commit()
def _dropBase_generateSupportRequestForSlapOS(self):
script_name = 'Base_generateSupportRequestForSlapOS'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
transaction.commit()
self.assertFalse(script_name in self.portal.portal_skins.custom.objectIds())
def test_Computer_checkState(self):
computer = self._makeComputer(self.new_id)
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":"%s" % (DateTime() - 1.1)}
)
self._simulateBase_generateSupportRequestForSlapOS()
try:
result = computer.Computer_checkState()
finally:
self._dropBase_generateSupportRequestForSlapOS()
self.assertEqual('Visited Base_generateSupportRequestForSlapOS',
result)
def test_SupportRequest_trySendNotificationMessage(self):
computer = self._makeComputer(self.new_id)
person = computer.getSourceAdministrationValue()
title = "Test Support Request %s" % self.new_id
text_content='Test NM content<br/>%s<br/>' % self.new_id
support_request = self.portal.support_request_module.newContent(\
title=title, description=title,
destination_decision=computer.getSourceAdministration(),
source_project_value=computer.getRelativeUrl())
support_request.validate()
time_before_next = 2
self.tic()
first_event = support_request.SupportRequest_trySendNotificationMessage(
message_title=title, message=text_content,
source_relative_url=person.getRelativeUrl(),
interval_of_day=time_before_next
)
self.assertNotEqual(first_event, None)
first_event.edit(start_date=(DateTime() - 1.8))
self.tic()
event = support_request.SupportRequest_trySendNotificationMessage(
message_title=title, message=text_content,
source_relative_url=person.getRelativeUrl(),
interval_of_day=time_before_next
)
self.assertEqual(event, None)
time_before_next = 1
event = support_request.SupportRequest_trySendNotificationMessage(
message_title=title, message=text_content,
source_relative_url=person.getRelativeUrl(),
interval_of_day=time_before_next
)
self.assertEqual(event.getTitle(), title)
def test_Computer_checkState_empty_cache(self):
computer = self._makeComputer(self.new_id)
self._simulateBase_generateSupportRequestForSlapOS()
try:
result = computer.Computer_checkState()
finally:
self._dropBase_generateSupportRequestForSlapOS()
self.assertEqual('Visited Base_generateSupportRequestForSlapOS',
result)
def test_SoftwareInstance_checkState_error_new_instance(self):
host_sub = self._makeHostingSubscription(self.new_id)
self._makeSoftwareInstance(host_sub,self.generateNewSoftwareReleaseUrl())
instance = host_sub.getPredecessorValue()
memcached_dict = self.portal.portal_memcached.getMemcachedDict(
key_prefix='slap_tool',
plugin_path='portal_memcached/default_memcached_plugin')
memcached_dict[instance.getReference()] = json.dumps(
{"created_at":"%s" % DateTime(), "text":"#error "}
)
self._simulateBase_generateSupportRequestForSlapOS()
try:
result = instance.SoftwareInstance_checkState()
finally:
self._dropBase_generateSupportRequestForSlapOS()
self.assertNotEqual('Visited Base_generateSupportRequestForSlapOS',
result)
def test_SoftwareInstance_checkState_error_out_time(self):
host_sub = self._makeHostingSubscription(self.new_id)
self._makeSoftwareInstance(host_sub,self.generateNewSoftwareReleaseUrl())
instance = host_sub.getPredecessorValue()
instance.workflow_history['edit_workflow'] = [{
'comment':'edit',
'error_message': '',
'actor': 'ERP5TypeTestCase',
'state': 'current',
'time': DateTime('2012/11/30 11:11'),
'action': 'edit'
}]
memcached_dict = self.portal.portal_memcached.getMemcachedDict(
key_prefix='slap_tool',
plugin_path='portal_memcached/default_memcached_plugin')
memcached_dict[instance.getReference()] = json.dumps(
{"created_at":"%s" % DateTime(), "text":"#error "}
)
self._simulateBase_generateSupportRequestForSlapOS()
try:
result = instance.SoftwareInstance_checkState()
finally:
self._dropBase_generateSupportRequestForSlapOS()
self.assertEqual('Visited Base_generateSupportRequestForSlapOS',
result)
def test_SoftwareInstance_checkState_error_in_time(self):
host_sub = self._makeHostingSubscription(self.new_id)
self._makeSoftwareInstance(host_sub,self.generateNewSoftwareReleaseUrl())
instance = host_sub.getPredecessorValue()
memcached_dict = self.portal.portal_memcached.getMemcachedDict(
key_prefix='slap_tool',
plugin_path='portal_memcached/default_memcached_plugin')
memcached_dict[instance.getReference()] = json.dumps(
{"created_at":"%s" % (DateTime()), "text":"#error "}
)
self._simulateBase_generateSupportRequestForSlapOS()
try:
result = instance.SoftwareInstance_checkState()
finally:
self._dropBase_generateSupportRequestForSlapOS()
self.assertNotEqual('Visited Base_generateSupportRequestForSlapOS',
result)
def test_SoftwareInstance_checkState_access_in_time(self):
host_sub = self._makeHostingSubscription(self.new_id)
self._makeSoftwareInstance(host_sub,self.generateNewSoftwareReleaseUrl())
instance = host_sub.getPredecessorValue()
memcached_dict = self.portal.portal_memcached.getMemcachedDict(
key_prefix='slap_tool',
plugin_path='portal_memcached/default_memcached_plugin')
memcached_dict[instance.getReference()] = json.dumps(
{"created_at":"%s" % (DateTime()), "text":"#access "}
)
self._simulateBase_generateSupportRequestForSlapOS()
try:
result = instance.SoftwareInstance_checkState()
finally:
self._dropBase_generateSupportRequestForSlapOS()
self.assertNotEqual('Visited Base_generateSupportRequestForSlapOS',
result)
def test_SoftwareInstance_checkState_empty_cache(self):
host_sub = self._makeHostingSubscription(self.new_id)
self._makeSoftwareInstance(host_sub,self.generateNewSoftwareReleaseUrl())
instance = host_sub.getPredecessorValue()
self._simulateBase_generateSupportRequestForSlapOS()
try:
result = instance.SoftwareInstance_checkState()
finally:
self._dropBase_generateSupportRequestForSlapOS()
self.assertNotEqual('Visited Base_generateSupportRequestForSlapOS',
result)
def test_SoftwareRelease_testForAllocation(self):
software_release = self._makeSoftwareRelease(self.new_id)
computer = self._makeComputer(self.new_id)
computer.edit(allocation_scope = 'open/public', capacity_scope = 'open')
self._makeComputerPartitions(computer)
self._makeSoftwareInstallation(
self.new_id, computer, software_release.getUrlString()
)
self.tic()
self._simulateBase_generateSupportRequestForSlapOS()
try:
result = software_release.SoftwareRelease_testForAllocation()
finally:
self._dropBase_generateSupportRequestForSlapOS()
self.assertNotEqual('Visited Base_generateSupportRequestForSlapOS',
result)
def test_SoftwareRelease_testForAllocation_no_public_computer(self):
software_release = self._makeSoftwareRelease(self.new_id)
computer = self._makeComputer(self.new_id)
computer.edit(capacity_scope = 'open')
self._makeComputerPartitions(computer)
self._makeSoftwareInstallation(
self.new_id, computer, software_release.getUrlString()
)
self.tic()
self._simulateBase_generateSupportRequestForSlapOS()
try:
result = software_release.SoftwareRelease_testForAllocation()
finally:
self._dropBase_generateSupportRequestForSlapOS()
self.assertEqual('Visited Base_generateSupportRequestForSlapOS',
result)
def test_SoftwareRelease_testForAllocation_no_open_computer(self):
software_release = self._makeSoftwareRelease(self.new_id)
computer = self._makeComputer(self.new_id)
computer.edit(allocation_scope = 'open/public')
self._makeComputerPartitions(computer)
self._makeSoftwareInstallation(
self.new_id, computer, software_release.getUrlString()
)
self.tic()
self._simulateBase_generateSupportRequestForSlapOS()
try:
result = software_release.SoftwareRelease_testForAllocation()
finally:
self._dropBase_generateSupportRequestForSlapOS()
self.assertEqual('Visited Base_generateSupportRequestForSlapOS',
result)
def test_SoftwareRelease_testForAllocation_no_free_partitions(self):
software_release = self._makeSoftwareRelease(self.new_id)
computer = self._makeComputer(self.new_id)
computer.edit(allocation_scope = 'open/public', capacity_scope = 'open')
self._makeComputerPartitions(computer)
computer.partition1.markBusy()
computer.partition2.markBusy()
computer.partition3.markBusy()
computer.partition4.markBusy()
self._makeSoftwareInstallation(
self.new_id, computer, software_release.getUrlString()
)
self.tic()
self._simulateBase_generateSupportRequestForSlapOS()
try:
result = software_release.SoftwareRelease_testForAllocation()
finally:
self._dropBase_generateSupportRequestForSlapOS()
self.assertEqual('Visited Base_generateSupportRequestForSlapOS',
result)
def _simulateSoftwareInstance_checkState(self):
script_name = 'SoftwareInstance_checkState'
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, **kw',
'# Script body\n'
"""portal_workflow = context.portal_workflow
portal_workflow.doActionFor(context, action='edit_action', comment='Visited by SoftwareInstance_checkState') """ )
transaction.commit()
def _dropSoftwareInstance_checkState(self):
script_name = 'SoftwareInstance_checkState'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
transaction.commit()
def testHostingSubscription_CheckInstanceState(self):
host_sub = self._makeHostingSubscription(self.new_id)
self._makeSoftwareInstance(host_sub, self.generateNewSoftwareReleaseUrl())
instance = host_sub.getPredecessorValue()
self._simulateSoftwareInstance_checkState()
try:
host_sub.HostingSubscription_CheckInstanceState()
self.tic()
finally:
self._dropSoftwareInstance_checkState()
self.assertEqual('Visited by SoftwareInstance_checkState',
instance.workflow_history['edit_workflow'][-1]['comment'])
def _simulateHostingSubscription_CheckInstanceState(self):
script_name = 'HostingSubscription_CheckInstanceState'
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, **kw',
'# Script body\n'
"""portal_workflow = context.portal_workflow
portal_workflow.doActionFor(context, action='edit_action', comment='Visited by HostingSubscription_CheckInstanceState') """ )
transaction.commit()
def _dropHostingSubscription_CheckInstanceState(self):
script_name = 'HostingSubscription_CheckInstanceState'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
transaction.commit()
def test_Computer_checkSoftwareInstanceState(self):
computer = self._makeComputer(self.new_id)
self._makeComputerPartitions(computer)
host_sub = self._makeHostingSubscription(self.new_id)
self._makeSoftwareInstance(host_sub, self.generateNewSoftwareReleaseUrl())
instance = host_sub.getPredecessorValue()
instance.edit(aggregate_value = computer.partition1.getRelativeUrl())
computer.partition1.markBusy()
self.tic()
self._simulateHostingSubscription_CheckInstanceState()
try:
computer.Computer_checkSoftwareInstanceState()
self.tic()
finally:
self._dropHostingSubscription_CheckInstanceState()
self.assertEqual('Visited by HostingSubscription_CheckInstanceState',
host_sub.workflow_history['edit_workflow'][-1]['comment'])
def test_Computer_checkSoftwareInstanceState_instance_not_allocated(self):
computer = self._makeComputer(self.new_id)
host_sub = self._makeHostingSubscription(self.new_id)
self._makeSoftwareInstance(host_sub, self.generateNewSoftwareReleaseUrl())
instance = host_sub.getPredecessorValue()
self.tic()
self._simulateHostingSubscription_CheckInstanceState()
try:
computer.Computer_checkSoftwareInstanceState()
self.tic()
finally:
self._dropHostingSubscription_CheckInstanceState()
self.assertNotEqual('Visited by HostingSubscription_CheckInstanceState',
instance.workflow_history['edit_workflow'][-1]['comment'])
def _simulateComputer_checkState(self):
script_name = 'Computer_checkState'
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, **kw',
'# Script body\n'
"""portal_workflow = context.portal_workflow
portal_workflow.doActionFor(context, action='edit_action', comment='Visited by Computer_checkState') """ )
transaction.commit()
def _dropComputer_checkState(self):
script_name = 'Computer_checkState'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
transaction.commit()
def test_Alarm_checkComputerState(self):
computer = self._makeComputer(self.new_id)
computer.edit(allocation_scope = 'open/public')
self._simulateComputer_checkState()
try:
self.portal.portal_alarms.slapos_check_computer_state.activeSense()
self.tic()
finally:
self._dropComputer_checkState()
self.assertEqual('Visited by Computer_checkState',
computer.workflow_history['edit_workflow'][-1]['comment'])
def test_Alarm_checkComputerState_no_public_computer(self):
computer = self._makeComputer(self.new_id)
self._simulateComputer_checkState()
try:
self.portal.portal_alarms.slapos_check_computer_state.activeSense()
self.tic()
finally:
self._dropComputer_checkState()
self.assertNotEqual('Visited by Computer_checkState',
computer.workflow_history['edit_workflow'][-1]['comment'])
def _simulateComputer_checkSoftwareInstanceState(self):
script_name = 'Computer_checkSoftwareInstanceState'
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, **kw',
'# Script body\n'
"""portal_workflow = context.portal_workflow
portal_workflow.doActionFor(context, action='edit_action', comment='Visited by Computer_checkSoftwareInstanceState') """ )
transaction.commit()
def _dropComputer_checkSoftwareInstanceState(self):
script_name = 'Computer_checkSoftwareInstanceState'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
transaction.commit()
def test_Alarm_checkSoftwareInstanceState(self):
computer = self._makeComputer(self.new_id)
computer.edit(allocation_scope = 'open/public')
self._simulateComputer_checkSoftwareInstanceState()
try:
self.portal.portal_alarms.slapos_check_software_instance_state.activeSense()
self.tic()
finally:
self._dropComputer_checkSoftwareInstanceState()
self.assertEqual('Visited by Computer_checkSoftwareInstanceState',
computer.workflow_history['edit_workflow'][-1]['comment'])
def test_Alarm_checkSoftwareInstanceState_no_public_computer(self):
computer = self._makeComputer(self.new_id)
self._simulateComputer_checkSoftwareInstanceState()
try:
self.portal.portal_alarms.slapos_check_software_instance_state.activeSense()
self.tic()
finally:
self._dropComputer_checkSoftwareInstanceState()
self.assertNotEqual('Visited by Computer_checkSoftwareInstanceState',
computer.workflow_history['edit_workflow'][-1]['comment'])
def _simulateSoftwareRelease_testForAllocation(self):
script_name = 'SoftwareRelease_testForAllocation'
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, **kw',
'# Script body\n'
"""portal_workflow = context.portal_workflow
portal_workflow.doActionFor(context, action='edit_action', comment='Visited by SoftwareRelease_testForAllocation') """ )
transaction.commit()
def _dropSoftwareRelease_testForAllocation(self):
script_name = 'SoftwareRelease_testForAllocation'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
transaction.commit()
def test_Alarm_checkCloudIsFull(self):
software_release_preference = self.portal.portal_preferences.getPreferredSoftwareReleaseToTestForSpace()
if software_release_preference != '':
software_release_list = software_release_preference.split("\n")
software_release = self._makeSoftwareRelease(
self.new_id, software_release_list[0])
self._simulateSoftwareRelease_testForAllocation()
try:
self.portal.portal_alarms.slapos_check_cloud_is_full.activeSense()
self.tic()
finally:
self._dropSoftwareRelease_testForAllocation()
self.assertEqual('Visited by SoftwareRelease_testForAllocation',
software_release.workflow_history['edit_workflow'][-1]['comment'])
def _simulateComputer_checkAndUpdateAllocationScope(self):
script_name = 'Computer_checkAndUpdateAllocationScope'
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, **kw',
'# Script body\n'
"""portal_workflow = context.portal_workflow
portal_workflow.doActionFor(context, action='edit_action', comment='Visited by Computer_checkAndUpdateAllocationScope') """ )
transaction.commit()
def _dropComputer_checkAndUpdateAllocationScope(self):
script_name = 'Computer_checkAndUpdateAllocationScope'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
transaction.commit()
def test_Alarm_notAllowedAllocationScope_OpenPublic(self):
computer = self._makeComputer(self.new_id)
computer.edit(allocation_scope = 'open/public')
self._simulateComputer_checkAndUpdateAllocationScope()
try:
self.portal.portal_alarms.slapos_crm_check_update_allocation_scope.activeSense()
self.tic()
finally:
self._dropComputer_checkAndUpdateAllocationScope()
self.assertEqual('Visited by Computer_checkAndUpdateAllocationScope',
computer.workflow_history['edit_workflow'][-1]['comment'])
def test_Alarm_notAllowedAllocationScope_OpenFriend(self):
computer = self._makeComputer(self.new_id)
computer.edit(allocation_scope = 'open/friend')
self._simulateComputer_checkAndUpdateAllocationScope()
try:
self.portal.portal_alarms.slapos_crm_check_update_allocation_scope.activeSense()
self.tic()
finally:
self._dropComputer_checkAndUpdateAllocationScope()
self.assertEqual('Visited by Computer_checkAndUpdateAllocationScope',
computer.workflow_history['edit_workflow'][-1]['comment'])
def test_Alarm_notAllowedAllocationScope_OpenPersonal(self):
computer = self._makeComputer(self.new_id)
computer.edit(allocation_scope = 'open/personal')
self._simulateComputer_checkAndUpdateAllocationScope()
try:
self.portal.portal_alarms.slapos_crm_check_update_allocation_scope.activeSense()
self.tic()
finally:
self._dropComputer_checkAndUpdateAllocationScope()
self.assertNotEqual('Visited by Computer_checkAndUpdateAllocationScope',
computer.workflow_history['edit_workflow'][-1]['comment'])
def test_Alarm_AllowedAllocationScope_OpenPersonal_old_computer(self):
computer = self._makeComputer(self.new_id)
computer.edit(allocation_scope = 'open/personal')
def getModificationDate(self):
return DateTime() - 50
from Products.ERP5Type.Base import Base
self._simulateComputer_checkAndUpdateAllocationScope()
original_get_modification = Base.getModificationDate
Base.getModificationDate = getModificationDate
try:
self.portal.portal_alarms.slapos_crm_check_update_allocation_scope.activeSense()
self.tic()
finally:
Base.getModificationDate = original_get_modification
self._dropComputer_checkAndUpdateAllocationScope()
self.assertEqual('Visited by Computer_checkAndUpdateAllocationScope',
computer.workflow_history['edit_workflow'][-1]['comment'])
def test_Alarm_AllowedAllocationScope_OpenPersonalWithSoftwareInstallation(self):
computer = self._makeComputer(self.new_id)
computer.edit(allocation_scope = 'open/personal')
software_installation = self._makeSoftwareInstallation(
self.new_id, computer, "http://...")
def getModificationDate(self):
return DateTime() - 50
from Products.ERP5Type.Base import Base
self._simulateComputer_checkAndUpdateAllocationScope()
original_get_modification = Base.getModificationDate
Base.getModificationDate = getModificationDate
try:
self.portal.portal_alarms.slapos_crm_check_update_allocation_scope.activeSense()
self.tic()
finally:
Base.getModificationDate = original_get_modification
self._dropComputer_checkAndUpdateAllocationScope()
self.assertNotEqual('Visited by Computer_checkAndUpdateAllocationScope',
computer.workflow_history['edit_workflow'][-1]['comment'])
def _simulateHostingSubscription_checkSofwareInstanceState(self):
script_name = 'HostingSubscription_checkSofwareInstanceAllocationState'
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, **kw',
'# Script body\n'
"""portal_workflow = context.portal_workflow
portal_workflow.doActionFor(context, action='edit_action', comment='Visited by HostingSubscription_checkSofwareInstanceAllocationState') """ )
transaction.commit()
def _dropHostingSubscription_checkSofwareInstanceState(self):
script_name = 'HostingSubscription_checkSofwareInstanceAllocationState'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
transaction.commit()
def test_Alarm_findAndNofitiyUnallocatedSoftwareInstance(self):
host_sub = self._makeHostingSubscription(self.new_id)
self._makeSoftwareInstance(host_sub, self.generateNewSoftwareReleaseUrl())
instance = host_sub.getPredecessorValue()
self.assertEqual(instance.getAggregate(""), "")
self._simulateHostingSubscription_checkSofwareInstanceState()
try:
self.portal.portal_alarms.slapos_crm_check_partially_allocated_instance.activeSense()
self.tic()
finally:
self._dropHostingSubscription_checkSofwareInstanceState()
self.assertEqual('Visited by HostingSubscription_checkSofwareInstanceAllocationState',
host_sub.workflow_history['edit_workflow'][-1]['comment'])
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
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7