blob: 7bfe16050a2c15436b1f3a817e3b80c06fc629a2 [file] [log] [blame]
# 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()