Skip to content

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

slapos_cloud: Add test for InstanceTree.requestTransfer

parent d107a555
Pipeline #19081 failed with stage
    in 0 seconds
    Hide whitespace changes
    Inline Side-by-side
    Showing with 280 additions and 360 deletions
    master/bt5/slapos_cloud/TestTemplateItem/portal_components/test.erp5.testSlapOSCloudCreateMovementSkins.py deleted 100644 → 0
    View file @ d107a555
    # -*- coding:utf-8 -*-
    ##############################################################################
    #
    # Copyright (c) 2021 Nexedi SA and Contributors. All Rights Reserved.
    #
    # 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
    #
    ##############################################################################
    from erp5.component.test.SlapOSTestCaseMixin import SlapOSTestCaseMixin
    from time import sleep
    from zExceptions import Unauthorized
    class TestInstanceTreecreateMovement(SlapOSTestCaseMixin):
    def _makeInstanceTree(self):
    instance_tree = self.portal.instance_tree_module\
    .template_instance_tree.Base_createCloneDocument(batch_mode=1)
    instance_tree.edit(reference='TESTINTT-%s' % instance_tree.getId(),
    title='TESTINTT-%s' % instance_tree.getId())
    instance_tree.validate()
    self.tic()
    return instance_tree
    def _makeProject(self):
    project = self.portal.project_module.newContent()
    project.edit(reference="TESTPROJ-%s" % project.getId())
    project.validate()
    self.tic()
    return project
    def _makeOrganisation(self):
    organisation = self.portal.organisation_module.newContent()
    organisation.edit(reference="TESTSITE-%s" % organisation.getId())
    organisation.validate()
    self.tic()
    return organisation
    def testUnauthorized(self):
    instance_tree = self._makeInstanceTree()
    self.assertRaises(Unauthorized, instance_tree.InstanceTree_createMovement)
    destination_section = self.makePerson(user=1)
    self.assertEqual(1 , len(destination_section.objectValues( portal_type="ERP5 Login")))
    self.login(destination_section.getUserId())
    self.assertRaises(Unauthorized, instance_tree.InstanceTree_createMovement)
    self.login()
    other_user = self.makePerson(user=1)
    self.assertEqual(1 , len(other_user.objectValues(portal_type="ERP5 Login")))
    instance_tree.setDestinationSectionValue(destination_section)
    self.tic()
    self.assertRaises(Unauthorized, instance_tree.InstanceTree_createMovement)
    self.login(other_user.getUserId())
    self.assertRaises(Unauthorized, instance_tree.InstanceTree_createMovement)
    self.login(destination_section.getUserId())
    self.assertEqual(instance_tree.InstanceTree_createMovement(), None)
    def test_project(self):
    instance_tree = self._makeInstanceTree()
    destination_section = self.makePerson(user=1)
    instance_tree.setDestinationSectionValue(destination_section)
    project = self._makeProject()
    other_project = self._makeProject()
    self.tic()
    self.login(destination_section.getUserId())
    self.assertEqual(instance_tree.Item_getCurrentProjectValue(), None)
    self.assertEqual(instance_tree.Item_getCurrentOwnerValue(), None)
    self.assertEqual(instance_tree.Item_getCurrentSiteValue(), None)
    # Place in a project
    self.assertEqual(instance_tree.InstanceTree_createMovement(
    destination_project=project.getRelativeUrl()), None)
    self.tic()
    self.assertEqual(instance_tree.Item_getCurrentProjectValue(), project)
    self.assertEqual(instance_tree.Item_getCurrentOwnerValue(), destination_section)
    self.assertEqual(instance_tree.Item_getCurrentSiteValue(), None)
    self.assertEqual(1,
    len(instance_tree.getAggregateRelatedList(portal_type="Internal Packing List Line"))
    )
    # Ensure that we don't have 2 new Internal Packing lists in the same second
    sleep(1)
    self.login(destination_section.getUserId())
    # We don't remove from Project if destination project is not provided
    self.assertEqual(instance_tree.InstanceTree_createMovement(), None)
    self.tic()
    self.assertEqual(instance_tree.Item_getCurrentProjectValue(), project)
    self.assertEqual(instance_tree.Item_getCurrentOwnerValue(), destination_section)
    self.assertEqual(2,
    len(instance_tree.getAggregateRelatedList(portal_type="Internal Packing List Line"))
    )
    # Ensure that we don't have 2 new Internal Packing lists in the same second
    sleep(1)
    # Place in another project
    self.assertEqual(instance_tree.InstanceTree_createMovement(
    destination_project=other_project.getRelativeUrl()), None)
    self.tic()
    self.assertEqual(instance_tree.Item_getCurrentProjectValue(), other_project)
    self.assertEqual(instance_tree.Item_getCurrentOwnerValue(), destination_section)
    self.assertEqual(instance_tree.Item_getCurrentSiteValue(), None)
    self.assertEqual(3,
    len(instance_tree.getAggregateRelatedList(portal_type="Internal Packing List Line"))
    )
    self.login(destination_section.getUserId())
    # Ensure that we don't have 2 new Internal Packing lists in the same second
    sleep(1)
    # We don't remove from Project if destination project is not provided
    self.assertEqual(instance_tree.InstanceTree_createMovement(), None)
    self.tic()
    self.assertEqual(instance_tree.Item_getCurrentProjectValue(), other_project)
    self.assertEqual(instance_tree.Item_getCurrentOwnerValue(), destination_section)
    self.assertEqual(instance_tree.Item_getCurrentSiteValue(), None)
    self.assertEqual(4,
    len(instance_tree.getAggregateRelatedList(portal_type="Internal Packing List Line"))
    )
    def test_owner(self):
    instance_tree = self._makeInstanceTree()
    destination_section = self.makePerson(user=1)
    instance_tree.setDestinationSectionValue(destination_section)
    organisation = self._makeOrganisation()
    other_organisation = self._makeOrganisation()
    self.tic()
    self.login(destination_section.getUserId())
    self.assertEqual(instance_tree.Item_getCurrentProjectValue(), None)
    self.assertEqual(instance_tree.Item_getCurrentOwnerValue(), None)
    self.assertEqual(instance_tree.InstanceTree_createMovement(
    destination=organisation.getRelativeUrl()), None)
    self.tic()
    self.assertEqual(instance_tree.Item_getCurrentProjectValue(), None)
    self.assertEqual(instance_tree.Item_getCurrentOwnerValue(), destination_section)
    self.assertEqual(instance_tree.Item_getCurrentSiteValue(), organisation)
    self.assertEqual(1,
    len(instance_tree.getAggregateRelatedList(portal_type="Internal Packing List Line"))
    )
    # Ensure that we don't have 2 new Internal Packing lists in the same second
    sleep(1)
    self.login(destination_section.getUserId())
    self.assertEqual(instance_tree.InstanceTree_createMovement(), None)
    self.tic()
    self.assertEqual(instance_tree.Item_getCurrentProjectValue(), None)
    self.assertEqual(instance_tree.Item_getCurrentOwnerValue(), destination_section)
    self.assertEqual(instance_tree.Item_getCurrentSiteValue(), organisation)
    # Ensure that we don't have 2 new Internal Packing lists in the same second
    sleep(1)
    # Place in another project
    self.assertEqual(instance_tree.InstanceTree_createMovement(
    destination=other_organisation.getRelativeUrl()), None)
    self.tic()
    self.assertEqual(3,
    len(instance_tree.getAggregateRelatedList(portal_type="Internal Packing List Line"))
    )
    self.assertEqual(instance_tree.Item_getCurrentProjectValue(), None)
    self.assertEqual(instance_tree.Item_getCurrentOwnerValue(), destination_section)
    self.assertEqual(instance_tree.Item_getCurrentSiteValue(), other_organisation)
    self.assertEqual(3,
    len(instance_tree.getAggregateRelatedList(portal_type="Internal Packing List Line"))
    )
    self.login(destination_section.getUserId())
    # Ensure that we don't have 2 new Internal Packing lists in the same second
    sleep(1)
    # We don't remove from Project if destination project is not provided
    self.assertEqual(instance_tree.InstanceTree_createMovement(), None)
    self.tic()
    self.assertEqual(instance_tree.Item_getCurrentProjectValue(), None)
    self.assertEqual(instance_tree.Item_getCurrentOwnerValue(), destination_section)
    self.assertEqual(instance_tree.Item_getCurrentSiteValue(), other_organisation)
    self.assertEqual(4,
    len(instance_tree.getAggregateRelatedList(portal_type="Internal Packing List Line"))
    )
    master/bt5/slapos_cloud/TestTemplateItem/portal_components/test.erp5.testSlapOSCloudCreateMovementSkins.xml deleted 100644 → 0
    View file @ d107a555
    <?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>_recorded_property_dict</string> </key>
    <value>
    <persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent>
    </value>
    </item>
    <item>
    <key> <string>default_reference</string> </key>
    <value> <string>testSlapOSCloudCreateMovementSkins</string> </value>
    </item>
    <item>
    <key> <string>default_source_reference</string> </key>
    <value>
    <none/>
    </value>
    </item>
    <item>
    <key> <string>description</string> </key>
    <value>
    <none/>
    </value>
    </item>
    <item>
    <key> <string>id</string> </key>
    <value> <string>test.erp5.testSlapOSCloudCreateMovementSkins</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/>
    </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">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>
    <pickle>
    <dictionary>
    <item>
    <key> <string>data</string> </key>
    <value>
    <dictionary>
    <item>
    <key> <string>component_validation_workflow</string> </key>
    <value>
    <persistent> <string encoding="base64">AAAAAAAAAAQ=</string> </persistent>
    </value>
    </item>
    </dictionary>
    </value>
    </item>
    </dictionary>
    </pickle>
    </record>
    <record id="4" aka="AAAAAAAAAAQ=">
    <pickle>
    <global name="WorkflowHistoryList" module="Products.ERP5Type.Workflow"/>
    </pickle>
    <pickle>
    <dictionary>
    <item>
    <key> <string>_log</string> </key>
    <value>
    <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>
    </value>
    </item>
    </dictionary>
    </pickle>
    </record>
    </ZopeData>
    master/bt5/slapos_cloud/TestTemplateItem/portal_components/test.erp5.testSlapOSCloudInstanceSlapInterfaceWorkflow.py
    View file @ 82a3b8c6
    ......@@ -21,6 +21,8 @@
    from erp5.component.test.SlapOSTestCaseMixin import SlapOSTestCaseMixin
    import transaction
    from unittest import expectedFailure
    from time import sleep
    from zExceptions import Unauthorized
    class TestSlapOSCoreInstanceSlapInterfaceWorkflow(SlapOSTestCaseMixin):
    """Tests instance.requestInstance"""
    ......@@ -859,3 +861,280 @@ class TestSlapOSCoreInstanceSlapInterfaceWorkflow(SlapOSTestCaseMixin):
    transaction.commit()
    self.assertEqual(bang_amount+1, self._countBang(self.software_instance))
    class TestSlapOSCoreInstanceSlapInterfaceWorkflowTransfer(SlapOSTestCaseMixin):
    """Tests instance.requestTransfer"""
    def afterSetUp(self):
    SlapOSTestCaseMixin.afterSetUp(self)
    portal = self.getPortalObject()
    new_id = self.generateNewId()
    self.request_kw = dict(
    software_release=self.generateNewSoftwareReleaseUrl(),
    software_title=self.generateNewSoftwareTitle(),
    software_type=self.generateNewSoftwareType(),
    instance_xml=self.generateSafeXml(),
    sla_xml=self.generateSafeXml(),
    shared=False,
    state="started"
    )
    # prepare part of tree
    self.instance_tree = portal.instance_tree_module\
    .template_instance_tree.Base_createCloneDocument(batch_mode=1)
    self.software_instance = portal.software_instance_module\
    .template_software_instance.Base_createCloneDocument(batch_mode=1)
    self.instance_tree.edit(
    title=self.request_kw['software_title'],
    reference="TESTHS-%s" % new_id,
    url_string=self.request_kw['software_release'],
    source_reference=self.request_kw['software_type'],
    text_content=self.request_kw['instance_xml'],
    sla_xml=self.request_kw['sla_xml'],
    root_slave=self.request_kw['shared'],
    successor=self.software_instance.getRelativeUrl()
    )
    self.instance_tree.validate()
    self.portal.portal_workflow._jumpToStateFor(self.instance_tree, 'start_requested')
    self.software_instance.edit(
    title=self.request_kw['software_title'],
    reference="TESTSI-%s" % new_id,
    url_string=self.request_kw['software_release'],
    source_reference=self.request_kw['software_type'],
    text_content=self.request_kw['instance_xml'],
    sla_xml=self.request_kw['sla_xml'],
    specialise=self.instance_tree.getRelativeUrl()
    )
    self.portal.portal_workflow._jumpToStateFor(self.software_instance, 'start_requested')
    self.software_instance.validate()
    self.tic()
    person_user = self.makePerson()
    self.tic()
    # Login as new user
    self.login(person_user.getUserId())
    new_person = self.portal.portal_membership.getAuthenticatedMember().getUserValue()
    self.assertEqual(person_user.getRelativeUrl(), new_person.getRelativeUrl())
    def beforeTearDown(self):
    transaction.abort()
    if 'request_instance' in self.software_instance.REQUEST:
    self.software_instance.REQUEST['request_instance'] = None
    def _makeProject(self):
    project = self.portal.project_module.newContent()
    project.edit(reference="TESTPROJ-%s" % project.getId())
    project.validate()
    self.tic()
    return project
    def _makeOrganisation(self):
    organisation = self.portal.organisation_module.newContent()
    organisation.edit(reference="TESTSITE-%s" % organisation.getId())
    organisation.validate()
    self.tic()
    return organisation
    def test_RequesterInstance_requestTransfer_Unauthorized(self):
    destination_section = self.portal.portal_membership.getAuthenticatedMember().getUserValue()
    self.login()
    self.assertRaises(Unauthorized, self.instance_tree.requestTransfer,
    destination=None,
    destination_project=None)
    self.login(destination_section.getUserId())
    self.assertRaises(Unauthorized, self.instance_tree.requestTransfer,
    destination=None,
    destination_project=None)
    self.login()
    other_user = self.makePerson(user=1)
    self.assertEqual(1 , len(other_user.objectValues(portal_type="ERP5 Login")))
    self.instance_tree.setDestinationSectionValue(destination_section)
    self.tic()
    self.assertRaises(Unauthorized, self.instance_tree.requestTransfer,
    destination=None,
    destination_project=None)
    self.login(other_user.getUserId())
    self.assertRaises(Unauthorized, self.instance_tree.requestTransfer,
    destination=None,
    destination_project=None)
    self.login(destination_section.getUserId())
    self.assertEqual(self.instance_tree.requestTransfer(
    destination=None,
    destination_project=None), None)
    def test_RequesterInstance_requestTransfer_project(self):
    destination_section = self.portal.portal_membership.getAuthenticatedMember().getUserValue()
    self.instance_tree.setDestinationSectionValue(destination_section)
    self.login()
    project = self._makeProject()
    other_project = self._makeProject()
    self.tic()
    self.login(destination_section.getUserId())
    self.assertEqual(self.instance_tree.Item_getCurrentProjectValue(), None)
    self.assertEqual(self.instance_tree.Item_getCurrentOwnerValue(), None)
    self.assertEqual(self.instance_tree.Item_getCurrentSiteValue(), None)
    # Place in a project
    self.assertEqual(self.instance_tree.requestTransfer(
    destination=None,
    destination_project=project.getRelativeUrl()), None)
    self.tic()
    self.assertEqual(
    self.instance_tree.Item_getCurrentProjectValue(), project)
    self.assertEqual(
    self.instance_tree.Item_getCurrentOwnerValue(), destination_section)
    self.assertEqual(
    self.instance_tree.Item_getCurrentSiteValue(), None)
    self.assertEqual(1,
    len(self.instance_tree.getAggregateRelatedList(portal_type="Internal Packing List Line"))
    )
    # Ensure that we don't have 2 new Internal Packing lists in the same second
    sleep(1)
    self.login(destination_section.getUserId())
    # We don't remove from Project if destination project is not provided
    self.assertEqual(self.instance_tree.requestTransfer(
    destination=None,
    destination_project=None), None)
    self.tic()
    self.assertEqual(self.instance_tree.Item_getCurrentProjectValue(), project)
    self.assertEqual(self.instance_tree.Item_getCurrentOwnerValue(), destination_section)
    self.assertEqual(2,
    len(self.instance_tree.getAggregateRelatedList(portal_type="Internal Packing List Line"))
    )
    # Ensure that we don't have 2 new Internal Packing lists in the same second
    sleep(1)
    # Place in another project
    self.assertEqual(self.instance_tree.requestTransfer(
    destination=None,
    destination_project=other_project.getRelativeUrl()), None)
    self.tic()
    self.assertEqual(self.instance_tree.Item_getCurrentProjectValue(), other_project)
    self.assertEqual(self.instance_tree.Item_getCurrentOwnerValue(), destination_section)
    self.assertEqual(self.instance_tree.Item_getCurrentSiteValue(), None)
    self.assertEqual(3,
    len(self.instance_tree.getAggregateRelatedList(portal_type="Internal Packing List Line"))
    )
    self.login(destination_section.getUserId())
    # Ensure that we don't have 2 new Internal Packing lists in the same second
    sleep(1)
    # We don't remove from Project if destination project is not provided
    self.assertEqual(self.instance_tree.requestTransfer(
    destination_project=None,
    destination=None), None)
    self.tic()
    self.assertEqual(self.instance_tree.Item_getCurrentProjectValue(), other_project)
    self.assertEqual(self.instance_tree.Item_getCurrentOwnerValue(), destination_section)
    self.assertEqual(self.instance_tree.Item_getCurrentSiteValue(), None)
    self.assertEqual(4,
    len(self.instance_tree.getAggregateRelatedList(portal_type="Internal Packing List Line"))
    )
    def test_RequesterInstance_requestTransfer_owner(self):
    destination_section = self.portal.portal_membership.getAuthenticatedMember().getUserValue()
    self.instance_tree.setDestinationSectionValue(destination_section)
    self.login()
    organisation = self._makeOrganisation()
    other_organisation = self._makeOrganisation()
    self.tic()
    self.login(destination_section.getUserId())
    self.assertEqual(self.instance_tree.Item_getCurrentProjectValue(), None)
    self.assertEqual(self.instance_tree.Item_getCurrentOwnerValue(), None)
    self.assertEqual(self.instance_tree.requestTransfer(
    destination=organisation.getRelativeUrl(),
    destination_project=None), None)
    self.tic()
    self.assertEqual(self.instance_tree.Item_getCurrentProjectValue(), None)
    self.assertEqual(self.instance_tree.Item_getCurrentOwnerValue(), destination_section)
    self.assertEqual(self.instance_tree.Item_getCurrentSiteValue(), organisation)
    self.assertEqual(1,
    len(self.instance_tree.getAggregateRelatedList(portal_type="Internal Packing List Line"))
    )
    # Ensure that we don't have 2 new Internal Packing lists in the same second
    sleep(1)
    self.login(destination_section.getUserId())
    self.assertEqual(self.instance_tree.requestTransfer(
    destination_project=None,
    destination=None), None)
    self.tic()
    self.assertEqual(self.instance_tree.Item_getCurrentProjectValue(), None)
    self.assertEqual(self.instance_tree.Item_getCurrentOwnerValue(), destination_section)
    self.assertEqual(self.instance_tree.Item_getCurrentSiteValue(), organisation)
    # Ensure that we don't have 2 new Internal Packing lists in the same second
    sleep(1)
    # Place in another project
    self.assertEqual(self.instance_tree.requestTransfer(
    destination_project=None,
    destination=other_organisation.getRelativeUrl()), None)
    self.tic()
    self.assertEqual(3,
    len(self.instance_tree.getAggregateRelatedList(portal_type="Internal Packing List Line"))
    )
    self.assertEqual(self.instance_tree.Item_getCurrentProjectValue(), None)
    self.assertEqual(self.instance_tree.Item_getCurrentOwnerValue(), destination_section)
    self.assertEqual(self.instance_tree.Item_getCurrentSiteValue(), other_organisation)
    self.assertEqual(3,
    len(self.instance_tree.getAggregateRelatedList(portal_type="Internal Packing List Line"))
    )
    self.login(destination_section.getUserId())
    # Ensure that we don't have 2 new Internal Packing lists in the same second
    sleep(1)
    # We don't remove from Project if destination project is not provided
    self.assertEqual(self.instance_tree.requestTransfer(
    destination_project=None,
    destination=None), None)
    self.tic()
    self.assertEqual(self.instance_tree.Item_getCurrentProjectValue(), None)
    self.assertEqual(self.instance_tree.Item_getCurrentOwnerValue(), destination_section)
    self.assertEqual(self.instance_tree.Item_getCurrentSiteValue(), other_organisation)
    self.assertEqual(4,
    len(self.instance_tree.getAggregateRelatedList(portal_type="Internal Packing List Line"))
    )
    master/bt5/slapos_cloud/bt/template_test_id_list
    View file @ 82a3b8c6
    ......@@ -13,5 +13,4 @@ test.erp5.testSlapOSCloudConstraint
    test.erp5.testSlapOSCloudUpgrader
    test.erp5.testSlapOSCloudShadow
    test.erp5.SlapOSTestCaseMixin
    test.erp5.SlapOSTestCaseDefaultScenarioMixin
    test.erp5.testSlapOSCloudCreateMovementSkins
    \ No newline at end of file
    test.erp5.SlapOSTestCaseDefaultScenarioMixin
    \ 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