xos/synchronizer/model_policies/test_model_policy_vegtenant.py

# Copyright 2017-present Open Networking Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


import unittest
from mock import patch, call, Mock, PropertyMock
import mock

import os, sys

test_path=os.path.abspath(os.path.dirname(os.path.realpath(__file__)))
service_dir=os.path.join(test_path, "../../../..")
xos_dir=os.path.join(test_path, "../../..")
if not os.path.exists(os.path.join(test_path, "new_base")):
    xos_dir=os.path.join(test_path, "../../../../../../orchestration/xos/xos")
    services_dir=os.path.join(xos_dir, "../../xos_services")

# While transitioning from static to dynamic load, the path to find neighboring xproto files has changed. So check
# both possible locations...
def get_models_fn(service_name, xproto_name):
    name = os.path.join(service_name, "xos", xproto_name)
    if os.path.exists(os.path.join(services_dir, name)):
        return name
    else:
        name = os.path.join(service_name, "xos", "synchronizer", "models", xproto_name)
        if os.path.exists(os.path.join(services_dir, name)):
            return name
    raise Exception("Unable to find service=%s xproto=%s" % (service_name, xproto_name))

class TestModelPolicyVsgTenant(unittest.TestCase):
    def setUp(self):
        global VEGTenantPolicy, LeastLoadedNodeScheduler, MockObjectList

        self.sys_path_save = sys.path
        sys.path.append(xos_dir)
        sys.path.append(os.path.join(xos_dir, 'synchronizers', 'new_base'))

        config = os.path.join(test_path, "test_config.yaml")
        from xosconfig import Config
        Config.clear()
        Config.init(config, 'synchronizer-config-schema.yaml')

        from synchronizers.new_base.mock_modelaccessor_build import build_mock_modelaccessor
        build_mock_modelaccessor(xos_dir, services_dir, [get_models_fn("vEG", "veg.xproto"),
                                                         get_models_fn("addressmanager", "addressmanager.xproto")])

        import synchronizers.new_base.modelaccessor
        import synchronizers.new_base.model_policies.model_policy_tenantwithcontainer
        import model_policy_vegtenant
        from model_policy_vegtenant import VEGTenantPolicy, model_accessor
        from synchronizers.new_base.model_policies.model_policy_tenantwithcontainer import LeastLoadedNodeScheduler

        from mock_modelaccessor import MockObjectList

        # import all class names to globals
        for (k, v) in model_accessor.all_model_classes.items():
            globals()[k] = v

        # Some of the functions we call have side-effects. For example, creating a VEGTenant may lead to creation of
        # tags. Ideally, this wouldn't happen, but it does. So make sure we reset the world.
        model_accessor.reset_all_object_stores()

        # attic functions that are not present in the mock model accessor
        VEGTenant.volt = PropertyMock(return_value = None)
        AddressManagerServiceInstance.set_attribute = Mock()

        self.policy = VEGTenantPolicy()
        self.tenant = VEGTenant()
        self.user = User(email="testadmin@test.org")
        self.tenant = VEGTenant(creator=self.user, id=1)
        self.flavor = Flavor(name="m1.small")
        self.npt_ctag = NetworkParameterType(name="c_tag", id=1)
        self.npt_stag = NetworkParameterType(name="s_tag", id=2)
        self.npt_neutron_port_name = NetworkParameterType(name="neutron_port_name", id=3)
        self.node = Node(hostname="my.node.com")
        self.slice = Slice(name="mysite_test1", default_flavor=self.flavor, default_isolation="vm")
        self.priv_template = NetworkTemplate(name="access_network", visibility="private")
        self.priv_network = Network(name="mysite_test1_private", template=self.priv_template)
        self.image = Image(name="trusty-server-multi-nic")
        self.deployment = Deployment(name="testdeployment")
        Tag.objects.item_list = []

    def tearDownb(self):
        sys.path = self.sys_path_save

    def test_handle_create(self):
        with patch.object(VEGTenantPolicy, "manage_container") as manage_container, \
             patch.object(VEGTenantPolicy, "manage_address_service_instance") as manage_address_service_instance, \
             patch.object(VEGTenantPolicy, "cleanup_orphans") as cleanup_orphans:
            self.policy.handle_create(self.tenant)
            manage_container.assert_called_with(self.tenant)
            manage_address_service_instance.assert_called_with(self.tenant)
            cleanup_orphans.assert_called_with(self.tenant)

    def test_handle_update(self):
        with patch.object(VEGTenantPolicy, "manage_container") as manage_container, \
             patch.object(VEGTenantPolicy, "manage_address_service_instance") as manage_address_service_instance, \
             patch.object(VEGTenantPolicy, "cleanup_orphans") as cleanup_orphans:
            self.policy.handle_create(self.tenant)
            manage_container.assert_called_with(self.tenant)
            manage_address_service_instance.assert_called_with(self.tenant)
            cleanup_orphans.assert_called_with(self.tenant)

    def test_handle_delete_asi_exist(self):
        with patch.object(AddressManagerServiceInstance, "delete") as amsi_delete:
            vrtenant = AddressManagerServiceInstance()
            self.tenant.address_service_instance = vrtenant
            self.policy.handle_delete(self.tenant)
            # The delete model policy should be called.
            #   NOTE: This will change when vEG is brought up to compliance with R-CORD vSG. See comments in VSG
            #         model policy test case.
            amsi_delete.assert_called()

    def test_handle_delete_asi_noexist(self):
        with patch.object(AddressManagerServiceInstance, "delete") as amsi_delete:
            self.tenant.address_service_instance = None
            self.policy.handle_delete(self.tenant)
            amsi_delete.assert_not_called()

