Skip to content

S
slapos.core
Commit bebf7b6a authored by Rafael Monnerat's avatar Rafael Monnerat
Browse files
Options

slapos_crm: Refactor, clean up and optimize Support Request Generation for Monitoring

parent 339fba9c
Show whitespace changes
Inline Side-by-side
Showing with 1632 additions and 1588 deletions
master/bt5/slapos_crm/PathTemplateItem/portal_alarms/slapos_check_computer_state.xml → master/bt5/slapos_crm/PathTemplateItem/portal_alarms/slapos_crm_check_computer_state.xml
View file @ bebf7b6a
......@@ -12,9 +12,7 @@
</item>
<item>
<key> <string>description</string> </key>
<value>
<none/>
</value>
<value> <string>Check if a public or a friend computer contacted master recently and create a ticket if the computer stops to contact master after some time.</string> </value>
</item>
<item>
<key> <string>enabled</string> </key>
......@@ -22,7 +20,7 @@
</item>
<item>
<key> <string>id</string> </key>
<value> <string>slapos_check_computer_state</string> </value>
<value> <string>slapos_crm_check_computer_state</string> </value>
</item>
<item>
<key> <string>periodicity_hour</string> </key>
......
master/bt5/slapos_crm/PathTemplateItem/portal_alarms/slapos_crm_check_partially_allocated_instance.xml → master/bt5/slapos_crm/PathTemplateItem/portal_alarms/slapos_crm_check_instance_in_error.xml
View file @ bebf7b6a
......@@ -8,7 +8,7 @@
<dictionary>
<item>
<key> <string>active_sense_method_id</string> </key>
<value> <string>Alarm_findAndNofitiyUnallocatedSoftwareInstance</string> </value>
<value> <string>Alarm_checkHostingSubscriptionState</string> </value>
</item>
<item>
<key> <string>description</string> </key>
......@@ -20,7 +20,7 @@
</item>
<item>
<key> <string>id</string> </key>
<value> <string>slapos_crm_check_partially_allocated_instance</string> </value>
<value> <string>slapos_crm_check_instance_in_error</string> </value>
</item>
<item>
<key> <string>periodicity_hour</string> </key>
......
master/bt5/slapos_crm/PathTemplateItem/portal_alarms/slapos_check_software_instance_state.xml → master/bt5/slapos_crm/PathTemplateItem/portal_alarms/slapos_crm_check_update_personal_allocation_scope.xml
View file @ bebf7b6a
......@@ -8,11 +8,11 @@
<dictionary>
<item>
<key> <string>active_sense_method_id</string> </key>
<value> <string>Alarm_checkSoftwareInstanceState</string> </value>
<value> <string>Alarm_checkAndUpdatePersonalComputerAllocationScope</string> </value>
</item>
<item>
<key> <string>description</string> </key>
<value> <string>Vefity the state of the instances which are hosted on public servers.</string> </value>
<value> <string>Automatically turn allocation scope of personal computer to closed if the computer were never used.</string> </value>
</item>
<item>
<key> <string>enabled</string> </key>
......@@ -20,7 +20,7 @@
</item>
<item>
<key> <string>id</string> </key>
<value> <string>slapos_check_software_instance_state</string> </value>
<value> <string>slapos_crm_check_update_personal_allocation_scope</string> </value>
</item>
<item>
<key> <string>periodicity_hour</string> </key>
......@@ -28,21 +28,17 @@
<tuple/>
</value>
</item>
<item>
<key> <string>periodicity_hour_frequency</string> </key>
<value> <int>1</int> </value>
</item>
<item>
<key> <string>periodicity_minute</string> </key>
<value>
<tuple/>
<tuple>
<int>5</int>
</tuple>
</value>
</item>
<item>
<key> <string>periodicity_minute_frequency</string> </key>
<value>
<none/>
</value>
<value> <int>1</int> </value>
</item>
<item>
<key> <string>periodicity_month</string> </key>
......@@ -68,7 +64,7 @@
</tuple>
<state>
<tuple>
<float>1377608640.0</float>
<float>1399420800.0</float>
<string>GMT</string>
</tuple>
</state>
......@@ -93,7 +89,7 @@
</item>
<item>
<key> <string>title</string> </key>
<value> <string>Check instances\' state</string> </value>
<value> <string>Check and Update Personal Computer Allocation Scope</string> </value>
</item>
</dictionary>
</pickle>
......
master/bt5/slapos_crm/SkinTemplateItem/portal_skins/slapos_crm_monitoring/Alarm_checkAndUpdateComputerAllocationScope.xml
View file @ bebf7b6a
......@@ -50,13 +50,10 @@
</item>
<item>
<key> <string>_body</string> </key>
<value> <string encoding="cdata"><![CDATA[
portal = context.getPortalObject()\n
<value> <string>portal = context.getPortalObject()\n
\n
category_public = portal.restrictedTraverse("portal_categories/allocation_scope/open/public", None)\n
category_friend = portal.restrictedTraverse("portal_categories/allocation_scope/open/friend", None)\n
category_personal = portal.restrictedTraverse("portal_categories/allocation_scope/open/personal", None)\n
\n
if category_public is not None:\n
portal.portal_catalog.searchAndActivate(\n
......@@ -67,22 +64,8 @@ if category_public is not None:\n
activate_kw={\'tag\': tag}\n
)\n
\n
if category_personal is not None:\n
portal.portal_catalog.searchAndActivate(\n
portal_type=\'Computer\', \n
validation_state=\'validated\', \n
modification_date=(DateTime() - 30).strftime(\'<=%Y/%m/%d\'), \n
order_by=((\'modification_date\', "ASC"), ), \n
default_allocation_scope_uid=category_personal.getUid(), \n
left_join_list=[\'aggregate_related_uid\'], \n
aggregate_related_uid=None,\n
method_id=\'Computer_checkAndUpdatePersonalAllocationScope\',\n
activate_kw={\'tag\': tag})\n
\n
context.activate(after_tag=tag).getId()\n
]]></string> </value>
</string> </value>
</item>
<item>
<key> <string>_params</string> </key>
......
master/bt5/slapos_crm/SkinTemplateItem/portal_skins/slapos_crm_monitoring/Alarm_findAndNofitiyUnallocatedSoftwareInstance.xml → master/bt5/slapos_crm/SkinTemplateItem/portal_skins/slapos_crm_monitoring/Alarm_checkAndUpdatePersonalComputerAllocationScope.xml
View file @ bebf7b6a
......@@ -54,32 +54,19 @@
portal = context.getPortalObject()\n
\n
#Get all unallocated Software Instance\n
select_dict= {\'default_aggregate_uid\': None}\n
software_instance_list = context.portal_catalog(\n
portal_type=["Software Instance", "Slave Instance"],\n
validation_state="validated",\n
# Do not fetch destroyed instances\n
# XXX slap_state=["start_requested", "stop_requested"],\n
default_aggregate_uid=None,\n
select_dict=select_dict,\n
left_join_list=select_dict.keys(),\n
group_by="specialise_uid"\n
)\n
category_personal = portal.restrictedTraverse("portal_categories/allocation_scope/open/personal", None)\n
\n
# Get the list of concerned Hosting Subscription reference\n
hs_reference_list = [si.getSpecialiseReference() for si in software_instance_list \n
if si.getSlapState() in [\'start_requested\', \'stop_requested\']\n
]\n
\n
if len(hs_reference_list) > 0:\n
if category_personal is not None:\n
portal.portal_catalog.searchAndActivate(\n
portal_type=\'Hosting Subscription\',\n
validation_state=\'validated\',\n
reference=hs_reference_list,\n
method_id=\'HostingSubscription_checkSofwareInstanceAllocationState\',\n
activate_kw = {\'tag\':tag}\n
)\n
portal_type=\'Computer\', \n
validation_state=\'validated\', \n
modification_date=(DateTime() - 30).strftime(\'<=%Y/%m/%d\'), \n
order_by=((\'modification_date\', "ASC"), ), \n
default_allocation_scope_uid=category_personal.getUid(), \n
left_join_list=[\'aggregate_related_uid\'], \n
aggregate_related_uid=None,\n
method_id=\'Computer_checkAndUpdatePersonalAllocationScope\',\n
activate_kw={\'tag\': tag})\n
\n
context.activate(after_tag=tag).getId()\n
......@@ -92,7 +79,7 @@ context.activate(after_tag=tag).getId()\n
</item>
<item>
<key> <string>id</string> </key>
<value> <string>Alarm_findAndNofitiyUnallocatedSoftwareInstance</string> </value>
<value> <string>Alarm_checkAndUpdatePersonalComputerAllocationScope</string> </value>
</item>
</dictionary>
</pickle>
......
master/bt5/slapos_crm/SkinTemplateItem/portal_skins/slapos_crm_monitoring/Alarm_checkSoftwareInstanceState.xml → master/bt5/slapos_crm/SkinTemplateItem/portal_skins/slapos_crm_monitoring/Alarm_checkHostingSubscriptionState.xml
View file @ bebf7b6a
......@@ -51,16 +51,12 @@
<item>
<key> <string>_body</string> </key>
<value> <string>portal = context.getPortalObject()\n
category_uid = portal.restrictedTraverse(\n
"portal_categories/allocation_scope/open/public", None).getUid()\n
\n
if category_uid is not None:\n
portal.portal_catalog.searchAndActivate(\n
portal_type = \'Computer\',\n
validation_state = \'validated\',\n
default_allocation_scope_uid = category_uid,\n
method_id = \'Computer_checkSoftwareInstanceState\',\n
activate_kw = {\'tag\':tag} \n
portal.portal_catalog.searchAndActivate(\n
portal_type=\'Hosting Subscription\',\n
validation_state=\'validated\',\n
method_id=\'HostingSubscription_checkSofwareInstanceState\',\n
activate_kw = {\'tag\':tag}\n
)\n
\n
context.activate(after_tag=tag).getId()\n
......@@ -72,7 +68,7 @@ context.activate(after_tag=tag).getId()\n
</item>
<item>
<key> <string>id</string> </key>
<value> <string>Alarm_checkSoftwareInstanceState</string> </value>
<value> <string>Alarm_checkHostingSubscriptionState</string> </value>
</item>
</dictionary>
</pickle>
......
master/bt5/slapos_crm/SkinTemplateItem/portal_skins/slapos_crm_monitoring/Base_generateSupportRequestForSlapOS.xml
View file @ bebf7b6a
......@@ -61,6 +61,8 @@ elif source_project_value.getPortalType() == "Software Instance":\n
destination_decision = source_project_value.getSpecialiseValue().getDestinationSection()\n
elif source_project_value.getPortalType() == "Hosting Subscription":\n
destination_decision = source_project_value.getDestinationSection()\n
elif source_project_value.getPortalType() == "Software Installation":\n
destination_decision = source_project_value.getDestinationSection()\n
else:\n
destination_decision = None\n
\n
......@@ -68,33 +70,41 @@ if portal.ERP5Site_isSupportRequestCreationClosed(destination_decision):\n
# Stop ticket creation\n
return\n
\n
if not title.startswith(\'[MONITORING]\'):\n
title = \'[MONITORING] \' + title\n
support_request_in_progress = portal.portal_catalog.getResultValue(\n
portal_type = \'Support Request\',\n
title = title,\n
simulation_state = ["validated","submitted","suspended"],\n
source_project_uid = source_project_value.getUid()\n
simulation_state = ["validated", "submitted", "suspended"],\n
source_project_uid = source_project_value.getUid(),\n
)\n
\n
if support_request_in_progress is None:\n
ressource = portal.service_module.\\\n
if support_request_in_progress is not None:\n
return support_request_in_progress\n
\n
support_request_in_progress = context.REQUEST.get("support_request_in_progress", None)\n
\n
if support_request_in_progress is not None:\n
return portal.restrictedTraverse(support_request_in_progress)\n
\n
resource = portal.service_module.\\\n
slapos_crm_monitoring.getRelativeUrl()\n
support_request = portal.\\\n
\n
support_request = portal.\\\n
support_request_module.\\\n
slapos_crm_support_request_template_for_monitoring.\\\n
Base_createCloneDocument(batch_mode=1)\n
support_request.edit(\n
support_request.edit(\n
title = title,\n
description = description,\n
start_date = DateTime(),\n
destination_decision=destination_decision,\n
source_project_value = source_relative_url,\n
ressource=ressource\n
source_project_value = source_project_value,\n
resource=resource\n
)\n
support_request.validate()\n
support_request.validate()\n
\n
context.REQUEST.set("support_request_in_progress", support_request.getRelativeUrl())\n
\n
return support_request.getRelativeUrl()\n
return support_request\n
</string> </value>
</item>
<item>
......
master/bt5/slapos_crm/SkinTemplateItem/portal_skins/slapos_crm_monitoring/Computer_checkAndUpdateAllocationScope.xml
View file @ bebf7b6a
......@@ -74,33 +74,21 @@ if not person.Person_isServiceProvider():\n
}\n
\n
# Create a ticket (or re-open it) for this issue!\n
support_request = None\n
request_title = \'We have changed allocation scope for %s\' % computer_reference\n
request_description = \'Allocation scope has been changed to \' \\\n
\'%s for %s\' % (target_allocation_scope, computer_reference)\n
\n
support_request_url = context.Base_generateSupportRequestForSlapOS(\n
support_request = context.Base_generateSupportRequestForSlapOS(\n
request_title,\n
request_description,\n
computer.getRelativeUrl()\n
)\n
if support_request_url:\n
support_request = portal.restrictedTraverse(support_request_url)\n
support_request.suspend()\n
else:\n
# XXX - Base_generateSupportRequestForSlapOS return None if the \n
# support_request already exist, but we want to use it again so...\n
support_request = portal.portal_catalog.getResultValue(\n
portal_type = \'Support Request\',\n
title = request_title,\n
simulation_state = [\'suspended\', \'open\'],\n
source_project_uid = computer.getUid()\n
)\n
if support_request is None:\n
# Existing ticket not found, can not create event for the moment\n
return\n
\n
if support_request.getSimulationState() != "validated":\n
support_request.validate()\n
\n
# Send notification message\n
message = request_description\n
notification_message = portal.portal_notifications.getDocumentValue(\n
reference=notification_message_reference)\n
\n
......@@ -108,16 +96,17 @@ if not person.Person_isServiceProvider():\n
mapping_dict = {\'computer_title\':computer.getTitle(),\n
\'computer_id\':computer_reference,\n
\'allocation_scope\':allocation_scope}\n
\n
message = notification_message.asText(\n
substitution_method_parameter_dict={\'mapping_dict\':mapping_dict})\n
else:\n
message = request_description\n
substitution_method_parameter_dict={\'mapping_dict\': mapping_dict})\n
\n
event = support_request.SupportRequest_trySendNotificationMessage(request_title,\n
message, person.getRelativeUrl())\n
event = support_request.SupportRequest_trySendNotificationMessage(\n
request_title, message, person.getRelativeUrl())\n
\n
if event is not None:\n
computer.edit(**edit_kw)\n
\n
return support_request\n
</string> </value>
</item>
<item>
......
master/bt5/slapos_crm/SkinTemplateItem/portal_skins/slapos_crm_monitoring/Computer_checkAndUpdatePersonalAllocationScope.xml
View file @ bebf7b6a
......@@ -50,7 +50,7 @@
</item>
<item>
<key> <string>_body</string> </key>
<value> <string>context.Computer_checkAndUpdateAllocationScope(\n
<value> <string>return context.Computer_checkAndUpdateAllocationScope(\n
target_allocation_scope = \'close/termination\',\n
notification_message_reference=\'slapos-crm-computer_personal_allocation_scope.notification\')\n
</string> </value>
......
master/bt5/slapos_crm/SkinTemplateItem/portal_skins/slapos_crm_monitoring/Computer_checkState.xml
View file @ bebf7b6a
......@@ -76,8 +76,8 @@ try:\n
d = memcached_dict[reference]\n
d = json.loads(d)\n
last_contact = DateTime(d.get(\'created_at\'))\n
if (DateTime() - last_contact) > 1:\n
description = "The Computer %s (%s) has not contacted the server for more than 24 hours" \\\n
if (DateTime() - last_contact) > 0.01:\n
description = "The Computer %s (%s) has not contacted the server for more than 30 minutes" \\\n
"(last contact date: %s)" % (computer_title, reference, last_contact)\n
else:\n
should_notify = False\n
......@@ -85,58 +85,37 @@ except KeyError:\n
ticket_title = "[MONITORING] No information about %s" % reference\n
description = "The Computer %s (%s) has not contacted the server (No Contact Information)" % (\n
computer_title, reference)\n
\n
\n
if should_notify:\n
support_request_url = context.Base_generateSupportRequestForSlapOS(\n
support_request = context.Base_generateSupportRequestForSlapOS(\n
ticket_title,\n
description,\n
context.getRelativeUrl()\n
)\n
\n
support_request = None\n
\n
person = context.getSourceAdministrationValue(portal_type="Person")\n
if not person:\n
return support_request_url\n
\n
if support_request_url:\n
support_request = portal.restrictedTraverse(support_request_url, None)\n
else:\n
# XXX - the support request might already exists\n
support_request = portal.portal_catalog.getResultValue(\n
portal_type = \'Support Request\',\n
title = ticket_title,\n
simulation_state = \'validated\',\n
source_project_uid = context.getUid()\n
)\n
if support_request is None:\n
# Existing ticket not found, can not create event for the moment\n
return support_request_url\n
\n
return support_request\n
\n
# Send Notification message\n
notification_reference = \'slapos-crm-computer_check_state.notification\'\n
notification_message = portal.portal_notifications.getDocumentValue(\n
reference=notification_reference)\n
if notification_message is None:\n
message = """Dear user,\n
%s.\n
Do not hesitate to visit the web forum (http://community.slapos.org/forum) in case of question.\n
\n
Regards,\n
\n
The slapos team""" % description\n
if notification_message is None:\n
message = """%s""" % description\n
else:\n
mapping_dict = {\'computer_title\':context.getTitle(),\n
\'computer_id\':reference,\n
\'last_contact\':last_contact}\n
message = notification_message.asText(\n
substitution_method_parameter_dict={\'mapping_dict\':mapping_dict})\n
\n
\n
support_request.SupportRequest_trySendNotificationMessage(\n
ticket_title.replace(\'[MONITORING] \', \'\'),\n
message, person.getRelativeUrl(), 5)\n
ticket_title,\n
message, person.getRelativeUrl())\n
\n
return support_request_url\n
return support_request\n
]]></string> </value>
......
master/bt5/slapos_crm/SkinTemplateItem/portal_skins/slapos_crm_monitoring/ERP5Site_isSupportRequestCreationClosed.xml
View file @ bebf7b6a
......@@ -50,7 +50,9 @@
</item>
<item>
<key> <string>_body</string> </key>
<value> <string># HARDCODED LIMIT TO BE MOVED TO GLOBAL PREFERENCES\n
<value> <string encoding="cdata"><![CDATA[
# HARDCODED LIMIT TO BE MOVED TO GLOBAL PREFERENCES\n
limit = 5\n
\n
kw[\'limit\'] = limit\n
......@@ -62,8 +64,10 @@ if destination_decision:\n
\n
support_request_list = context.portal_catalog(**kw)\n
\n
return len(support_request_list) == limit\n
</string> </value>
return len(support_request_list) >= limit\n
]]></string> </value>
</item>
<item>
<key> <string>_params</string> </key>
......
master/bt5/slapos_crm/SkinTemplateItem/portal_skins/slapos_crm_monitoring/HostingSubscription_checkSofwareInstanceAllocationState.xml → master/bt5/slapos_crm/SkinTemplateItem/portal_skins/slapos_crm_monitoring/HostingSubscription_checkSoftwareInstanceState.xml
View file @ bebf7b6a
......@@ -62,38 +62,45 @@ if portal.ERP5Site_isSupportRequestCreationClosed():\n
# Stop ticket creation\n
return\n
\n
hs_title = hosting_subscription.getTitle()\n
date_check_limit = addToDate(DateTime(), to_add={\'hour\': -4})\n
date_check_limit = addToDate(DateTime(), to_add={\'hour\': -1})\n
\n
if (date_check_limit - hosting_subscription.getCreationDate()) < 0:\n
# Too early to check\n
return\n
\n
#if not source_instance:\n
# return\n
\n
software_instance_list = hosting_subscription.getSpecialiseRelatedValueList(\n
portal_type=["Software Instance", "Slave Instance"])\n
\n
has_newest_allocated_instance = False\n
has_unallocated_instance = False\n
source_instance = hosting_subscription.getPredecessorValue()\n
failing_instance = None\n
\n
if not source_instance:\n
return\n
software_instance_list = [source_instance]\n
predecessors_instance = source_instance.getPredecessorValueList()\n
if len(predecessors_instance) > 0:\n
software_instance_list.extend(predecessors_instance)\n
\n
# Check if at least one software Instance is Allocated\n
for instance in software_instance_list:\n
if instance.getAggregate("") != "":\n
if (instance.getCreationDate() - date_check_limit) < 0:\n
# This instance is created for more than 4 hours\n
if instance.getSlapState() not in ["start_requested", "stop_requested"]:\n
continue\n
\n
if (date_check_limit - instance.getCreationDate()) < 0:\n
continue\n
\n
if instance.getAggregateValue() is not None:\n
has_newest_allocated_instance = True\n
if instance.getPortalType() == "Software Instance" and \\\n
instance.SoftwareInstance_hasReportedError():\n
return context.HostingSubscription_createSupportRequestEvent(\n
instance, \'slapos-crm-hosting-subscription-instance-state.notification\')\n
else:\n
has_unallocated_instance = True\n
failing_instance = instance\n
\n
if has_unallocated_instance and has_newest_allocated_instance:\n
return context.Base_generateSupportRequestForSlapOS(\n
"Hosting Subscription %s is partially allocated" % hs_title,\n
"%s has allocated instance(s) but, the instance %s (%s) has been unallocated for more than 4 hours." % (\n
hs_title, failing_instance.getTitle(), failing_instance.getAbsoluteUrl()),\n
hosting_subscription.getRelativeUrl())\n
else:\n
return\n
if has_unallocated_instance and has_newest_allocated_instance:\n
return context.HostingSubscription_createSupportRequestEvent(\n
failing_instance, \'slapos-crm-hosting-subscription-instance-allocation.notification\')\n
\n
return\n
]]></string> </value>
......@@ -104,7 +111,7 @@ else:\n
</item>
<item>
<key> <string>id</string> </key>
<value> <string>HostingSubscription_checkSofwareInstanceAllocationState</string> </value>
<value> <string>HostingSubscription_checkSoftwareInstanceState</string> </value>
</item>
</dictionary>
</pickle>
......
master/bt5/slapos_crm/SkinTemplateItem/portal_skins/slapos_crm_monitoring/HostingSubscription_createSupportRequestEvent.xml 0 → 100644
View file @ bebf7b6a
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PythonScript" module="Products.PythonScripts.PythonScript"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>Script_magic</string> </key>
<value> <int>3</int> </value>
</item>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
</klass>
<tuple/>
<state>
<dictionary>
<item>
<key> <string>_asgns</string> </key>
<value>
<dictionary>
<item>
<key> <string>name_container</string> </key>
<value> <string>container</string> </value>
</item>
<item>
<key> <string>name_context</string> </key>
<value> <string>context</string> </value>
</item>
<item>
<key> <string>name_m_self</string> </key>
<value> <string>script</string> </value>
</item>
<item>
<key> <string>name_subpath</string> </key>
<value> <string>traverse_subpath</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</state>
</object>
</value>
</item>
<item>
<key> <string>_body</string> </key>
<value> <string>portal = context.getPortalObject()\n
\n
ticket_title = "Hosting Subscription %s is failing." % context.getTitle()\n
\n
description = "%s contains software instances which are unallocated or reporting errors." % (\n
context.getTitle())\n
\n
support_request = context.Base_generateSupportRequestForSlapOS(\n
ticket_title,\n
description,\n
context.getRelativeUrl())\n
\n
if support_request is None:\n
return\n
\n
person = context.getDestinationSectionValue(portal_type="Person")\n
if not person:\n
return\n
\n
if support_request.getSimulationState() != "validated":\n
support_request.validate()\n
\n
# Send Notification message\n
message = description\n
\n
notification_reference = notification_message_reference\n
notification_message = portal.portal_notifications.getDocumentValue(\n
reference=notification_reference)\n
if notification_message is not None:\n
mapping_dict = {\'hosting_subscription_title\':context.getTitle(),\n
\'instance\': instance.getTitle()}\n
\n
message = notification_message.asText(\n
substitution_method_parameter_dict={\'mapping_dict\':mapping_dict})\n
\n
return support_request.SupportRequest_trySendNotificationMessage(\n
ticket_title, message, person.getRelativeUrl())\n
</string> </value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string>instance, notification_message_reference</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>HostingSubscription_createSupportRequestEvent</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
master/bt5/slapos_crm/SkinTemplateItem/portal_skins/slapos_crm_monitoring/Person_isServiceProvider.xml
View file @ bebf7b6a
......@@ -54,7 +54,8 @@
return True\n
\n
for assignment in context.contentValues(portal_type="Assignment"):\n
if assignment.getRole() == \'service_provider\':\n
if assignment.getValidationState() == "open" and \\\n
assignment.getRole() == \'service_provider\':\n
return True\n
\n
return False\n
......
master/bt5/slapos_crm/SkinTemplateItem/portal_skins/slapos_crm_monitoring/HostingSubscription_CheckInstanceState.xml → master/bt5/slapos_crm/SkinTemplateItem/portal_skins/slapos_crm_monitoring/SoftwareInstallation_hasReportedError.xml
View file @ bebf7b6a
......@@ -50,20 +50,28 @@
</item>
<item>
<key> <string>_body</string> </key>
<value> <string>hosting_subscription = context\n
<value> <string>from DateTime import DateTime\n
import json\n
\n
instance = hosting_subscription.getPredecessorValue()\n
if instance is None:\n
return\n
instance_list = [instance]\n
instance_list.extend(instance.getPredecessorValueList())\n
memcached_dict = context.getPortalObject().portal_memcached.getMemcachedDict(\n
key_prefix=\'slap_tool\',\n
plugin_path=\'portal_memcached/default_memcached_plugin\')\n
\n
for sub_instance in instance_list:\n
if sub_instance and \\\n
sub_instance.activate().SoftwareInstance_checkState():\n
# This instance is in failing state\n
# One notification per hosting subscription\n
break\n
try:\n
d = memcached_dict[context.getReference()]\n
except KeyError:\n
# Information not available\n
return None\n
\n
d = json.loads(d)\n
result = d[\'text\']\n
last_contact = DateTime(d.get(\'created_at\'))\n
\n
# Optimise by checking memcache information first.\n
if result.startswith(\'#error \'):\n
return last_contact\n
\n
return None\n
</string> </value>
</item>
<item>
......@@ -72,7 +80,7 @@ for sub_instance in instance_list:\n
</item>
<item>
<key> <string>id</string> </key>
<value> <string>HostingSubscription_CheckInstanceState</string> </value>
<value> <string>SoftwareInstallation_hasReportedError</string> </value>
</item>
</dictionary>
</pickle>
......
master/bt5/slapos_crm/SkinTemplateItem/portal_skins/slapos_crm_monitoring/SoftwareInstance_checkState.xml deleted 100644 → 0
View file @ 339fba9c
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PythonScript" module="Products.PythonScripts.PythonScript"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>Script_magic</string> </key>
<value> <int>3</int> </value>
</item>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
</klass>
<tuple/>
<state>
<dictionary>
<item>
<key> <string>_asgns</string> </key>
<value>
<dictionary>
<item>
<key> <string>name_container</string> </key>
<value> <string>container</string> </value>
</item>
<item>
<key> <string>name_context</string> </key>
<value> <string>context</string> </value>
</item>
<item>
<key> <string>name_m_self</string> </key>
<value> <string>script</string> </value>
</item>
<item>
<key> <string>name_subpath</string> </key>
<value> <string>traverse_subpath</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</state>
</object>
</value>
</item>
<item>
<key> <string>_body</string> </key>
<value> <string encoding="cdata"><![CDATA[
from DateTime import DateTime\n
import json\n
\n
portal = context.getPortalObject()\n
\n
if portal.ERP5Site_isSupportRequestCreationClosed():\n
# Stop ticket creation\n
return\n
\n
if (DateTime() - context.getCreationDate()) < 2:\n
# Ignore recently created instances.\n
return\n
\n
reference = context.getReference()\n
memcached_dict = context.getPortalObject().portal_memcached.getMemcachedDict(\n
key_prefix=\'slap_tool\',\n
plugin_path=\'portal_memcached/default_memcached_plugin\')\n
\n
try:\n
d = memcached_dict[reference]\n
except KeyError:\n
return\n
\n
d = json.loads(d)\n
result = d[\'text\']\n
last_contact = DateTime(d.get(\'created_at\'))\n
\n
# Optimise by checking memcache information first.\n
if result.startswith(\'#error \'):\n
service = context.getSpecialiseValue(portal_type=\'Hosting Subscription\')\n
hosting_title = service.getTitle()\n
ticket_title = "Service %s in error state" % service.getReference()\n
description = "The instance %s of service %s has been in error state (last contact date: %s)" % (\n
reference, hosting_title, last_contact)\n
support_request_url = context.Base_generateSupportRequestForSlapOS(\n
ticket_title, description, service.getRelativeUrl())\n
\n
if not support_request_url:\n
return\n
\n
support_request = portal.restrictedTraverse(support_request_url, None)\n
person_url = service.getDestinationSection()\n
\n
if support_request is None or not person_url:\n
return support_request_url\n
\n
notification_message = portal.portal_notifications.getDocumentValue(\n
reference=\'slapos-crm-hosting_subscription_state.notification\')\n
if notification_message is None:\n
message = """Dear user,\n
%s.\n
Do not hesitate to visit the web forum (http://community.slapos.org/forum) in case of question.\n
\n
Regards,\n
\n
The slapos team""" % description\n
else:\n
mapping_dict = {\'hosting_title\':hosting_title,\n
\'hosting_url\':service.getRelativeUrl(),\n
\'instance_title\':context.getTitle(),\n
\'last_contact\':last_contact}\n
message = notification_message.asText(\n
substitution_method_parameter_dict={\'mapping_dict\':mapping_dict})\n
\n
support_request.SupportRequest_trySendNotificationMessage(\n
ticket_title, message, person_url, 0)\n
\n
return support_request_url\n
]]></string> </value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string></string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>SoftwareInstance_checkState</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
master/bt5/slapos_crm/SkinTemplateItem/portal_skins/slapos_crm_monitoring/Computer_checkSoftwareInstanceState.xml → master/bt5/slapos_crm/SkinTemplateItem/portal_skins/slapos_crm_monitoring/SoftwareInstance_hasReportedError.xml
View file @ bebf7b6a
......@@ -50,25 +50,30 @@
</item>
<item>
<key> <string>_body</string> </key>
<value> <string>portal = context.getPortalObject()\n
<value> <string>from DateTime import DateTime\n
import json\n
\n
partition_list = portal.portal_catalog(portal_type = \'Computer Partition\',\n
free_for_request = 0,\n
parent_uid = context.getUid()\n
)\n
memcached_dict = context.getPortalObject().portal_memcached.getMemcachedDict(\n
key_prefix=\'slap_tool\',\n
plugin_path=\'portal_memcached/default_memcached_plugin\')\n
\n
hosting_subscription_list = []\n
for partition in partition_list:\n
software_instance = partition.getAggregateRelatedValue(\n
portal_type=\'Software Instance\')\n
if software_instance:\n
hosting = software_instance.getSpecialiseValue(\n
portal_type=\'Hosting Subscription\')\n
if hosting and not hosting in hosting_subscription_list:\n
hosting_subscription_list.append(hosting)\n
hosting.HostingSubscription_CheckInstanceState()\n
\n
return len(hosting_subscription_list)\n
if context.getAggregateValue(portal_type="Computer Partition") is not None:\n
try:\n
d = memcached_dict[context.getReference()]\n
except KeyError:\n
return \n
\n
d = json.loads(d)\n
result = d[\'text\']\n
last_contact = DateTime(d.get(\'created_at\'))\n
\n
# Optimise by checking memcache information first.\n
if result.startswith(\'#error \'):\n
return last_contact\n
\n
# XXX time limit of 48 hours for run at least once.\n
\n
return None\n
</string> </value>
</item>
<item>
......@@ -77,7 +82,7 @@ return len(hosting_subscription_list)\n
</item>
<item>
<key> <string>id</string> </key>
<value> <string>Computer_checkSoftwareInstanceState</string> </value>
<value> <string>SoftwareInstance_hasReportedError</string> </value>
</item>
</dictionary>
</pickle>
......
master/bt5/slapos_crm/SkinTemplateItem/portal_skins/slapos_crm_monitoring/SupportRequest_trySendNotificationMessage.xml
View file @ bebf7b6a
......@@ -50,23 +50,23 @@
</item>
<item>
<key> <string>_body</string> </key>
<value> <string encoding="cdata"><![CDATA[
support_request = context\n
<value> <string>support_request = context\n
portal = context.getPortalObject()\n
\n
resource = portal.service_module.slapos_crm_information.getRelativeUrl()\n
# create Web message if needed for this ticket\n
last_event = context.portal_catalog.getResultValue(\n
title=message_title,\n
follow_up_uid=support_request.getUid(), \n
sort_on=[(\'delivery.start_date\', \'DESC\')],\n
)\n
if last_event and (interval_of_day > 0) and \\\n
(DateTime() - last_event.getStartDate() < interval_of_day):\n
# User has already been notified this last 24h.\n
return\n
if last_event:\n
# User has already been notified for this problem.\n
return last_event\n
\n
event = portal.event_module.slapos_crm_web_message_template.\\\n
Base_createCloneDocument(batch_mode=1)\n
\n
event.edit(\n
title=message_title,\n
text_content=message,\n
......@@ -78,14 +78,14 @@ event.edit(\n
event.stop()\n
event.deliver()\n
\n
event.immediateReindexObject()\n
\n
return event\n
]]></string> </value>
</string> </value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string>message_title, message, source_relative_url, interval_of_day=1</string> </value>
<value> <string>message_title, message, source_relative_url</string> </value>
</item>
<item>
<key> <string>id</string> </key>
......
master/bt5/slapos_crm/TestTemplateItem/portal_components/test.erp5.testSlapOSCRMAlarm.py
View file @ bebf7b6a
......@@ -2,6 +2,7 @@
import transaction
from Products.SlapOS.tests.testSlapOSMixin import \
testSlapOSMixin
from DateTime import DateTime
from Products.ERP5Type.tests.utils import createZODBPythonScript
class TestSlapOSCRMCreateRegularisationRequest(testSlapOSMixin):
......@@ -756,3 +757,341 @@ portal_workflow.doActionFor(context, action='edit_action', comment='Visited by R
self.assertNotEqual(
'Visited by RegularisationRequest_deleteHostingSubscriptionList',
ticket.workflow_history['edit_workflow'][-1]['comment'])
class TestSlapOSCrmMonitoringCheckComputerState(testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
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_check_public_computer_state(self):
self._makeComputer()
self.computer.edit(allocation_scope='open/public')
self._simulateComputer_checkState()
try:
self.portal.portal_alarms.slapos_crm_check_computer_state.activeSense()
self.tic()
finally:
self._dropComputer_checkState()
self.assertEqual('Visited by Computer_checkState',
self.computer.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_check_friend_computer_state(self):
self._makeComputer()
self.computer.edit(allocation_scope='open/friend')
self._simulateComputer_checkState()
try:
self.portal.portal_alarms.slapos_crm_check_computer_state.activeSense()
self.tic()
finally:
self._dropComputer_checkState()
self.assertEqual('Visited by Computer_checkState',
self.computer.workflow_history['edit_workflow'][-1]['comment'])
def _test_alarm_check_computer_state_not_selected(self, allocation_scope):
self._makeComputer()
self.computer.edit(allocation_scope=allocation_scope)
self._simulateComputer_checkState()
try:
self.portal.portal_alarms.slapos_crm_check_computer_state.activeSense()
self.tic()
finally:
self._dropComputer_checkState()
self.assertNotEqual('Visited by Computer_checkState',
self.computer.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_check_computer_state_no_public_computer(self):
self._test_alarm_check_computer_state_not_selected(
allocation_scope='open/personal')
def test_alarm_check_computer_state_closed_forever_computer(self):
self._test_alarm_check_computer_state_not_selected(
allocation_scope='closed/forever')
def test_alarm_check_computer_state_closed_mantainence_computer(self):
self._test_alarm_check_computer_state_not_selected(
allocation_scope='closed/maintenance')
def test_alarm_check_computer_state_closed_termination_computer(self):
self._test_alarm_check_computer_state_not_selected(
allocation_scope='closed/termination')
class TestSlapOSCrmMonitoringCheckComputerAllocationScope(testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def _makeSoftwareInstallation(self):
software_installation = self.portal\
.software_installation_module.template_software_installation\
.Base_createCloneDocument(batch_mode=1)
software_installation.edit(
url_string=self.generateNewSoftwareReleaseUrl(),
aggregate=self.computer.getRelativeUrl(),
reference='TESTSOFTINSTS-%s' % self.generateNewId(),
title='Start requested for %s' % self.computer.getUid()
)
software_installation.validate()
software_installation.requestStart()
return software_installation
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_not_allowed_allocation_scope_OpenPublic(self):
self._makeComputer()
self.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',
self.computer.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_not_allowed_allocation_scope_OpenFriend(self):
self._makeComputer()
self.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',
self.computer.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_not_allowed_allocationScope_open_personal(self):
self._makeComputer()
self.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',
self.computer.workflow_history['edit_workflow'][-1]['comment'])
class TestSlapOSCrmMonitoringCheckComputerPersonalAllocationScope(testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def _makeSoftwareInstallation(self):
software_installation = self.portal\
.software_installation_module.template_software_installation\
.Base_createCloneDocument(batch_mode=1)
software_installation.edit(
url_string=self.generateNewSoftwareReleaseUrl(),
aggregate=self.computer.getRelativeUrl(),
reference='TESTSOFTINSTS-%s' % self.generateNewId(),
title='Start requested for %s' % self.computer.getUid()
)
software_installation.validate()
software_installation.requestStart()
return software_installation
def _simulateComputer_checkAndUpdatePersonalAllocationScope(self):
script_name = 'Computer_checkAndUpdatePersonalAllocationScope'
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_checkAndUpdatePersonalAllocationScope') """ )
transaction.commit()
def _dropComputer_checkAndUpdatePersonalAllocationScope(self):
script_name = 'Computer_checkAndUpdatePersonalAllocationScope'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
transaction.commit()
def test_alarm_allowed_allocation_scope_OpenPersonal_old_computer(self):
self._makeComputer()
self.computer.edit(allocation_scope = 'open/personal')
def getModificationDate(self):
return DateTime() - 50
from Products.ERP5Type.Base import Base
self._simulateComputer_checkAndUpdatePersonalAllocationScope()
original_get_modification = Base.getModificationDate
Base.getModificationDate = getModificationDate
try:
self.portal.portal_alarms.slapos_crm_check_update_personal_allocation_scope.activeSense()
self.tic()
finally:
Base.getModificationDate = original_get_modification
self._dropComputer_checkAndUpdatePersonalAllocationScope()
self.assertEqual('Visited by Computer_checkAndUpdatePersonalAllocationScope',
self.computer.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_allowed_allocation_scope_OpenPersonalWithSoftwareInstallation(self):
self._makeComputer()
self.computer.edit(allocation_scope = 'open/personal')
self._makeSoftwareInstallation()
def getModificationDate(self):
return DateTime() - 50
from Products.ERP5Type.Base import Base
self._simulateComputer_checkAndUpdatePersonalAllocationScope()
original_get_modification = Base.getModificationDate
Base.getModificationDate = getModificationDate
try:
self.portal.portal_alarms.slapos_crm_check_update_personal_allocation_scope.activeSense()
self.tic()
finally:
Base.getModificationDate = original_get_modification
self._dropComputer_checkAndUpdatePersonalAllocationScope()
self.assertNotEqual('Visited by Computer_checkAndUpdatePersonalAllocationScope',
self.computer.workflow_history['edit_workflow'][-1]['comment'])
class TestSlapOSCrmMonitoringCheckInstanceInError(testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def _makeHostingSubscription(self):
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()
new_id = self.generateNewId()
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):
kw = dict(
software_release=hosting_subscription.getUrlString(),
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 _simulateHostingSubscription_checkSofwareInstanceState(self):
script_name = 'HostingSubscription_checkSofwareInstanceState'
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_checkSofwareInstanceState') """ )
transaction.commit()
def _dropHostingSubscription_checkSofwareInstanceState(self):
script_name = 'HostingSubscription_checkSofwareInstanceState'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
transaction.commit()
def test_alarm_check_instance_in_error_validated_hosting_subscription(self):
host_sub = self._makeHostingSubscription()
self._simulateHostingSubscription_checkSofwareInstanceState()
try:
self.portal.portal_alarms.slapos_crm_check_instance_in_error.activeSense()
self.tic()
finally:
self._dropHostingSubscription_checkSofwareInstanceState()
self.assertEqual('Visited by HostingSubscription_checkSofwareInstanceState',
host_sub.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_check_instance_in_error_archived_hosting_subscription(self):
host_sub = self._makeHostingSubscription()
host_sub.archive()
self._simulateHostingSubscription_checkSofwareInstanceState()
try:
self.portal.portal_alarms.slapos_crm_check_instance_in_error.activeSense()
self.tic()
finally:
self._dropHostingSubscription_checkSofwareInstanceState()
self.assertNotEqual('Visited by HostingSubscription_checkSofwareInstanceState',
host_sub.workflow_history['edit_workflow'][-1]['comment'])
master/bt5/slapos_crm/TestTemplateItem/portal_components/test.erp5.testSlapOSCRMAlarm.xml
View file @ bebf7b6a
......@@ -6,10 +6,22 @@
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>_recorded_property_dict</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent>
</value>
</item>
<item>
<key> <string>default_reference</string> </key>
<value> <string>testSlapOSCRMAlarm</string> </value>
</item>
<item>
<key> <string>description</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>test.erp5.testSlapOSCRMAlarm</string> </value>
......@@ -43,13 +55,28 @@
<item>
<key> <string>workflow_history</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent>
<persistent> <string encoding="base64">AAAAAAAAAAM=</string> </persistent>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="2" aka="AAAAAAAAAAI=">
<pickle>
<global name="PersistentMapping" module="Persistence.mapping"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>data</string> </key>
<value>
<dictionary/>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="3" aka="AAAAAAAAAAM=">
<pickle>
<global name="PersistentMapping" module="Persistence.mapping"/>
</pickle>
......@@ -62,7 +89,7 @@
<item>
<key> <string>component_validation_workflow</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAM=</string> </persistent>
<persistent> <string encoding="base64">AAAAAAAAAAQ=</string> </persistent>
</value>
</item>
</dictionary>
......@@ -71,7 +98,7 @@
</dictionary>
</pickle>
</record>
<record id="3" aka="AAAAAAAAAAM=">
<record id="4" aka="AAAAAAAAAAQ=">
<pickle>
<global name="WorkflowHistoryList" module="Products.ERP5Type.patches.WorkflowTool"/>
</pickle>
......
master/bt5/slapos_crm/TestTemplateItem/portal_components/test.erp5.testSlapOSCRMSkins.py
View file @ bebf7b6a
......@@ -204,7 +204,7 @@ The slapos team
@simulate('Entity_statBalance', '*args, **kwargs', 'return "1"')
def test_addRegularisationRequest_existing_invalidated_ticket(self):
person = self.createPerson()
ticket, event = person.Person_checkToCreateRegularisationRequest()
ticket = person.Person_checkToCreateRegularisationRequest()[0]
ticket.invalidate()
transaction.commit()
self.tic()
......@@ -213,7 +213,6 @@ The slapos team
self.assertNotEquals(event2, None)
def test_addRegularisationRequest_REQUEST_disallowed(self):
date = DateTime()
person = self.createPerson()
self.assertRaises(
Unauthorized,
......@@ -1328,7 +1327,7 @@ class TestSlapOSRegularisationRequest_stopHostingSubscriptionList(
def test_stopHostingSubscriptionList_other_subscription(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
hosting_subscription = self.createHostingSubscription()
self.createHostingSubscription()
ticket.edit(
source_project_value=person,
......@@ -1349,7 +1348,7 @@ class TestSlapOSRegularisationRequest_stopHostingSubscriptionList(
'*args, **kwargs',
'raise NotImplementedError, "Should not have been called"')
def test_stopHostingSubscriptionList_no_person(self):
person = self.createPerson()
self.createPerson()
ticket = self.createRegularisationRequest()
ticket.edit(
......@@ -1372,7 +1371,7 @@ class TestSlapOSRegularisationRequest_stopHostingSubscriptionList(
def test_stopHostingSubscriptionList_not_suspended(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
hosting_subscription = self.createHostingSubscription()
self.createHostingSubscription()
ticket.edit(
source_project_value=person,
......@@ -1394,7 +1393,7 @@ class TestSlapOSRegularisationRequest_stopHostingSubscriptionList(
def test_stopHostingSubscriptionList_other_resource(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
hosting_subscription = self.createHostingSubscription()
self.createHostingSubscription()
ticket.edit(
source_project_value=person,
......@@ -1704,7 +1703,7 @@ class TestSlapOSRegularisationRequest_deleteHostingSubscriptionList(
def test_deleteHostingSubscriptionList_other_subscription(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
hosting_subscription = self.createHostingSubscription()
self.createHostingSubscription()
ticket.edit(
source_project_value=person,
......@@ -1725,7 +1724,7 @@ class TestSlapOSRegularisationRequest_deleteHostingSubscriptionList(
'*args, **kwargs',
'raise NotImplementedError, "Should not have been called"')
def test_deleteHostingSubscriptionList_no_person(self):
person = self.createPerson()
self.createPerson()
ticket = self.createRegularisationRequest()
ticket.edit(
......@@ -1748,7 +1747,7 @@ class TestSlapOSRegularisationRequest_deleteHostingSubscriptionList(
def test_deleteHostingSubscriptionList_not_suspended(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
hosting_subscription = self.createHostingSubscription()
self.createHostingSubscription()
ticket.edit(
source_project_value=person,
......@@ -1770,7 +1769,7 @@ class TestSlapOSRegularisationRequest_deleteHostingSubscriptionList(
def test_deleteHostingSubscriptionList_other_resource(self):
person = self.createPerson()
ticket = self.createRegularisationRequest()
hosting_subscription = self.createHostingSubscription()
self.createHostingSubscription()
ticket.edit(
source_project_value=person,
......@@ -1802,30 +1801,25 @@ class TestSlapOSComputer_notifyWrongAllocationScope(testSlapOSMixin):
def _cancelTestSupportRequestList(self):
for support_request in self.portal.portal_catalog(
portal_type="Support Request",
title="[MONITORING] % TESTCOMPT-%",
title="%%TESTCOMPT-%",
simulation_state=["validated", "suspended"]):
support_request.invalidate()
self.tic()
def _makeComputer(self,new_id):
def _makeComputer(self):
super(TestSlapOSComputer_notifyWrongAllocationScope, self)._makeComputer()
# 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
self.computer.edit(
source_administration_value=self._makePerson()
)
computer.validate()
return computer
return self.computer
def _makePerson(self, new_id):
def _makePerson(self):
# Clone computer document
person = self.portal.person_module.template_member\
.Base_createCloneDocument(batch_mode=1)
person.edit(reference='TESTPERSON-%s' % (new_id, ))
person.edit(reference='TESTPERSON-%s' % (self.generateNewId(), ))
person.immediateReindexObject()
return person
......@@ -1839,7 +1833,7 @@ class TestSlapOSComputer_notifyWrongAllocationScope(testSlapOSMixin):
return assignment
def _getGeneratedSupportRequest(self, computer):
request_title = '[MONITORING] We have changed allocation scope for %s' % \
request_title = '%%We have changed allocation scope for %s' % \
computer.getReference()
support_request = self.portal.portal_catalog.getResultValue(
portal_type = 'Support Request',
......@@ -1860,20 +1854,21 @@ class TestSlapOSComputer_notifyWrongAllocationScope(testSlapOSMixin):
return notification_message.getRelativeUrl()
@simulate('ERP5Site_isSupportRequestCreationClosed', '*args, **kwargs','return 0')
@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',
'message_title, message, source_relative_url',
'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' \
'%s %s %s" % (message_title, message, source_relative_url))\n' \
'return 1')
def test_computerNotAllowedAllocationScope_OpenPublic(self):
computer = self._makeComputer(self.new_id)
computer = self._makeComputer()
person = computer.getSourceAdministrationValue()
self._updatePersonAssignment(person, 'role/member')
......@@ -1881,87 +1876,89 @@ class TestSlapOSComputer_notifyWrongAllocationScope(testSlapOSMixin):
self._makeNotificationMessage(computer.getReference())
computer.edit(allocation_scope='open/public')
computer.Computer_checkAndUpdateAllocationScope()
ticket = computer.Computer_checkAndUpdateAllocationScope()
self.tic()
self.assertEquals(computer.getAllocationScope(), 'open/personal')
ticket = self._getGeneratedSupportRequest(computer)
self.assertEquals(ticket.getSimulationState(), 'suspended')
#ticket = self._getGeneratedSupportRequest(computer)
self.assertNotEquals(None, ticket)
self.assertEquals(ticket.getSimulationState(), 'validated')
self.assertEqual('Visited by SupportRequest_trySendNotificationMessage ' \
'%s %s %s %s' % \
'%s %s %s' % \
('We have changed allocation scope for %s' % computer.getReference(),
'Test NM content\n%s\n' % computer.getReference(), person.getRelativeUrl(), '1'),
'Test NM content\n%s\n' % computer.getReference(), person.getRelativeUrl()),
ticket.workflow_history['edit_workflow'][-1]['comment'])
@simulate('ERP5Site_isSupportRequestCreationClosed', '*args, **kwargs','return 0')
@simulate('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',
'message_title, message, source_relative_url',
'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' \
'%s %s %s" % (message_title, message, source_relative_url))\n' \
'return 1')
def test_computerNotAllowedAllocationScope_OpenFriend(self):
computer = self._makeComputer(self.new_id)
computer = self._makeComputer()
person = computer.getSourceAdministrationValue()
self._updatePersonAssignment(person, 'role/member')
self.portal.REQUEST['test_computerNotAllowedAllocationScope_OpenFriend'] = \
self._makeNotificationMessage(computer.getReference())
friend_person = self._makePerson(self.generateNewId())
friend_person = self._makePerson()
computer.edit(allocation_scope='open/friend',
destination_section=friend_person.getRelativeUrl())
computer.Computer_checkAndUpdateAllocationScope()
ticket = computer.Computer_checkAndUpdateAllocationScope()
self.tic()
self.assertEquals(computer.getAllocationScope(), 'open/personal')
ticket = self._getGeneratedSupportRequest(computer)
self.assertEquals(ticket.getSimulationState(), 'suspended')
self.assertEquals(ticket.getSimulationState(), 'validated')
self.assertEqual('Visited by SupportRequest_trySendNotificationMessage ' \
'%s %s %s %s' % \
'%s %s %s' % \
('We have changed allocation scope for %s' % computer.getReference(),
'Test NM content\n%s\n' % computer.getReference(), person.getRelativeUrl(), '1'),
'Test NM content\n%s\n' % computer.getReference(), person.getRelativeUrl()),
ticket.workflow_history['edit_workflow'][-1]['comment'])
@simulate('ERP5Site_isSupportRequestCreationClosed', '*args, **kwargs','return 0')
@simulate('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',
'message_title, message, source_relative_url',
'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' \
'%s %s %s" % (message_title, message, source_relative_url))\n' \
'return 1')
def test_computerToCloseAllocationScope_OpenPersonal(self):
computer = self._makeComputer(self.new_id)
computer = self._makeComputer()
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()
computer.edit(allocation_scope='open/personal')
support_request = computer.Computer_checkAndUpdatePersonalAllocationScope()
self.tic()
self.assertEquals('validated', support_request.getSimulationState())
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' % \
'%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'])
'Test NM content\n%s\n' % computer.getReference(), person.getRelativeUrl()),
support_request.workflow_history['edit_workflow'][-1]['comment'])
def test_computerNormalAllocationScope_OpenPersonal(self):
computer = self._makeComputer(self.new_id)
computer = self._makeComputer()
person = computer.getSourceAdministrationValue()
self._updatePersonAssignment(person, 'role/member')
......@@ -1972,7 +1969,7 @@ class TestSlapOSComputer_notifyWrongAllocationScope(testSlapOSMixin):
def test_computerAllowedAllocationScope_OpenPublic(self):
computer = self._makeComputer(self.new_id)
computer = self._makeComputer()
person = computer.getSourceAdministrationValue()
self._updatePersonAssignment(person, 'role/service_provider')
......@@ -1983,11 +1980,11 @@ class TestSlapOSComputer_notifyWrongAllocationScope(testSlapOSMixin):
def test_computerAllowedAllocationScope_OpenFriend(self):
computer = self._makeComputer(self.new_id)
computer = self._makeComputer()
person = computer.getSourceAdministrationValue()
self._updatePersonAssignment(person, 'role/service_provider')
friend_person = self._makePerson(self.generateNewId())
friend_person = self._makePerson()
computer.edit(allocation_scope='open/friend',
destination_section=friend_person.getRelativeUrl())
computer.Computer_checkAndUpdateAllocationScope()
......@@ -1995,65 +1992,138 @@ class TestSlapOSComputer_notifyWrongAllocationScope(testSlapOSMixin):
self.assertEquals(computer.getAllocationScope(), 'open/friend')
class TestSlapOSComputer_CheckState(testSlapOSMixin):
class TestSlapOSPerson_isServiceProvider(testSlapOSMixin):
def beforeTearDown(self):
self._cancelTestSupportRequestList()
transaction.abort()
def afterSetUp(self):
super(TestSlapOSComputer_CheckState, self).afterSetUp()
super(TestSlapOSPerson_isServiceProvider, self).afterSetUp()
def test_Person_isServiceProvider(self):
person = self.portal.person_module.template_member\
.Base_createCloneDocument(batch_mode=1)
person.edit(reference='TESTPERSON-%s' % (self.generateNewId(), ))
self.assertFalse(person.Person_isServiceProvider())
person.setRole("service_provider")
self.assertTrue(person.Person_isServiceProvider())
def test_Person_isServiceProvider_assignment(self):
person = self.portal.person_module.template_member\
.Base_createCloneDocument(batch_mode=1)
person.edit(reference='TESTPERSON-%s' % (self.generateNewId(), ))
self.assertFalse(person.Person_isServiceProvider())
assignment = person.newContent(portal_type="Assignment",
role="service_provider")
self.assertFalse(person.Person_isServiceProvider())
assignment.open()
self.assertTrue(person.Person_isServiceProvider())
class TestSlapOSisSupportRequestCreationClosed(testSlapOSMixin):
def beforeTearDown(self):
transaction.abort()
def afterSetUp(self):
super(TestSlapOSisSupportRequestCreationClosed, 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):
def _makePerson(self):
# 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
person.edit(reference='TESTPERSON-%s' % (self.generateNewId(), ))
person.immediateReindexObject()
return person
def test_ERP5Site_isSupportRequestCreationClosed(self):
person = self._makePerson()
other_person = self._makePerson()
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()
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())
class TestSlapOSGenerateSupportRequestForSlapOS(testSlapOSMixin):
def afterSetUp(self):
super(TestSlapOSGenerateSupportRequestForSlapOS, self).afterSetUp()
self.tic()
self._cancelTestSupportRequestList()
def _makeHostingSubscription(self, new_id):
def _cancelTestSupportRequestList(self):
for support_request in self.portal.portal_catalog(
portal_type="Support Request",
simulation_state=["validated", "suspended"]):
support_request.invalidate()
self.tic()
def _makeHostingSubscription(self):
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()
new_id = self.generateNewId()
hosting_subscription.edit(
title= "Test hosting sub ticket %s" % new_id,
reference="TESTHST-%s" % new_id,
......@@ -2076,6 +2146,274 @@ class TestSlapOSComputer_CheckState(testSlapOSMixin):
hosting_subscription.requestStart(**kw)
hosting_subscription.requestInstance(**kw)
def _makeComputer(self):
super(TestSlapOSGenerateSupportRequestForSlapOS, self)._makeComputer()
# Clone computer document
self.computer.edit(
source_administration_value=self._makePerson()
)
return self.computer
def _makePerson(self):
# Clone computer document
person = self.portal.person_module.template_member\
.Base_createCloneDocument(batch_mode=1)
person.edit(reference='TESTPERSON-%s' % (self.generateNewId(), ))
person.immediateReindexObject()
return person
def _makeSoftwareInstallation(self):
self._makeComputer()
software_installation = self.portal\
.software_installation_module.template_software_installation\
.Base_createCloneDocument(batch_mode=1)
software_installation.edit(
url_string=self.generateNewSoftwareReleaseUrl(),
aggregate=self.computer.getRelativeUrl(),
reference='TESTSOFTINSTS-%s' % self.generateNewId(),
title='Start requested for %s' % self.computer.getUid()
)
software_installation.validate()
software_installation.requestStart()
return software_installation
def test_computer_Base_generateSupportRequestForSlapOS(self):
self._makeComputer()
title = "Test Support Request %s" % self.computer.getReference()
support_request = self.computer.Base_generateSupportRequestForSlapOS(
title, title, self.computer.getRelativeUrl()
)
self.tic()
self.assertNotEqual(support_request, None)
self.assertEqual(support_request.getSimulationState(), "validated")
self.assertEqual(support_request.getRelativeUrl(),
self.portal.REQUEST.get("support_request_in_progress", None))
# The support request is added to computer owner.
self.assertEquals(support_request.getDestinationDecision(),
self.computer.getSourceAdministration())
self.assertEquals(support_request.getTitle(), title)
self.assertEquals(support_request.getDescription(), title)
self.assertEquals(support_request.getSourceProjectValue(),
self.computer)
def test_software_instance_Base_generateSupportRequestForSlapOS(self):
hosting_subscription = self._makeHostingSubscription()
self._makeSoftwareInstance(hosting_subscription,
self.generateNewSoftwareReleaseUrl())
instance = hosting_subscription.getPredecessorValue()
title = "Test Support Request %s" % instance.getReference()
support_request = instance.Base_generateSupportRequestForSlapOS(
title, title, instance.getRelativeUrl()
)
self.tic()
self.assertNotEqual(support_request, None)
self.assertEqual(support_request.getSimulationState(), "validated")
self.assertEqual(support_request.getRelativeUrl(),
self.portal.REQUEST.get("support_request_in_progress", None))
# The support request is added to computer owner.
self.assertEquals(support_request.getDestinationDecision(),
hosting_subscription.getDestinationSection())
self.assertEquals(support_request.getTitle(), title)
self.assertEquals(support_request.getDescription(), title)
self.assertEquals(support_request.getSourceProjectValue(),
instance)
def test_hosting_subscription_Base_generateSupportRequestForSlapOS(self):
hosting_subscription = self._makeHostingSubscription()
title = "Test Support Request %s" % hosting_subscription.getReference()
support_request = hosting_subscription.Base_generateSupportRequestForSlapOS(
title, title, hosting_subscription.getRelativeUrl()
)
self.tic()
self.assertNotEqual(support_request, None)
self.assertEqual(support_request.getSimulationState(), "validated")
self.assertEqual(support_request.getRelativeUrl(),
self.portal.REQUEST.get("support_request_in_progress", None))
# The support request is added to computer owner.
self.assertEquals(support_request.getDestinationDecision(),
hosting_subscription.getDestinationSection())
self.assertEquals(support_request.getTitle(), title)
self.assertEquals(support_request.getDescription(), title)
self.assertEquals(support_request.getSourceProjectValue(),
hosting_subscription)
def test_software_installation_Base_generateSupportRequestForSlapOS(self):
software_installation = self._makeSoftwareInstallation()
title = "Test Support Request %s" % software_installation.generateNewId()
support_request = software_installation.Base_generateSupportRequestForSlapOS(
title, title, software_installation.getRelativeUrl()
)
self.tic()
self.assertNotEqual(support_request, None)
self.assertEqual(support_request.getSimulationState(), "validated")
self.assertEqual(support_request.getRelativeUrl(),
self.portal.REQUEST.get("support_request_in_progress", None))
# The support request is added to computer owner.
self.assertEquals(support_request.getDestinationDecision(),
software_installation.getDestinationSection())
self.assertEquals(support_request.getTitle(), title)
self.assertEquals(support_request.getDescription(), title)
self.assertEquals(support_request.getSourceProjectValue(),
software_installation)
def test_Base_generateSupportRequestForSlapOS_do_not_recreate_if_open(self):
self._makeComputer()
title = "Test Support Request %s" % self.computer.getReference()
support_request = self.computer.Base_generateSupportRequestForSlapOS(
title, title, self.computer.getRelativeUrl()
)
self.tic()
self.portal.REQUEST.set("support_request_in_progress", None)
same_support_request = self.computer.Base_generateSupportRequestForSlapOS(
title, title, self.computer.getRelativeUrl()
)
self.assertEqual(support_request, same_support_request)
def test_Base_generateSupportRequestForSlapOS_do_not_recreate_if_suspended(self):
self._makeComputer()
title = "Test Support Request %s" % self.computer.getReference()
support_request = self.computer.Base_generateSupportRequestForSlapOS(
title, title, self.computer.getRelativeUrl()
)
support_request.suspend()
self.tic()
self.portal.REQUEST.set("support_request_in_progress", None)
same_support_request = self.computer.Base_generateSupportRequestForSlapOS(
title, title, self.computer.getRelativeUrl()
)
self.assertEqual(support_request, same_support_request)
def test_Base_generateSupportRequestForSlapOS_recreate_if_closed(self):
self._makeComputer()
title = "Test Support Request %s" % self.computer.getReference()
support_request = self.computer.Base_generateSupportRequestForSlapOS(
title, title, self.computer.getRelativeUrl())
self.tic()
support_request.invalidate()
self.tic()
self.portal.REQUEST.set("support_request_in_progress", None)
support_request = self.computer.Base_generateSupportRequestForSlapOS(
title, title, self.computer.getRelativeUrl()
)
self.tic()
self.assertNotEqual(support_request, None)
def test_Base_generateSupportRequestForSlapOS_recreate(self):
self._makeComputer()
title = "Test Support Request %s" % self.computer.getRelativeUrl()
support_request = self.computer.Base_generateSupportRequestForSlapOS(
title, title, self.computer.getRelativeUrl())
same_support_request = self.computer.Base_generateSupportRequestForSlapOS(
title, title, self.computer.getRelativeUrl()
)
self.assertEqual(support_request, same_support_request)
def test_Base_generateSupportRequestForSlapOS_inprogress(self):
in_progress = self.portal.support_request_module.getRelativeUrl()
self.portal.REQUEST.set("support_request_in_progress", in_progress)
self._makeComputer()
title = "Test Support Request %s" % self.computer.getRelativeUrl()
support_request = self.computer.Base_generateSupportRequestForSlapOS(
title, title, self.computer.getRelativeUrl())
self.assertEqual(support_request.getRelativeUrl(), in_progress)
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 _makeSupportRequest(self):
support_request = self.portal.\
support_request_module.\
slapos_crm_support_request_template_for_monitoring.\
Base_createCloneDocument(batch_mode=1)
return support_request
def _cancelTestSupportRequestList(self):
for support_request in self.portal.portal_catalog(
portal_type="Support Request",
title="% 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, computer_uid, request_title):
support_request = self.portal.portal_catalog.getResultValue(
portal_type = 'Support Request',
title = request_title,
simulation_state = 'validated',
source_project_uid = computer_uid
)
return support_request
def _makeComputer(self):
super(TestSlapOSComputer_CheckState, self)._makeComputer()
# Clone computer document
self.computer.edit(
source_administration_value=self._makePerson()
)
return self.computer
def _makePerson(self):
# Clone computer document
person = self.portal.person_module.template_member\
.Base_createCloneDocument(batch_mode=1)
person.edit(reference='TESTPERSON-%s' % (self.generateNewId(), ))
person.immediateReindexObject()
return person
def _simulateBase_generateSupportRequestForSlapOS(self):
script_name = 'Base_generateSupportRequestForSlapOS'
if script_name in self.portal.portal_skins.custom.objectIds():
......@@ -2084,7 +2422,7 @@ class TestSlapOSComputer_CheckState(testSlapOSMixin):
script_name,
'*args, **kw',
'# Script body\n'
"""return 'Visited Base_generateSupportRequestForSlapOS'""")
"""return context.getPortalObject().REQUEST['_simulateBase_generateSupportRequestForSlapOS']""")
transaction.commit()
def _dropBase_generateSupportRequestForSlapOS(self):
......@@ -2094,9 +2432,9 @@ class TestSlapOSComputer_CheckState(testSlapOSMixin):
transaction.commit()
self.assertFalse(script_name in self.portal.portal_skins.custom.objectIds())
@simulate('ERP5Site_isSupportRequestCreationClosed', '*args, **kwargs','return 0')
def test_Computer_checkState_call_support_request(self):
computer = self._makeComputer(self.new_id)
computer = self._makeComputer()
memcached_dict = self.portal.portal_memcached.getMemcachedDict(
key_prefix='slap_tool',
plugin_path='portal_memcached/default_memcached_plugin')
......@@ -2106,51 +2444,60 @@ class TestSlapOSComputer_CheckState(testSlapOSMixin):
)
self._simulateBase_generateSupportRequestForSlapOS()
support_request = self._makeSupportRequest()
self.portal.REQUEST.set('_simulateBase_generateSupportRequestForSlapOS',
support_request)
try:
result = computer.Computer_checkState()
computer_support_request = computer.Computer_checkState()
finally:
self._dropBase_generateSupportRequestForSlapOS()
self.assertEqual('Visited Base_generateSupportRequestForSlapOS',
result)
self.assertEqual(support_request,
computer_support_request)
@simulate('ERP5Site_isSupportRequestCreationClosed', '*args, **kwargs','return 0')
def test_Computer_checkState_empty_cache(self):
computer = self._makeComputer(self.new_id)
computer = self._makeComputer()
self._simulateBase_generateSupportRequestForSlapOS()
support_request = self._makeSupportRequest()
self.portal.REQUEST.set('_simulateBase_generateSupportRequestForSlapOS',
support_request)
try:
result = computer.Computer_checkState()
computer_support_request = computer.Computer_checkState()
finally:
self._dropBase_generateSupportRequestForSlapOS()
self.assertEqual('Visited Base_generateSupportRequestForSlapOS',
result)
self.assertEqual(support_request,
computer_support_request)
@simulate('ERP5Site_isSupportRequestCreationClosed', '*args, **kwargs','return 0')
@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',
'message_title, message, source_relative_url',
'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))')
'%s %s %s" % (message_title, message, source_relative_url))')
def test_Computer_checkState_notify(self):
computer = self._makeComputer(self.new_id)
computer = self._makeComputer()
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)}
{"created_at":"%s" % (DateTime() - 0.1)}
)
message_interval_per_day = 5
self.portal.REQUEST['test_Computer_checkState_notify'] = \
self._makeNotificationMessage(computer.getReference())
......@@ -2162,29 +2509,29 @@ class TestSlapOSComputer_CheckState(testSlapOSMixin):
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] ', ''),
'%s %s %s' % ( \
ticket_title,
'Test NM content\n%s\n' % computer.getReference(),
person.getRelativeUrl(), message_interval_per_day),
person.getRelativeUrl()),
ticket.workflow_history['edit_workflow'][-1]['comment'])
@simulate('ERP5Site_isSupportRequestCreationClosed', '*args, **kwargs','return 0')
@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',
'message_title, message, source_relative_url',
'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))')
'%s %s %s" % (message_title, message, source_relative_url))')
def test_Computer_checkState_empty_cache_notify(self):
computer = self._makeComputer(self.new_id)
computer = self._makeComputer()
person = computer.getSourceAdministrationValue()
message_interval_per_day = 5
self.portal.REQUEST['test_Computer_checkState_empty_cache_notify'] = \
self._makeNotificationMessage(computer.getReference())
......@@ -2195,62 +2542,507 @@ class TestSlapOSComputer_CheckState(testSlapOSMixin):
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] ', ''),
'%s %s %s' % ( \
ticket_title,
'Test NM content\n%s\n' % computer.getReference(),
person.getRelativeUrl(), message_interval_per_day),
person.getRelativeUrl()),
ticket.workflow_history['edit_workflow'][-1]['comment'])
class TestSlapOSHostingSubscription_createSupportRequestEvent(testSlapOSMixin):
def _makeNotificationMessage(self, reference):
notification_message = self.portal.notification_message_module.newContent(
portal_type="Notification Message",
title='%s created an event' % reference,
text_content='Test NM content<br/>%s<br/>' % reference,
content_type='text/html',
)
return notification_message.getRelativeUrl()
def _makeHostingSubscription(self):
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()
new_id = self.generateNewId()
hosting_subscription.edit(
title= "Test hosting sub ticket %s" % new_id,
reference="TESTHST-%s" % new_id,
destination_section_value=person
)
return hosting_subscription
def _getGeneratedSupportRequest(self, hosting_suscription_uid):
support_request = self.portal.portal_catalog.getResultValue(
portal_type = 'Support Request',
simulation_state = "validated",
source_project_uid = hosting_suscription_uid
)
return support_request
@simulate('ERP5Site_isSupportRequestCreationClosed', '*args, **kwargs','return 0')
@simulate('NotificationTool_getDocumentValue',
'reference=None',
'assert reference == "slapos-crm-hosting_subscription_state.notification"\n' \
'assert reference == "test-slapos-crm-check.notification"\n' \
'return context.restrictedTraverse(' \
'context.REQUEST["test_SoftwareInstance_checkState"])')
'context.REQUEST["testHostingSubscription_createSupportRequestEvent"])')
@simulate('SupportRequest_trySendNotificationMessage',
'message_title, message, source_relative_url, interval_of_day=1',
'message_title, message, source_relative_url',
'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'
}]
'%s %s %s" % (message_title, message, source_relative_url))')
def testHostingSubscription_createSupportRequestEvent(self):
hosting_subscription = self._makeHostingSubscription()
person = hosting_subscription.getDestinationSectionValue()
self.portal.REQUEST['testHostingSubscription_createSupportRequestEvent'] = \
self._makeNotificationMessage(hosting_subscription.getReference())
hosting_subscription.HostingSubscription_createSupportRequestEvent(
hosting_subscription, "test-slapos-crm-check.notification")
self.tic()
ticket_title = "Hosting Subscription %s is failing." % hosting_subscription.getTitle()
ticket = self._getGeneratedSupportRequest(
hosting_subscription.getUid())
self.assertNotEqual(ticket, None)
self.assertEqual(ticket.getSimulationState(), "validated")
self.assertEqual('Visited by SupportRequest_trySendNotificationMessage ' \
'%s %s %s' % ( \
ticket_title,
'Test NM content\n%s\n' % hosting_subscription.getReference(),
person.getRelativeUrl()),
ticket.workflow_history['edit_workflow'][-1]['comment'])
ticket.suspend()
self.tic()
self.assertEquals(None, self._getGeneratedSupportRequest(
hosting_subscription.getUid()))
hosting_subscription.HostingSubscription_createSupportRequestEvent(
hosting_subscription, "test-slapos-crm-check.notification")
self.tic()
previous_ticket = ticket
ticket = self._getGeneratedSupportRequest(
hosting_subscription.getUid())
self.assertEqual(ticket, previous_ticket)
self.assertEqual(ticket.getSimulationState(), "validated")
self.assertEqual('Visited by SupportRequest_trySendNotificationMessage ' \
'%s %s %s' % ( \
ticket_title.replace('', ''),
'Test NM content\n%s\n' % hosting_subscription.getReference(),
person.getRelativeUrl()),
ticket.workflow_history['edit_workflow'][-1]['comment'])
@simulate('ERP5Site_isSupportRequestCreationClosed', '*args, **kwargs','return 1')
def testHostingSubscription_createSupportRequestEvent_closed(self):
hosting_subscription = self._makeHostingSubscription()
self.assertEquals(None,
hosting_subscription.HostingSubscription_createSupportRequestEvent(
hosting_subscription, "test-slapos-crm-check.notification"))
@simulate('ERP5Site_isSupportRequestCreationClosed', '*args, **kwargs','return 0')
def testHostingSubscription_createSupportRequestEvent_no_person(self):
hosting_subscription = self._makeHostingSubscription()
hosting_subscription.setDestinationSectionValue(None)
self.assertEquals(None,
hosting_subscription.HostingSubscription_createSupportRequestEvent(
hosting_subscription, "test-slapos-crm-check.notification"))
class TestSlapOSHasError(testSlapOSMixin):
def _makeSoftwareRelease(self, software_release_url=None):
software_release = self.portal.software_release_module\
.template_software_release.Base_createCloneDocument(batch_mode=1)
new_id = self.generateNewId()
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, software_release_url):
software_installation = self.portal\
.software_installation_module.template_software_installation\
.Base_createCloneDocument(batch_mode=1)
new_id = self.generateNewId()
software_installation.edit(
url_string=software_release_url,
aggregate=self.computer.getRelativeUrl(),
reference='TESTSOFTINSTS-%s' % new_id,
title='Start requested for %s' % self.computer.getUid()
)
software_installation.validate()
software_installation.requestStart()
return software_installation
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 _makeHostingSubscription(self):
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()
new_id = self.generateNewId()
hosting_subscription.edit(
title= "Test hosting sub ticket %s" % new_id,
reference="TESTHST-%s" % new_id,
destination_section_value=person
)
return hosting_subscription
def _makeComputerPartitionList(self):
for i in range(1, 5):
id_ = 'partition%s' % (i, )
p = self.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 test_SoftwareInstance_hasReportedError(self):
hosting_subscription = self._makeHostingSubscription()
self._makeSoftwareInstance(hosting_subscription,
self.generateNewSoftwareReleaseUrl())
instance = hosting_subscription.getPredecessorValue()
self._makeComputer()
self._makeComputerPartitionList()
memcached_dict = self.portal.portal_memcached.getMemcachedDict(
key_prefix='slap_tool',
plugin_path='portal_memcached/default_memcached_plugin')
error_date = DateTime()
memcached_dict[instance.getReference()] = json.dumps(
{"created_at":"%s" % error_date, "text": "#error "}
)
self.assertEquals(instance.SoftwareInstance_hasReportedError(), None)
instance.setAggregateValue(self.computer.partition1)
self.assertEquals(instance.SoftwareInstance_hasReportedError(), error_date)
memcached_dict[instance.getReference()] = json.dumps(
{"created_at":"%s" % DateTime(), "text":"#error "}
{"created_at":"%s" % error_date, "text": "#access "}
)
self.assertEquals(instance.SoftwareInstance_hasReportedError(), None)
def test_SoftwareInstallation_hasReportedError(self):
software_release = self._makeSoftwareRelease()
self._makeComputer()
installation = self._makeSoftwareInstallation(
software_release.getUrlString()
)
message_interval_per_day = 0
self.portal.REQUEST['test_SoftwareInstance_checkState'] = \
self._makeNotificationMessage(instance.getReference())
memcached_dict = self.portal.portal_memcached.getMemcachedDict(
key_prefix='slap_tool',
plugin_path='portal_memcached/default_memcached_plugin')
self.assertEquals(installation.SoftwareInstallation_hasReportedError(), None)
error_date = DateTime()
memcached_dict[installation.getReference()] = json.dumps(
{"created_at":"%s" % error_date, "text": "#error "}
)
self.assertEquals(installation.SoftwareInstallation_hasReportedError(), error_date)
memcached_dict[installation.getReference()] = json.dumps(
{"created_at":"%s" % error_date, "text": "#building "}
)
self.assertEquals(installation.SoftwareInstallation_hasReportedError(), None)
@simulate('ERP5Site_isSupportRequestCreationClosed', '','return 0')
@simulate('HostingSubscription_createSupportRequestEvent',
'instance, notification_message_reference',
'return "Visited by HostingSubscription_createSupportRequestEvent ' \
'%s %s" % (instance.getUid(), notification_message_reference)')
def testHostingSubscription_checkSoftwareInstanceState(self):
date = DateTime()
def getCreationDate(*args, **kwargs):
return date - 2
from Products.ERP5Type.Base import Base
original_get_creation = Base.getCreationDate
Base.getCreationDate = getCreationDate
try:
hosting_subscription = self._makeHostingSubscription()
self.assertEquals(hosting_subscription.getCreationDate(), date - 2)
self._makeSoftwareInstance(hosting_subscription,
self.generateNewSoftwareReleaseUrl())
instance = hosting_subscription.getPredecessorValue()
self.assertEquals(instance.getCreationDate(), date - 2)
self._makeComputer()
self._makeComputerPartitionList()
instance.setAggregateValue(self.computer.partition1)
memcached_dict = self.portal.portal_memcached.getMemcachedDict(
key_prefix='slap_tool',
plugin_path='portal_memcached/default_memcached_plugin')
error_date = DateTime()
memcached_dict[instance.getReference()] = json.dumps(
{"created_at":"%s" % error_date, "text": "#error "}
)
self.assertEquals(
'Visited by HostingSubscription_createSupportRequestEvent %s %s' % \
(instance.getUid(),
"slapos-crm-hosting-subscription-instance-state.notification"),
hosting_subscription.HostingSubscription_checkSoftwareInstanceState())
memcached_dict[instance.getReference()] = json.dumps(
{"created_at":"%s" % error_date, "text": "#access "}
)
self.assertEquals(None,
hosting_subscription.HostingSubscription_checkSoftwareInstanceState())
finally:
Base.getCreationDate = original_get_creation
self.portal.portal_types.resetDynamicDocumentsOnceAtTransactionBoundary()
transaction.commit()
@simulate('ERP5Site_isSupportRequestCreationClosed', '','return 0')
@simulate('HostingSubscription_createSupportRequestEvent',
'instance, notification_message_reference',
'return "Visited by HostingSubscription_createSupportRequestEvent ' \
'%s %s" % (instance.getRelativeUrl(), notification_message_reference)')
def testHostingSubscription_checkSoftwareInstanceState_partially_allocation(self):
date = DateTime()
def getCreationDate(*args, **kwargs):
return date - 2
from Products.ERP5Type.Base import Base
original_get_creation = Base.getCreationDate
Base.getCreationDate = getCreationDate
try:
hosting_subscription = self._makeHostingSubscription()
self.assertEquals(hosting_subscription.getCreationDate(), date - 2)
self._makeSoftwareInstance(hosting_subscription,
self.generateNewSoftwareReleaseUrl())
instance = hosting_subscription.getPredecessorValue()
self.assertEquals(instance.getCreationDate(), date - 2)
self._makeComputer()
self._makeComputerPartitionList()
instance.setAggregateValue(self.computer.partition1)
kw = dict(
software_release=hosting_subscription.getUrlString(),
software_type=self.generateNewSoftwareType(),
instance_xml=self.generateSafeXml(),
sla_xml=self.generateSafeXml(),
shared=False,
software_title="Another INstance %s" % self.generateNewId(),
state='started'
)
instance.requestInstance(**kw)
self.tic()
ticket_url = instance.SoftwareInstance_checkState()
memcached_dict = self.portal.portal_memcached.getMemcachedDict(
key_prefix='slap_tool',
plugin_path='portal_memcached/default_memcached_plugin')
error_date = DateTime()
memcached_dict[instance.getReference()] = json.dumps(
{"created_at":"%s" % error_date, "text": "#access "}
)
self.assertEquals(
'Visited by HostingSubscription_createSupportRequestEvent %s %s' % \
(instance.getPredecessor(portal_type="Software Instance"),
"slapos-crm-hosting-subscription-instance-allocation.notification"),
hosting_subscription.HostingSubscription_checkSoftwareInstanceState())
kw["state"] = "destroyed"
instance.requestInstance(**kw)
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'])
self.assertEquals(
None,
hosting_subscription.HostingSubscription_checkSoftwareInstanceState())
finally:
Base.getCreationDate = original_get_creation
self.portal.portal_types.resetDynamicDocumentsOnceAtTransactionBoundary()
transaction.commit()
@simulate('ERP5Site_isSupportRequestCreationClosed', '','return 0')
def testHostingSubscription_checkSoftwareInstanceState_too_early(self):
hosting_subscription = self._makeHostingSubscription()
self._makeSoftwareInstance(hosting_subscription,
self.generateNewSoftwareReleaseUrl())
instance = hosting_subscription.getPredecessorValue()
self._makeComputer()
self._makeComputerPartitionList()
instance.setAggregateValue(self.computer.partition1)
memcached_dict = self.portal.portal_memcached.getMemcachedDict(
key_prefix='slap_tool',
plugin_path='portal_memcached/default_memcached_plugin')
error_date = DateTime()
memcached_dict[instance.getReference()] = json.dumps(
{"created_at":"%s" % error_date, "text": "#error "}
)
self.assertEquals(
None,
hosting_subscription.HostingSubscription_checkSoftwareInstanceState())
class TestSupportRequestTrySendNotificationMessage(testSlapOSMixin):
def _makeComputer(self):
super(TestSupportRequestTrySendNotificationMessage, self)._makeComputer()
# Clone computer document
self.computer.edit(
source_administration_value=self._makePerson()
)
return self.computer
def _makePerson(self):
# Clone computer document
person = self.portal.person_module.template_member\
.Base_createCloneDocument(batch_mode=1)
person.edit(reference='TESTPERSON-%s' % (self.generateNewId(), ))
person.immediateReindexObject()
return person
def test_SupportRequest_trySendNotificationMessage(self):
self._makeComputer()
person = self.computer.getSourceAdministrationValue()
title = "Test Support Request %s" % self.computer.getReference()
text_content='Test NM content<br/>%s<br/>' % self.computer.getReference()
support_request = self.portal.support_request_module.newContent(\
title=title, description=title,
destination_decision=self.computer.getSourceAdministration(),
source_project_value=self.computer.getRelativeUrl())
support_request.validate()
self.tic()
first_event = support_request.SupportRequest_trySendNotificationMessage(
message_title=title, message=text_content,
source_relative_url=person.getRelativeUrl()
)
self.assertNotEqual(first_event, None)
self.assertEquals(
support_request.getFollowUpRelatedList(),
[first_event.getRelativeUrl()])
self.assertEquals(title, first_event.getTitle())
self.assertEquals(text_content, first_event.getTextContent())
self.assertNotEquals(None, first_event.getStartDate())
self.assertEquals("service_module/slapos_crm_information",
first_event.getResource())
self.assertEquals(first_event.getSource(), person.getRelativeUrl())
self.assertEquals(first_event.getFollowUp(), support_request.getRelativeUrl())
event = support_request.SupportRequest_trySendNotificationMessage(
message_title=title, message=text_content,
source_relative_url=person.getRelativeUrl()
)
self.assertEqual(event, first_event)
self.assertEquals(title, event.getTitle())
self.assertEquals(text_content, event.getTextContent())
self.assertNotEquals(None, event.getStartDate())
self.assertEquals("service_module/slapos_crm_information",
event.getResource())
self.assertEquals(event.getSource(), person.getRelativeUrl())
title += "__zz"
event = support_request.SupportRequest_trySendNotificationMessage(
message_title=title, message=text_content,
source_relative_url=person.getRelativeUrl(),
)
self.assertEqual(event.getTitle(), title)
self.assertEquals(text_content, event.getTextContent())
self.assertNotEquals(None, event.getStartDate())
self.assertEquals("service_module/slapos_crm_information",
event.getResource())
self.assertEquals(event.getSource(), person.getRelativeUrl())
another_support_request = self.portal.support_request_module.newContent(\
title=title, description=title,
destination_decision=self.computer.getSourceAdministration(),
source_project_value=self.computer.getRelativeUrl())
another_support_request.validate()
self.tic()
another_first_event = \
another_support_request.SupportRequest_trySendNotificationMessage(
message_title=title, message=text_content,
source_relative_url=person.getRelativeUrl())
self.assertNotEqual(another_first_event, None)
self.assertEquals(
another_support_request.getFollowUpRelatedList(),
[another_first_event.getRelativeUrl()])
self.assertEquals(title, another_first_event.getTitle())
self.assertEquals(text_content, another_first_event.getTextContent())
self.assertNotEquals(None, another_first_event.getStartDate())
self.assertEquals("service_module/slapos_crm_information",
another_first_event.getResource())
self.assertEquals(another_first_event.getSource(), person.getRelativeUrl())
self.assertEquals(another_first_event.getFollowUp(),
another_support_request.getRelativeUrl())
\ No newline at end of file
master/bt5/slapos_crm/TestTemplateItem/portal_components/test.erp5.testSlapOSCRMSkins.xml
View file @ bebf7b6a
......@@ -6,10 +6,22 @@
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>_recorded_property_dict</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent>
</value>
</item>
<item>
<key> <string>default_reference</string> </key>
<value> <string>testSlapOSCRMSkins</string> </value>
</item>
<item>
<key> <string>description</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>test.erp5.testSlapOSCRMSkins</string> </value>
......@@ -33,20 +45,7 @@
<item>
<key> <string>text_content_warning_message</string> </key>
<value>
<tuple>
<string>W:206, 12: Unused variable \'event\' (unused-variable)</string>
<string>W:215, 4: Unused variable \'date\' (unused-variable)</string>
<string>W:1328, 4: Unused variable \'hosting_subscription\' (unused-variable)</string>
<string>W:1349, 4: Unused variable \'person\' (unused-variable)</string>
<string>W:1372, 4: Unused variable \'hosting_subscription\' (unused-variable)</string>
<string>W:1394, 4: Unused variable \'hosting_subscription\' (unused-variable)</string>
<string>W:1704, 4: Unused variable \'hosting_subscription\' (unused-variable)</string>
<string>W:1725, 4: Unused variable \'person\' (unused-variable)</string>
<string>W:1748, 4: Unused variable \'hosting_subscription\' (unused-variable)</string>
<string>W:1770, 4: Unused variable \'hosting_subscription\' (unused-variable)</string>
<string>W:1807, 2: Arguments number differs from overridden method (arguments-differ)</string>
<string>W:2033, 2: Arguments number differs from overridden method (arguments-differ)</string>
</tuple>
<tuple/>
</value>
</item>
<item>
......@@ -56,13 +55,28 @@
<item>
<key> <string>workflow_history</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent>
<persistent> <string encoding="base64">AAAAAAAAAAM=</string> </persistent>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="2" aka="AAAAAAAAAAI=">
<pickle>
<global name="PersistentMapping" module="Persistence.mapping"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>data</string> </key>
<value>
<dictionary/>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="3" aka="AAAAAAAAAAM=">
<pickle>
<global name="PersistentMapping" module="Persistence.mapping"/>
</pickle>
......@@ -75,7 +89,7 @@
<item>
<key> <string>component_validation_workflow</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAM=</string> </persistent>
<persistent> <string encoding="base64">AAAAAAAAAAQ=</string> </persistent>
</value>
</item>
</dictionary>
......@@ -84,7 +98,7 @@
</dictionary>
</pickle>
</record>
<record id="3" aka="AAAAAAAAAAM=">
<record id="4" aka="AAAAAAAAAAQ=">
<pickle>
<global name="WorkflowHistoryList" module="Products.ERP5Type.patches.WorkflowTool"/>
</pickle>
......
master/bt5/slapos_crm/TestTemplateItem/portal_components/test.erp5.testSlapOSCRMSupportRequestGeneration.py deleted 100644 → 0
View file @ 339fba9c
# -*- 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'])
master/bt5/slapos_crm/TestTemplateItem/portal_components/test.erp5.testSlapOSCRMSupportRequestGeneration.xml deleted 100644 → 0
View file @ 339fba9c
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="Test Component" module="erp5.portal_type"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>default_reference</string> </key>
<value> <string>testSlapOSCRMSupportRequestGeneration</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>test.erp5.testSlapOSCRMSupportRequestGeneration</string> </value>
</item>
<item>
<key> <string>portal_type</string> </key>
<value> <string>Test Component</string> </value>
</item>
<item>
<key> <string>sid</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>text_content_error_message</string> </key>
<value>
<tuple/>
</value>
</item>
<item>
<key> <string>text_content_warning_message</string> </key>
<value>
<tuple>
<string>W:108, 0: Bad indentation. Found 5 spaces, expected 4 (bad-indentation)</string>
<string>W:111, 0: Bad indentation. Found 5 spaces, expected 4 (bad-indentation)</string>
<string>W:117, 0: Bad indentation. Found 5 spaces, expected 4 (bad-indentation)</string>
<string>W:118, 0: Bad indentation. Found 5 spaces, expected 4 (bad-indentation)</string>
<string>W:120, 0: Bad indentation. Found 5 spaces, expected 4 (bad-indentation)</string>
<string>W: 68, 2: Arguments number differs from overridden method (arguments-differ)</string>
<string>W: 95, 2: Arguments number differs from overridden method (arguments-differ)</string>
<string>W:895, 4: Unused variable \'software_installation\' (unused-variable)</string>
</tuple>
</value>
</item>
<item>
<key> <string>version</string> </key>
<value> <string>erp5</string> </value>
</item>
<item>
<key> <string>workflow_history</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="2" aka="AAAAAAAAAAI=">
<pickle>
<global name="PersistentMapping" module="Persistence.mapping"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>data</string> </key>
<value>
<dictionary>
<item>
<key> <string>component_validation_workflow</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAM=</string> </persistent>
</value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="3" aka="AAAAAAAAAAM=">
<pickle>
<global name="WorkflowHistoryList" module="Products.ERP5Type.patches.WorkflowTool"/>
</pickle>
<pickle>
<tuple>
<none/>
<list>
<dictionary>
<item>
<key> <string>action</string> </key>
<value> <string>validate</string> </value>
</item>
<item>
<key> <string>validation_state</string> </key>
<value> <string>validated</string> </value>
</item>
</dictionary>
</list>
</tuple>
</pickle>
</record>
</ZopeData>
master/bt5/slapos_crm/bt/template_path_list
View file @ bebf7b6a
event_module/slapos_crm_web_message_template
person_module/allocation_tester
portal_alarms/slapos_check_computer_state
portal_alarms/slapos_check_software_instance_state
portal_alarms/slapos_crm_cancel_invoice
portal_alarms/slapos_crm_check_partially_allocated_instance
portal_alarms/slapos_crm_check_computer_state
portal_alarms/slapos_crm_check_instance_in_error
portal_alarms/slapos_crm_check_update_allocation_scope
portal_alarms/slapos_crm_check_update_personal_allocation_scope
portal_alarms/slapos_crm_create_regularisation_request
portal_alarms/slapos_crm_delete_hosting_subscription
portal_alarms/slapos_crm_invalidate_suspended_regularisation_request
......
master/bt5/slapos_crm/bt/template_test_id_list
View file @ bebf7b6a
test.erp5.testSlapOSCRMSkins
test.erp5.testSlapOSCRMAlarm
\ No newline at end of file
test.erp5.testSlapOSCRMSupportRequestGeneration
\ No newline at end of file
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