xos/synchronizer/tests/test_pull_ports.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 json
import os
import sys
import unittest
from mock import patch, PropertyMock, ANY, MagicMock
from unit_test_common import setup_sync_unit_test

def fake_connect_openstack_admin(self, service, required_version=None):
    return MagicMock()

class TestPullPorts(unittest.TestCase):

    def setUp(self):
        self.unittest_setup = setup_sync_unit_test(os.path.abspath(os.path.dirname(os.path.realpath(__file__))),
                                                   globals(),
                                                   [("openstack", "openstack.xproto")] )

        sys.path.append(os.path.join(os.path.abspath(os.path.dirname(os.path.realpath(__file__))), "../pull_steps"))

        from pull_ports import OpenStackPortPullStep
        self.step_class = OpenStackPortPullStep

        self.service = OpenStackService()
        self.site = Site(name="test-site")
        self.trust_domain = TrustDomain(owner=self.service, name="test-trust")
        self.flavor = Flavor(name="test-flavor", backend_handle=1114)
        self.node = Node(name="test-node", backend_handle=1113)
        self.slice = Slice(name="test-slice", trust_domain=self.trust_domain, backend_handle=1112)
        self.image = Image(name="test-image", backend_handle=1111)
        self.net_management = Network(name="management", backend_handle=1115)

    def tearDown(self):
        sys.path = self.unittest_setup["sys_path_save"]

    def test_pull_records(self):
        fakeconn = MagicMock()
        with patch.object(self.step_class, "connect_openstack_admin") as fake_connect_openstack_admin, \
             patch.object(Port.objects, "get_items") as port_objects, \
             patch.object(Network.objects, "get_items") as network_objects, \
             patch.object(OpenStackServiceInstance.objects, "get_items") as ossi_objects, \
             patch.object(OpenStackService.objects, "get_items") as osi_objects, \
             patch.object(Port, "save", autospec=True) as port_save:
            fake_connect_openstack_admin.return_value = fakeconn

            xos_instance = OpenStackServiceInstance(name="test-instance", slice=self.slice, image=self.image,
                                                    node=self.node, flavor=self.flavor, backend_handle=2112)

            port_objects.return_value = []
            network_objects.return_value = [self.net_management]
            ossi_objects.return_value = [xos_instance]
            osi_objects.return_value = [self.service]

            cn = ControllerNetwork(net_id=self.net_management.backend_handle)
            self.net_management.controllernetworks = self.unittest_setup["MockObjectList"]([cn])

            fakeconn.network.ports.return_value = [MagicMock(id=2111, device_id=xos_instance.backend_handle,
                                                             network_id=self.net_management.backend_handle,
                                                             fixed_ips=[{"ip_address": "1.2.3.4"}],
                                                             mac_address="11:22:33:44:55:66")]

            step = self.step_class()
            step.pull_records()

            self.assertEqual(port_save.call_count, 1)
            saved_port = port_save.call_args[0][0]

            self.assertEqual(saved_port.network, self.net_management)
            self.assertEqual(saved_port.ip, "1.2.3.4")
            self.assertEqual(saved_port.mac, "11:22:33:44:55:66")
            self.assertEqual(saved_port.port_id, 2111)



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