# TODO: vEG doesn't currently implement this functionality

#    def test_handle_delete_cleaup_instance(self):
#        with patch.object(VEGTenant.objects, "get_items") as vegtenant_objects, \
#             patch.object(Instance.objects, "get_items") as instance_objects, \
#             patch.object(Instance, "delete") as instance_delete:
#            veg_service = VEGService()
#            instance = Instance(id=1)
#            instance_objects.return_value = [instance]
#            self.tenant.address_service_instance = None
#            self.tenant.instance = instance
#            self.tenant.instance_id = instance.id
#            self.tenant.owner = veg_service
#            vegtenant_objects.return_value = [self.tenant]
#            self.policy.handle_delete(self.tenant)
#            instance_delete.assert_called()

# TODO: vEG doesn't currently implement this functionality

#    def test_handle_delete_cleanup_instance_live(self):
#        with patch.object(VEGTenant.objects, "get_items") as vegtenant_objects, \
#             patch.object(Instance.objects, "get_items") as instance_objects, \
#             patch.object(Instance, "delete") as instance_delete:
#            # Make sure if an Instance still has active VSG Tenants, that we don't clean it up
#            veg_service = VEGService()
#            instance = Instance(id=1)
#            instance_objects.return_value = [instance]
#            self.tenant.address_service_instance = None
#            self.tenant.instance = instance
#            self.tenant.instance_id = instance.id
#            self.tenant.owner = veg_service
#
#            other_tenant = VEGTenant()
#            other_tenant.address_service_instance = None
#            other_tenant.instance = instance
#            other_tenant.instance_id = instance.id
#            other_tenant.owner = veg_service
#
#            vegtenant_objects.return_value = [self.tenant, other_tenant]
#
#            self.policy.handle_delete(self.tenant)
#            instance_delete.assert_not_called()


# TODO: vEG doesn't currently implement this functionality

