Skip to content

S
slapos.core
Commit fabe344e authored by Romain Courteaud's avatar Romain Courteaud 🐙
Browse files
Options

slapos_pdm: drop Base_acceptUpgradeDecision

parent 30e10bef
Hide whitespace changes
Inline Side-by-side
Showing with 2 additions and 448 deletions
master/bt5/slapos_pdm/SkinTemplateItem/portal_skins/slapos_pdm/Base_acceptUpgradeDecision.py deleted 100644 → 0
View file @ 30e10bef
if reference is None:
raise ValueError("Missing Reference")
portal = context.getPortalObject()
upgrade_decision_list = portal.portal_catalog(
portal_type="Upgrade Decision",
reference=reference, limit=2)
if not len(upgrade_decision_list):
return context.Base_redirect("",
keep_items={"portal_status_message":
context.Base_translateString("Unable to find the Upgrade Decision.")})
if len(upgrade_decision_list) > 1:
raise ValueError("Duplicated reference for %s. Please contact site administrators." % reference)
upgrade_decision = upgrade_decision_list[0]
if upgrade_decision.getSimulationState() in ['draft', 'planned']:
message = "Sorry, the upgrade is not possible yet!"
elif upgrade_decision.getSimulationState() in ['cancelled', 'rejected']:
message = "Sorry, the upgrade is not possble, Upgrade Decision was Canceled or Rejected!"
elif upgrade_decision.getSimulationState() == 'started':
message = "This Upgrade Decision is already Started."
elif upgrade_decision.getSimulationState() in ['stopped', 'delivered']:
message = "This Upgrade Decision has been already processed."
elif upgrade_decision.getSimulationState() == 'confirmed':
message = "This Upgrade Decision has been requested, it will be processed in few minutes."
upgrade_decision.start()
return context.Base_redirect("",
keep_items={"portal_status_message":
context.Base_translateString(message)})
master/bt5/slapos_pdm/SkinTemplateItem/portal_skins/slapos_pdm/Base_acceptUpgradeDecision.xml deleted 100644 → 0
View file @ 30e10bef
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="PythonScript" module="Products.PythonScripts.PythonScript"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>_bind_names</string> </key>
<value>
<object>
<klass>
<global name="_reconstructor" module="copy_reg"/>
</klass>
<tuple>
<global name="NameAssignments" module="Shared.DC.Scripts.Bindings"/>
<global name="object" module="__builtin__"/>
<none/>
</tuple>
<state>
<dictionary>
<item>
<key> <string>_asgns</string> </key>
<value>
<dictionary>
<item>
<key> <string>name_container</string> </key>
<value> <string>container</string> </value>
</item>
<item>
<key> <string>name_context</string> </key>
<value> <string>context</string> </value>
</item>
<item>
<key> <string>name_m_self</string> </key>
<value> <string>script</string> </value>
</item>
<item>
<key> <string>name_subpath</string> </key>
<value> <string>traverse_subpath</string> </value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</state>
</object>
</value>
</item>
<item>
<key> <string>_params</string> </key>
<value> <string>reference</string> </value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>Base_acceptUpgradeDecision</string> </value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
master/bt5/slapos_pdm/TestTemplateItem/portal_components/test.erp5.testSlapOSPDMSkins.py
View file @ fabe344e
......@@ -23,7 +23,7 @@ from erp5.component.test.SlapOSTestCaseMixin import SlapOSTestCaseMixin, simulat
from DateTime import DateTime
class TestSlapOSPDMMixinSkins(SlapOSTestCaseMixin):
def afterSetUp(self):
def XXXafterSetUp(self):
SlapOSTestCaseMixin.afterSetUp(self)
self.new_id = self.generateNewId()
self.request_kw = dict(
......@@ -50,13 +50,6 @@ class TestSlapOSPDMMixinSkins(SlapOSTestCaseMixin):
p.markFree()
p.validate()
def _markComputePartitionBusy(self, compute_node, software_instance):
for partition in compute_node.contentValues(portal_type='Compute Partition'):
if partition.getSlapState() == 'free':
software_instance.edit(aggregate=partition.getRelativeUrl())
partition.markBusy()
break
def _requestSoftwareRelease(self, software_product_url, effective_date=None):
software_release = self._makeSoftwareRelease(self.project)
if not effective_date:
......@@ -170,235 +163,7 @@ class TestSlapOSPDMMixinSkins(SlapOSTestCaseMixin):
class TestSlapOSPDMSkins(TestSlapOSPDMMixinSkins):
def beforeTearDown(self):
id_list = []
for upgrade_decision in self.portal.portal_catalog(
portal_type="Upgrade Decision", reference="UD-TEST%"):
id_list.append(upgrade_decision.getId())
self.portal.upgrade_decision_module.manage_delObjects(id_list)
self.tic()
def test_tocheck_getSortedSoftwareReleaseListFromSoftwareProduct(self):
project = self.addProject()
software_product = self._makeSoftwareProduct(project)
self.tic()
release_list = software_product.SoftwareProduct_getSortedSoftwareReleaseList(
software_product.getReference())
self.assertEqual(release_list, [])
# published software release
software_release1 = self._makeSoftwareRelease(software_product)
software_release1.edit(aggregate_value=software_product.getRelativeUrl(),
url_string='http://example.org/1-%s.cfg' % self.new_id,
effective_date=(DateTime() + 5)
)
software_release1.publish()
software_release2 = self._makeSoftwareRelease(software_product)
software_release2.edit(aggregate_value=software_product.getRelativeUrl(),
url_string='http://example.org/2-%s.cfg' % self.new_id
)
software_release2.publish()
# 1 released software release, should not appear
software_release3 = self._makeSoftwareRelease(software_product)
self.assertTrue(software_release3.getValidationState() == 'released')
software_release3.edit(aggregate_value=software_product.getRelativeUrl(),
url_string='http://example.org/3-%s.cfg' % self.new_id
)
self.tic()
release_list = software_product.SoftwareProduct_getSortedSoftwareReleaseList(
software_product.getReference())
self.assertEqual([release.getUrlString() for release in release_list],
['http://example.org/2-%s.cfg' % self.new_id, 'http://example.org/1-%s.cfg' % self.new_id])
def test_tocheck_getSortedSoftwareReleaseListFromSoftwareProduct_Changed(self):
project = self.addProject()
software_product = self._makeSoftwareProduct(project)
release_list = software_product.SoftwareProduct_getSortedSoftwareReleaseList(
software_product.getReference())
self.assertEqual(release_list, [])
# 2 published software releases
software_release2 = self._makeSoftwareRelease(software_product)
software_release2.publish()
software_release2.edit(aggregate_value=software_product.getRelativeUrl(),
url_string='http://example.org/2-%s.cfg' % self.new_id,
effective_date=(DateTime() - 2)
)
# Newest software release
software_release1 = self._makeSoftwareRelease(software_product)
software_release1.publish()
software_release1.edit(aggregate_value=software_product.getRelativeUrl(),
url_string='http://example.org/1-%s.cfg' % self.new_id,
effective_date=DateTime()
)
self.tic()
release_list = software_product.SoftwareProduct_getSortedSoftwareReleaseList(
software_product.getReference())
self.assertEqual([release.getUrlString() for release in release_list],
['http://example.org/1-%s.cfg' % self.new_id, 'http://example.org/2-%s.cfg' % self.new_id])
release_list = software_product.SoftwareProduct_getSortedSoftwareReleaseList(
software_release_url='http://example.org/1-%s.cfg' % self.new_id)
self.assertEqual([release.getUrlString() for release in release_list],
['http://example.org/1-%s.cfg' % self.new_id, 'http://example.org/2-%s.cfg' % self.new_id])
def test_tocheckUpgradeDecision_getComputeNode(self):
project = self.addProject()
compute_node, _ = self._makeComputeNode(project)
upgrade_decision = self._makeUpgradeDecision()
upgrade_decision_line = self._makeUpgradeDecisionLine(upgrade_decision)
upgrade_decision_line.setAggregateValue(compute_node)
found_compute_node = upgrade_decision.UpgradeDecision_getAggregateValue("Compute Node")
self.assertEqual(compute_node.getRelativeUrl(),
found_compute_node.getRelativeUrl())
def test_tocheckUpgradeDecision_getComputeNode_2_lines(self):
project = self.addProject()
compute_node, _ = self._makeComputeNode(project)
upgrade_decision = self._makeUpgradeDecision()
upgrade_decision_line = self._makeUpgradeDecisionLine(upgrade_decision)
upgrade_decision_line.setAggregateValue(compute_node)
upgrade_decision_line = self._makeUpgradeDecisionLine(upgrade_decision)
found_compute_node = upgrade_decision.UpgradeDecision_getAggregateValue("Compute Node")
self.assertEqual(compute_node.getRelativeUrl(),
found_compute_node.getRelativeUrl())
def test_tocheckUpgradeDecision_getComputeNode_2_compute_node(self):
project = self.addProject()
compute_node, _ = self._makeComputeNode(project)
upgrade_decision = self._makeUpgradeDecision()
upgrade_decision_line = self._makeUpgradeDecisionLine(upgrade_decision)
upgrade_decision_line.setAggregateValue(compute_node)
upgrade_decision_line = self._makeUpgradeDecisionLine(upgrade_decision)
upgrade_decision_line.setAggregateValue(compute_node)
self.assertRaises(ValueError, upgrade_decision.UpgradeDecision_getAggregateValue,
document_portal_type="Compute Node")
def test_tocheckUpgradeDecision_getComputeNode_O_compute_node(self):
upgrade_decision = self._makeUpgradeDecision()
self._makeUpgradeDecisionLine(upgrade_decision)
found_compute_node = upgrade_decision.UpgradeDecision_getAggregateValue("Compute Node")
self.assertEqual(None, found_compute_node)
def test_tocheckUpgradeDecision_getInstanceTree(self):
instance_tree = self._makeInstanceTree()
upgrade_decision = self._makeUpgradeDecision()
upgrade_decision_line = self._makeUpgradeDecisionLine(upgrade_decision)
upgrade_decision_line.setAggregateValue(instance_tree)
found_instance_tree = upgrade_decision.UpgradeDecision_getAggregateValue("Instance Tree")
self.assertEqual(instance_tree.getRelativeUrl(),
found_instance_tree.getRelativeUrl())
def test_tocheckUpgradeDecision_getInstanceTree_2_lines(self):
instance_tree = self._makeInstanceTree()
upgrade_decision = self._makeUpgradeDecision()
upgrade_decision_line = self._makeUpgradeDecisionLine(upgrade_decision)
upgrade_decision_line.setAggregateValue(instance_tree)
upgrade_decision_line = self._makeUpgradeDecisionLine(upgrade_decision)
found_instance_tree = upgrade_decision.UpgradeDecision_getAggregateValue("Instance Tree")
self.assertEqual(instance_tree.getRelativeUrl(),
found_instance_tree.getRelativeUrl())
def test_tocheckUpgradeDecision_getInstanceTree_2_hosting(self):
instance_tree = self._makeInstanceTree()
upgrade_decision = self._makeUpgradeDecision()
upgrade_decision_line = self._makeUpgradeDecisionLine(upgrade_decision)
upgrade_decision_line.setAggregateValue(instance_tree)
upgrade_decision_line = self._makeUpgradeDecisionLine(upgrade_decision)
upgrade_decision_line.setAggregateValue(instance_tree)
self.assertRaises(ValueError, upgrade_decision.UpgradeDecision_getAggregateValue,
document_portal_type="Instance Tree")
def test_tocheckUpgradeDecision_getInstanceTree_O_hosting(self):
upgrade_decision = self._makeUpgradeDecision()
self._makeUpgradeDecisionLine(upgrade_decision)
found_instance_tree = upgrade_decision.UpgradeDecision_getAggregateValue("Instance Tree")
self.assertEqual(None, found_instance_tree)
def test_tocheckUpgradeDecision_getAggregateValue_SoftwareRelease(self):
project = self.addProject()
software_product = self._makeSoftwareProduct(project)
software_release = self._makeSoftwareRelease(software_product)
upgrade_decision = self._makeUpgradeDecision()
upgrade_decision_line = self._makeUpgradeDecisionLine(upgrade_decision)
upgrade_decision_line.setAggregateValue(software_release)
found_software_release = upgrade_decision.UpgradeDecision_getAggregateValue("Software Release")
self.assertEqual(software_release.getRelativeUrl(),
found_software_release.getRelativeUrl())
def test_tocheckUpgradeDecision_getAggregateValue_SoftwareRelease_2_lines(self):
project = self.addProject()
software_product = self._makeSoftwareProduct(project)
software_release = self._makeSoftwareRelease(software_product)
upgrade_decision = self._makeUpgradeDecision()
upgrade_decision_line = self._makeUpgradeDecisionLine(upgrade_decision)
upgrade_decision_line.setAggregateValue(software_release)
upgrade_decision_line = self._makeUpgradeDecisionLine(upgrade_decision)
found_software_release = upgrade_decision.UpgradeDecision_getAggregateValue("Software Release")
self.assertEqual(software_release.getRelativeUrl(),
found_software_release.getRelativeUrl())
def test_tocheckUpgradeDecision_getAggregateValue_SoftwareRelease_2_sr(self):
project = self.addProject()
software_product = self._makeSoftwareProduct(project)
software_release = self._makeSoftwareRelease(software_product)
upgrade_decision = self._makeUpgradeDecision()
upgrade_decision_line = self._makeUpgradeDecisionLine(upgrade_decision)
upgrade_decision_line.setAggregateValue(software_release)
upgrade_decision_line = self._makeUpgradeDecisionLine(upgrade_decision)
upgrade_decision_line.setAggregateValue(software_release)
self.assertRaises(ValueError, upgrade_decision.UpgradeDecision_getAggregateValue,
document_portal_type="Software Release")
def test_tocheckUpgradeDecision_getAggregateValue_SoftwareRelease_O_sr(self):
upgrade_decision = self._makeUpgradeDecision()
self._makeUpgradeDecisionLine(upgrade_decision)
found_software_release = upgrade_decision.UpgradeDecision_getAggregateValue("Software Release")
self.assertEqual(None, found_software_release)
def testUpgradeDecision_getAggregateUrlString_SoftwareRelease(self):
software_release = self._makeSoftwareRelease()
upgrade_decision = self._makeUpgradeDecision()
upgrade_decision_line = self._makeUpgradeDecisionLine(upgrade_decision)
upgrade_decision_line.setAggregateValue(software_release)
url_string = upgrade_decision.UpgradeDecision_getAggregateUrlString()
self.assertNotEqual(None, url_string)
self.assertEqual(software_release.getUrlString(), url_string)
def testSoftwareRelease_createUpgradeDecision_compute_node(self):
def test_tocheckSoftwareRelease_createUpgradeDecision_compute_node(self):
project = self.addProject()
person = self._makePerson(project)
compute_node, _ = self._makeComputeNode(project)
......@@ -724,120 +489,9 @@ class TestSlapOSPDMSkins(TestSlapOSPDMMixinSkins):
self.assertEqual(len(upgrade_decision2), 0)
self.assertEqual(upgrade_decision.getSimulationState(), 'started')
def test_tocheckBase_acceptUpgradeDecision_no_reference(self):
self._makeUpgradeDecision()
self.assertRaises(ValueError, self.portal.Base_acceptUpgradeDecision, None)
def test_tocheckBase_acceptUpgradeDecision_duplicated_reference(self):
upgrade_decision = self._makeUpgradeDecision()
upgrade_decision.setReference("UD-TESTBADREFERENCE")
upgrade_decision = self._makeUpgradeDecision()
upgrade_decision.setReference("UD-TESTBADREFERENCE")
self.tic()
self.assertRaises(ValueError, self.portal.Base_acceptUpgradeDecision, None)
def test_tocheckBase_acceptUpgradeDecision_no_upgrade_decision(self):
redirect_url = self.portal.Base_acceptUpgradeDecision('UD-UNEXISTING')
self.assertTrue(redirect_url.endswith(
"?portal_status_message=Unable%20to%20find%20the%20Upgrade%20Decision."),
"%s contains the wrong message" % redirect_url)
def test_tocheckBase_acceptUpgradeDecision_draft_upgrade_decision(self):
upgrade_decision = self._makeUpgradeDecision()
upgrade_decision.setReference("UD-TESTDRAFT")
self.tic()
redirect_url = self.portal.Base_acceptUpgradeDecision('UD-TESTDRAFT')
self.assertTrue(redirect_url.endswith(
"?portal_status_message=Sorry%2C%20the%20upgrade%20is%20not%20possible%20yet%21"),
"%s contains the wrong message" % redirect_url)
def test_tocheckBase_acceptUpgradeDecision_planned_upgrade_decision(self):
upgrade_decision = self._makeUpgradeDecision()
upgrade_decision.setReference("UD-TESTPLANNED")
upgrade_decision.plan()
self.tic()
redirect_url = self.portal.Base_acceptUpgradeDecision('UD-TESTPLANNED')
self.assertTrue(redirect_url.endswith(
"?portal_status_message=Sorry%2C%20the%20upgrade%20is%20not%20possible%20yet%21"),
"%s contains the wrong message" % redirect_url)
def test_tocheckBase_acceptUpgradeDecision_confirmed_upgrade_decision(self):
upgrade_decision = self._makeUpgradeDecision()
upgrade_decision.setReference("UD-TESTCONFIRMED")
upgrade_decision.confirm()
self.tic()
redirect_url = self.portal.Base_acceptUpgradeDecision('UD-TESTCONFIRMED')
self.assertTrue(redirect_url.endswith(
"?portal_status_message=This%20Upgrade%20Decision%20has%20been%20"\
"requested%2C%20it%20will%20be%20processed%20in%20few%20minutes."),
"%s contains the wrong message" % redirect_url)
self.assertEqual(upgrade_decision.getSimulationState(), 'started')
def test_tocheckBase_acceptUpgradeDecision_started_decision(self):
upgrade_decision = self._makeUpgradeDecision()
upgrade_decision.setReference("UD-TESTSTARTED")
upgrade_decision.start()
self.tic()
redirect_url = self.portal.Base_acceptUpgradeDecision('UD-TESTSTARTED')
self.assertTrue(redirect_url.endswith(
"?portal_status_message=This%20Upgrade%20Decision%20is%20already%20Started."),
"%s contains the wrong message" % redirect_url)
def test_tocheckBase_acceptUpgradeDecision_stop_decision(self):
upgrade_decision = self._makeUpgradeDecision()
upgrade_decision.setReference("UD-TESTSTOP")
upgrade_decision.start()
upgrade_decision.stop()
self.tic()
redirect_url = self.portal.Base_acceptUpgradeDecision('UD-TESTSTOP')
self.assertTrue(redirect_url.endswith(
"?portal_status_message=This%20Upgrade%20Decision%20has%20been%20already%20processed."),
"%s contains the wrong message" % redirect_url)
def test_tocheckBase_acceptUpgradeDecision_delivered_decision(self):
upgrade_decision = self._makeUpgradeDecision()
upgrade_decision.setReference("UD-TESTDELIVERED")
upgrade_decision.start()
upgrade_decision.stop()
upgrade_decision.deliver()
self.tic()
redirect_url = self.portal.Base_acceptUpgradeDecision('UD-TESTDELIVERED')
self.assertTrue(redirect_url.endswith(
"?portal_status_message=This%20Upgrade%20Decision%20has%20been%20already%20processed."),
"%s contains the wrong message" % redirect_url)
def test_tocheckBase_acceptUpgradeDecision_cancelled_decision(self):
upgrade_decision = self._makeUpgradeDecision()
upgrade_decision.setReference("UD-TESTCANCELLED")
upgrade_decision.cancel()
self.tic()
redirect_url = self.portal.Base_acceptUpgradeDecision('UD-TESTCANCELLED')
self.assertTrue(redirect_url.endswith(
"?portal_status_message=Sorry%2C%20the%20upgrade%20is%20not%20possble%2C%20Upgrade%20Decision%20was%20Canceled%20or%20Rejected%21"),
"%s contains the wrong message" % redirect_url)
def test_tocheckBase_acceptUpgradeDecision_rejected_decision(self):
upgrade_decision = self._makeUpgradeDecision()
upgrade_decision.setReference("UD-TESTREJECT")
upgrade_decision.cancel()
self.tic()
redirect_url = self.portal.Base_acceptUpgradeDecision('UD-TESTREJECT')
self.assertTrue(redirect_url.endswith(
"?portal_status_message=Sorry%2C%20the%20upgrade%20is%20not%20possble%2C%20Upgrade%20Decision%20was%20Canceled%20or%20Rejected%21"),
"%s contains the wrong message" % redirect_url)
def test_tocheckBase_rejectUpgradeDecision_no_reference(self):
self._makeUpgradeDecision()
self.assertRaises(ValueError, self.portal.Base_rejectUpgradeDecision, None)
def test_tocheckBase_rejectUpgradeDecision_duplicated_reference(self):
upgrade_decision = self._makeUpgradeDecision()
upgrade_decision.setReference("UD-TESTBADREFERENCE")
upgrade_decision = self._makeUpgradeDecision()
upgrade_decision.setReference("UD-TESTBADREFERENCE")
self.tic()
self.assertRaises(ValueError, self.portal.Base_acceptUpgradeDecision, None)
def test_tocheckBase_rejectUpgradeDecision_no_upgrade_decision(self):
redirect_url = self.portal.Base_rejectUpgradeDecision('UD-UNEXISTING')
......
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