Commit 056bd3b0 authored by Łukasz Nowak's avatar Łukasz Nowak

Test slapos_update_computer_capacity_scope alarm.

parent c28854ba
master 20240521_variousimprovements 20240704 20240827 20240905_fixup 20240913 20240914 20240925 20240930 20241004 20241008 20241021 20241106 20241107 20241114 20241127 20241209 20241230 20250108 alain-message amarisoft change_subscription_price cleanup cliff-cleanup crm_fix customprice cygwin-link-local event_attachment feat/cmmi_cache feat/openorderbasedsecurity feat/statuscleanup feat/testcase feature/cgroups feature/format-crumbled feature/manager-cpuset feature/option_create_tun feature/tun-cgroup-cleanup fix/slapos-ansible format_loop hosting_daily_invoice hotfix/proxy-root-instance localwip manager_update_parameter master-dev message-box new-api-master-client panel_computer_guid panel_filter_action_dict py3_fixup seb slap_tool_less_sql_query slapos-override-env slapproxy-connection-parameter splitopenorder subscription_change_request tomo_fix_slapos_format tomo_fix_slapos_node_boot tomo_listbox_update trigger_allocation_alarm upgrade_decision_event usage_report wip wip20210701 wip20210819 wip20220131 1.4.27 1.4.26 1.4.25 1.4.24 1.4.23 1.4.22 1.4.21 1.4.20 1.4.19 1.4.17 1.4.16 1.4.15 1.4.14 1.4.12 1.4.11 1.4.10 1.4.9 1.4.8 1.4.7 1.4.6 1.4.5 1.4.4 1.4.3 1.4.2 1.4.1 1.4.0 1.3.18 1.3.17 1.3.16 1.3.15 1.3.14 1.3.13 1.3.12 1.3.11 1.3.10 1.3.9 1.3.8 1.3.7 1.3.6.3 1.3.6.2 1.3.6.1 1.3.6 1.3.5 1.3.4 1.3.3 1.3.2 1.3.1 1.3.0 1.2.4.1 1.2.4 1.2.3.1 1.2.3 1.2.2 1.2.1 1.2.0 1.1.3 1.1.2 1.1.1 1.1.0 1.0.5 1.0.4 1.0.3 1.0.2 1.0.1 1.0.0 1.0.0rc6 1.0.0rc5 1.0.0rc4 1.0.0rc3 1.0.0-rc1 0.35.1 0.34 master-20180917
No related merge requests found
# Copyright (c) 2002-2012 Nexedi SA and Contributors. All Rights Reserved.
from Products.SlapOS.tests.testSlapOSMixin import \
testSlapOSMixin
import transaction
from Products.ERP5Type.tests.utils import createZODBPythonScript
import json
class TestSlapOSGarbageCollectDestroyedRootTreeAlarm(testSlapOSMixin):
def afterSetUp(self):
super(TestSlapOSGarbageCollectDestroyedRootTreeAlarm, self).afterSetUp()
self.computer = self.portal.computer_module.template_computer\
.Base_createCloneDocument(batch_mode=1)
self.computer.edit(
allocation_scope='open/public',
capacity_scope='open',
reference='TESTC-%s' % self.generateNewId(),
)
self.computer.validate()
memcached_dict = self.portal.portal_memcached.getMemcachedDict(
key_prefix='slap_tool',
plugin_path='portal_memcached/default_memcached_plugin')
memcached_dict[self.computer.getReference()] = json.dumps({
'text': '#access ok'
})
transaction.commit()
self.login()
def test_Computer_checkAndUpdateCapacityScope(self):
self.computer.Computer_checkAndUpdateCapacityScope()
self.assertEqual('open', self.computer.getCapacityScope())
def test_Computer_checkAndUpdateCapacityScope_no_capacity_quantity(self):
self._makeTree()
self.computer.edit(capacity_quantity=1)
partition = self.computer.newContent(portal_type='Computer Partition',
reference='part1')
partition.markFree()
partition.markBusy()
partition.validate()
self.software_instance.setAggregate(partition.getRelativeUrl())
self.tic()
self.computer.Computer_checkAndUpdateCapacityScope()
self.assertEqual('close', self.computer.getCapacityScope())
self.assertEqual('Computer capacity limit exceeded',
self.computer.workflow_history['edit_workflow'][-1]['comment'])
def test_Computer_checkAndUpdateCapacityScope_no_access(self):
self.computer.edit(reference='TESTC-%s' % self.generateNewId())
self.computer.Computer_checkAndUpdateCapacityScope()
self.assertEqual('close', self.computer.getCapacityScope())
self.assertEqual("Computer didn't contact the server",
self.computer.workflow_history['edit_workflow'][-1]['comment'])
def test_Computer_checkAndUpdateCapacityScope_close(self):
self.computer.edit(capacity_scope='close')
self.computer.Computer_checkAndUpdateCapacityScope()
self.assertEqual('open', self.computer.getCapacityScope())
def test_Computer_checkAndUpdateCapacityScope_with_error(self):
memcached_dict = self.portal.portal_memcached.getMemcachedDict(
key_prefix='slap_tool',
plugin_path='portal_memcached/default_memcached_plugin')
memcached_dict[self.computer.getReference()] = json.dumps({
'text': '#error not ok'
})
self.computer.Computer_checkAndUpdateCapacityScope()
self.assertEqual('close', self.computer.getCapacityScope())
self.assertEqual("Computer reported an error",
self.computer.workflow_history['edit_workflow'][-1]['comment'])
def test_Computer_checkAndUpdateCapacityScope_with_error_non_public(self):
memcached_dict = self.portal.portal_memcached.getMemcachedDict(
key_prefix='slap_tool',
plugin_path='portal_memcached/default_memcached_plugin')
memcached_dict[self.computer.getReference()] = json.dumps({
'text': '#error not ok'
})
self.computer.edit(allocation_scope='open/personal')
self.computer.Computer_checkAndUpdateCapacityScope()
self.assertEqual('open', self.computer.getCapacityScope())
def _simulateComputer_checkAndUpdateCapacityScope(self):
script_name = 'Computer_checkAndUpdateCapacityScope'
if script_name in self.portal.portal_skins.custom.objectIds():
raise ValueError('Precondition failed: %s exists in custom' % script_name)
createZODBPythonScript(self.portal.portal_skins.custom,
script_name,
'*args, **kwargs',
'# Script body\n'
"""portal_workflow = context.portal_workflow
portal_workflow.doActionFor(context, action='edit_action', comment='Visited by Computer_checkAndUpdateCapacityScope') """ )
transaction.commit()
def _dropComputer_checkAndUpdateCapacityScope(self):
script_name = 'Computer_checkAndUpdateCapacityScope'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
transaction.commit()
def test_alarm(self):
self._simulateComputer_checkAndUpdateCapacityScope()
try:
self.portal.portal_alarms.slapos_update_computer_capacity_scope.activeSense()
self.tic()
finally:
self._dropComputer_checkAndUpdateCapacityScope()
self.assertEqual(
'Visited by Computer_checkAndUpdateCapacityScope',
self.computer.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_non_public(self):
self.computer.edit(allocation_scope='open/personal')
self.tic()
self._simulateComputer_checkAndUpdateCapacityScope()
try:
self.portal.portal_alarms.slapos_update_computer_capacity_scope.activeSense()
self.tic()
finally:
self._dropComputer_checkAndUpdateCapacityScope()
self.assertNotEqual(
'Visited by Computer_checkAndUpdateCapacityScope',
self.computer.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_invalidated(self):
self.computer.invalidate()
self.tic()
self._simulateComputer_checkAndUpdateCapacityScope()
try:
self.portal.portal_alarms.slapos_update_computer_capacity_scope.activeSense()
self.tic()
finally:
self._dropComputer_checkAndUpdateCapacityScope()
self.assertNotEqual(
'Visited by Computer_checkAndUpdateCapacityScope',
self.computer.workflow_history['edit_workflow'][-1]['comment'])
212
\ No newline at end of file
213
\ No newline at end of file
......@@ -9,4 +9,5 @@ testSlapOSCoreSlapOSAssertHostingSubscriptionPredecessorAlarm
testSlapOSCoreSlapOSCloudInteractionWorkflow
testSlapOSCoreSlapOSFreeComputerPartitionAlarm
testSlapOSCoreSlapOSGarbageCollectDestroyedRootTreeAlarm
testSlapOSCoreSlapOSUpdateComputerCapacityScopeAlarm
testSlapOSShadow
\ 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