#    def test_handle_delete_cleanup_instance_and_tags_and_stuff(self):
#        with patch.object(ServiceInstance.objects, "get_items") as si_objects, \
#             patch.object(AddressManagerServiceInstance.objects, "get_items") as amsi_objects, \
#             patch.object(Tag.objects, "get_items") as tag_objects, \
#             patch.object(VEGTenant.objects, "get_items") as vegtenant_objects, \
#             patch.object(Instance.objects, "get_items") as instance_objects, \
#             patch.object(AddressManagerServiceInstance, "delete") as amsi_delete, \
#             patch.object(Tag, "delete") as tag_delete, \
#             patch.object(Instance, "delete") as instance_delete:
#            veg_service = VEGService()
#            am_instance = AddressManagerServiceInstance()
#            amsi_objects.return_value = [am_instance]
#            si_objects.return_value = [am_instance]  # AddressManagerServiceInstance is a subclass of ServiceInstance
#            instance = Instance(id=1)
#            instance_objects.return_value = [instance]
#            self.tenant.address_service_instance = None
#            self.tenant.instance = instance
#            self.tenant.instance_id = instance.id
#            self.tenant.owner = veg_service
#            vegtenant_objects.return_value = [self.tenant]
#            stag_tag = Tag(service_id=self.tenant.owner.id, content_type=instance.self_content_type_id,
#                           object_id=instance.id, name="s_tag")
#            vrouter_tag = Tag(service_id=self.tenant.owner.id, content_type=instance.self_content_type_id,
#                           object_id=instance.id, name="vm_vrouter_tenant", value=am_instance.id)
#            tag_objects.return_value = [stag_tag, vrouter_tag]
#            self.policy.handle_delete(self.tenant)
#            instance_delete.assert_called()
#            assert stag_tag.delete.called
#            assert vrouter_tag.delete.called
#            assert am_instance.delete.called

    def test_cleanup_orphans(self):
        with patch.object(AddressManagerServiceInstance.objects, "get_items") as amsi_objects, \
             patch.object(AddressManagerServiceInstance, "delete") as amsi_delete:
            vrtenant = AddressManagerServiceInstance(id=1)
            self.tenant.address_service_instance = vrtenant
            some_other_vrtenant = AddressManagerServiceInstance(id=2)
            link = ServiceInstanceLink(subscriber_service_instance=self.tenant, provider_service_instance=some_other_vrtenant)
            self.tenant.subscribed_links = MockObjectList(initial=[link])
            amsi_objects.return_value = [some_other_vrtenant]
            self.policy.cleanup_orphans(self.tenant)
            amsi_delete.assert_called()

    def test_find_instance_for_s_tag_noexist(self):
        with patch.object(Tag.objects, "get_items") as tag_objects:
            tag_objects.filter.return_value = []
            instance = self.policy.find_instance_for_s_tag(3)
            self.assertEqual(instance, None)

    def test_find_instance_for_s_tag(self):
        with patch.object(Tag, "objects") as tag_objects:
            tagged_instance = Instance()
            tag = Tag(content_object = tagged_instance)
            tag_objects.filter.return_value = [tag]
            instance = self.policy.find_instance_for_s_tag(3)
            self.assertEqual(instance, tagged_instance)

    def test_manage_container_no_volt(self):
        with self.assertRaises(Exception) as e:
            self.policy.manage_container(self.tenant)
        self.assertEqual(e.exception.message, "This VEG container has no volt")

    def test_manage_container_noinstance(self):
        with patch.object(VEGTenantPolicy, "find_or_make_instance_for_s_tag") as find_or_make_instance_for_s_tag, \
             patch.object(VEGTenant, "save") as tenant_save, \
             patch.object(VEGTenant, "volt") as volt:
            instance = Instance()
            volt.s_tag=222
            volt.c_tag=111
            find_or_make_instance_for_s_tag.return_value = instance
            self.policy.manage_container(self.tenant)
            self.assertEqual(self.tenant.instance, instance)
            tenant_save.assert_called()

    def test_manage_container_hasinstance(self):
        with patch.object(VEGTenantPolicy, "find_or_make_instance_for_s_tag") as find_or_make_instance_for_s_tag, \
             patch.object(VEGTenant, "save") as tenant_save, \
             patch.object(VEGTenant, "volt") as volt:
            instance = Instance()
            volt.s_tag=222
            volt.c_tag=111
            self.tenant.instance = instance
            self.policy.manage_container(self.tenant)
            find_or_make_instance_for_s_tag.assert_not_called()
            self.assertEqual(self.tenant.instance, instance)
            tenant_save.assert_not_called()

    def test_manage_container_deleted(self):
        with patch.object(VEGTenantPolicy, "find_or_make_instance_for_s_tag") as find_or_make_instance_for_s_tag, \
             patch.object(VEGTenant, "save") as tenant_save, \
             patch.object(VEGTenant, "volt") as volt:
            self.tenant.deleted = True
            self.policy.manage_container(self.tenant)
            find_or_make_instance_for_s_tag.assert_not_called()
            tenant_save.assert_not_called()

    def test_find_or_make_port_noexist(self):
        with patch.object(Port, "save") as port_save, \
             patch.object(Port, "objects") as port_objects:
            instance = Instance(id=123)
            network = Instance(id=456)
            port_objects.filter.return_value = []
            port=self.policy.find_or_make_port(instance, network)
            self.assertNotEqual(port, None)
            port_save.assert_called()

    def test_find_or_make_port_exists(self):
        with patch.object(Port, "save") as port_save, \
             patch.object(Port, "objects") as port_objects:
            someport = Port()
            def mock_port_filter(network_id, instance_id):
                if (network_id==456) and (instance_id==123):
                    return [someport]
                return None
            instance = Instance(id=123)
            network = Instance(id=456)
            port_objects.filter.side_effect = mock_port_filter
            port=self.policy.find_or_make_port(instance, network)
            self.assertEqual(port, someport)
            port_save.assert_not_called()

    def test_get_lan_network_noexist(self):
        with patch.object(VEGService.objects, "get_items") as vegservice_objects:
            vegservice=VEGService(name="myvegservice", id=1, slices=MockObjectList(initial=[self.slice]))
            vegservice_objects.return_value = [vegservice]
            self.tenant.owner = vegservice
            self.slice.networks = MockObjectList()
            with self.assertRaises(Exception) as e:
                self.policy.get_lan_network(self.tenant, None)
            self.assertEqual(e.exception.message, "No lan_network")

    def test_get_lan_network(self):
        with patch.object(VEGService.objects, "get_items") as vegservice_objects:
            vegservice=VEGService(name="myvegservice", id=1, slices=MockObjectList(initial=[self.slice]))
            vegservice_objects.return_value = [vegservice]
            self.tenant.owner = vegservice
            self.slice.networks = MockObjectList([self.priv_network])
            lan_network = self.policy.get_lan_network(self.tenant, None)
            self.assertEqual(lan_network, self.priv_network)

    def test_get_lan_network_toomany(self):
        with patch.object(VEGService.objects, "get_items") as vegservice_objects:
            some_other_network = Network(name="mysite_test1_private", template=self.priv_template)
            vegservice=VEGService(name="myvegservice", id=1, slices=MockObjectList(initial=[self.slice]))
            vegservice_objects.return_value = [vegservice]
            self.tenant.owner = vegservice
            self.slice.networks = MockObjectList([self.priv_network, some_other_network])
            with self.assertRaises(Exception) as e:
                lan_network = self.policy.get_lan_network(self.tenant, None)
            self.assertEqual(e.exception.message, "The vEG slice should only have one non-management private network")

    def test_port_set_parameter_noparamexist(self):
        with patch.object(NetworkParameterType.objects, "get_items") as npt_objects:
            npt_objects.return_value = [self.npt_stag]
            port = Port()
            self.policy.port_set_parameter(port, "s_tag", "123")
            self.assertNotEqual(NetworkParameter.objects.all(), [])
            param = NetworkParameter.objects.first()
            self.assertEqual(param.value, "123")
            self.assertEqual(param.parameter, self.npt_stag)

    def test_port_set_parameter_paramexist(self):
        with patch.object(NetworkParameterType.objects, "get_items") as npt_objects, \
             patch.object(NetworkParameter.objects, "get_items") as np_objects:
            port = Port(id=1)
            np_orig = NetworkParameter(parameter_id=self.npt_stag.id, value="456", object_id=port.id, content_type=port.self_content_type_id)
            np_objects.return_value = [np_orig]
            npt_objects.return_value = [self.npt_stag]
            self.policy.port_set_parameter(port, "s_tag", "123")
            self.assertEqual(NetworkParameter.objects.count(), 1)
            param = NetworkParameter.objects.first()
            self.assertEqual(param.value, "123")

    def test_find_or_make_instance_for_s_tag(self):
        with patch.object(NetworkParameterType.objects, "get_items") as npt_objects, \
             patch.object(Node.objects, "get_items") as node_objects, \
             patch.object(Flavor.objects, "get_items") as flavor_objects, \
             patch.object(VEGService.objects, "get_items") as vegservice_objects, \
             patch.object(VEGTenant, "volt") as volt, \
             patch.object(VEGTenant, "save") as tenant_save, \
             patch.object(VEGTenantPolicy, "get_image") as get_image, \
             patch.object(VEGTenantPolicy, "allocate_public_service_instance") as get_psi, \
             patch.object(LeastLoadedNodeScheduler, "pick") as pick, \
             patch.object(Node, "site_deployment") as site_deployment, \
             patch.object(Instance, "save") as instance_save, \
             patch.object(Instance, "delete") as instance_delete, \
             patch.object(VEGTenantPolicy, "port_set_parameter") as port_set_parameter:
            # setup mocks
            vrtenant = AddressManagerServiceInstance(public_ip="1.2.3.4", public_mac="01:02:03:04:05:06")
            vegservice=VEGService(name="myvegservice", id=1, slices=MockObjectList(initial=[self.slice]))
            vegservice_objects.return_value = [vegservice]
            self.tenant.owner = vegservice
            volt.s_tag=222
            volt.c_tag=111
            get_image.return_value = self.image
            get_psi.return_value = vrtenant
            pick.return_value = (self.node, None)
            site_deployment.deployment = self.deployment
            flavor_objects.return_value=[self.flavor]
            node_objects.return_value=[self.node]
            npt_objects.return_value=[self.npt_stag, self.npt_ctag, self.npt_neutron_port_name]
            self.slice.networks = MockObjectList([self.priv_network])
            # done setup mocks

            # call the function under test
            instance = self.policy.find_or_make_instance_for_s_tag(self.tenant, volt.s_tag)

            # make sure Instance was created
            self.assertNotEqual(instance, None)
            self.assertEqual(instance.creator.email, "testadmin@test.org")
            self.assertEqual(instance.image.name, "trusty-server-multi-nic")
            self.assertEqual(instance.flavor.name, "m1.small")
            self.assertEqual(instance.isolation, "vm")
            self.assertEqual(instance.node.hostname, "my.node.com")
            self.assertEqual(instance.slice.name, "mysite_test1")
            self.assertEqual(instance.parent, None)
            instance_save.assert_called()
            instance_delete.assert_not_called()

            # Access Network Port should have tags to c-tag and s-tag
            port = Port.objects.first()
            self.assertEqual(port.instance, instance)
            self.assertEqual(port.network, self.priv_network)
            port_set_parameter.assert_has_calls([mock.call(port, "c_tag", 111),
                                                 mock.call(port, "s_tag", 222),
                                                 mock.call(port, "neutron_port_name", "stag-222")])

            # The instance should be tagged with the s-tag
            tag = Tag.objects.get(name="s_tag")
            self.assertEqual(tag.value, "222")
            self.assertEqual(tag.object_id, instance.id)

            # The instance should have a tag pointing to its address_service_instance
            tag = Tag.objects.get(name="vm_vrouter_tenant")
            self.assertNotEqual(tag.value, vrtenant.id)
            self.assertEqual(tag.object_id, instance.id)

            # Allocate_public_service_instance should have been called
            get_psi.assert_called()

    def test_manage_address_service_instance(self):
        with patch.object(VEGTenantPolicy, "allocate_public_service_instance") as get_psi:
            vrtenant = AddressManagerServiceInstance(public_ip="1.2.3.4", public_mac="01:02:03:04:05:06")
            get_psi.return_value = vrtenant

            self.tenant.vrouter = None

            self.policy.manage_address_service_instance(self.tenant)

            get_psi.assert_called_with(address_pool_name="addresses_veg", subscriber_tenant=self.tenant)

if __name__ == '__main__':
    unittest.main()