CORD-2964 Implement new openstack models and steps
Change-Id: I32ac438e799f563b721e19ad7ebc8a033796c88e
diff --git a/xos/synchronizer/tests/test_pull_ports.py b/xos/synchronizer/tests/test_pull_ports.py
new file mode 100644
index 0000000..7bfe160
--- /dev/null
+++ b/xos/synchronizer/tests/test_pull_ports.py
@@ -0,0 +1,90 @@
+
+# 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()