blob: 7bfe16050a2c15436b1f3a817e3b80c06fc629a2 [file] [log] [blame]
Scott Baker62c7eaf2018-05-22 15:59:26 -07001
2# Copyright 2017-present Open Networking Foundation
3#
4# Licensed under the Apache License, Version 2.0 (the "License");
5# you may not use this file except in compliance with the License.
6# You may obtain a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15
16import json
17import os
18import sys
19import unittest
20from mock import patch, PropertyMock, ANY, MagicMock
21from unit_test_common import setup_sync_unit_test
22
23def fake_connect_openstack_admin(self, service, required_version=None):
24 return MagicMock()
25
26class TestPullPorts(unittest.TestCase):
27
28 def setUp(self):
29 self.unittest_setup = setup_sync_unit_test(os.path.abspath(os.path.dirname(os.path.realpath(__file__))),
30 globals(),
31 [("openstack", "openstack.xproto")] )
32
33 sys.path.append(os.path.join(os.path.abspath(os.path.dirname(os.path.realpath(__file__))), "../pull_steps"))
34
35 from pull_ports import OpenStackPortPullStep
36 self.step_class = OpenStackPortPullStep
37
38 self.service = OpenStackService()
39 self.site = Site(name="test-site")
40 self.trust_domain = TrustDomain(owner=self.service, name="test-trust")
41 self.flavor = Flavor(name="test-flavor", backend_handle=1114)
42 self.node = Node(name="test-node", backend_handle=1113)
43 self.slice = Slice(name="test-slice", trust_domain=self.trust_domain, backend_handle=1112)
44 self.image = Image(name="test-image", backend_handle=1111)
45 self.net_management = Network(name="management", backend_handle=1115)
46
47 def tearDown(self):
48 sys.path = self.unittest_setup["sys_path_save"]
49
50 def test_pull_records(self):
51 fakeconn = MagicMock()
52 with patch.object(self.step_class, "connect_openstack_admin") as fake_connect_openstack_admin, \
53 patch.object(Port.objects, "get_items") as port_objects, \
54 patch.object(Network.objects, "get_items") as network_objects, \
55 patch.object(OpenStackServiceInstance.objects, "get_items") as ossi_objects, \
56 patch.object(OpenStackService.objects, "get_items") as osi_objects, \
57 patch.object(Port, "save", autospec=True) as port_save:
58 fake_connect_openstack_admin.return_value = fakeconn
59
60 xos_instance = OpenStackServiceInstance(name="test-instance", slice=self.slice, image=self.image,
61 node=self.node, flavor=self.flavor, backend_handle=2112)
62
63 port_objects.return_value = []
64 network_objects.return_value = [self.net_management]
65 ossi_objects.return_value = [xos_instance]
66 osi_objects.return_value = [self.service]
67
68 cn = ControllerNetwork(net_id=self.net_management.backend_handle)
69 self.net_management.controllernetworks = self.unittest_setup["MockObjectList"]([cn])
70
71 fakeconn.network.ports.return_value = [MagicMock(id=2111, device_id=xos_instance.backend_handle,
72 network_id=self.net_management.backend_handle,
73 fixed_ips=[{"ip_address": "1.2.3.4"}],
74 mac_address="11:22:33:44:55:66")]
75
76 step = self.step_class()
77 step.pull_records()
78
79 self.assertEqual(port_save.call_count, 1)
80 saved_port = port_save.call_args[0][0]
81
82 self.assertEqual(saved_port.network, self.net_management)
83 self.assertEqual(saved_port.ip, "1.2.3.4")
84 self.assertEqual(saved_port.mac, "11:22:33:44:55:66")
85 self.assertEqual(saved_port.port_id, 2111)
86
87
88
89if __name__ == '__main__':
90 unittest.main()