blob: 755f4fb9e1e5c3795e0d7b796f5d6d55778173ca [file] [log] [blame]
Matteo Scandolo33523412018-04-12 15:21:13 -07001# Copyright 2017-present Open Networking Foundation
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14
15import unittest
16from mock import patch, call, Mock, PropertyMock
17import requests_mock
18
19import os, sys
20
21# Hack to load synchronizer framework
22test_path=os.path.abspath(os.path.dirname(os.path.realpath(__file__)))
23xos_dir=os.path.join(test_path, "../../..")
24if not os.path.exists(os.path.join(test_path, "new_base")):
25 xos_dir=os.path.join(test_path, "../../../../../../orchestration/xos/xos")
26 services_dir = os.path.join(xos_dir, "../../xos_services")
27sys.path.append(xos_dir)
28sys.path.append(os.path.join(xos_dir, 'synchronizers', 'new_base'))
29# END Hack to load synchronizer framework
30
31# generate model from xproto
32def get_models_fn(service_name, xproto_name):
33 name = os.path.join(service_name, "xos", xproto_name)
34 if os.path.exists(os.path.join(services_dir, name)):
35 return name
36 else:
37 name = os.path.join(service_name, "xos", "synchronizer", "models", xproto_name)
38 if os.path.exists(os.path.join(services_dir, name)):
39 return name
40 raise Exception("Unable to find service=%s xproto=%s" % (service_name, xproto_name))
41# END generate model from xproto
42
43class TestSyncOLTDevice(unittest.TestCase):
44
45 def setUp(self):
46 global DeferredException
47
48 self.sys_path_save = sys.path
49 sys.path.append(xos_dir)
50 sys.path.append(os.path.join(xos_dir, 'synchronizers', 'new_base'))
51
52 # Setting up the config module
53 from xosconfig import Config
Matteo Scandolof7ebb112018-09-18 16:17:22 -070054 config = os.path.join(test_path, "../test_config.yaml")
Matteo Scandolo33523412018-04-12 15:21:13 -070055 Config.clear()
56 Config.init(config, "synchronizer-config-schema.yaml")
57 # END Setting up the config module
58
59 from synchronizers.new_base.mock_modelaccessor_build import build_mock_modelaccessor
Matteo Scandolo19466a02018-05-16 17:43:39 -070060 # build_mock_modelaccessor(xos_dir, services_dir, [get_models_fn("olt-service", "volt.xproto")])
61
62 # FIXME this is to get jenkins to pass the tests, somehow it is running tests in a different order
63 # and apparently it is not overriding the generated model accessor
64 build_mock_modelaccessor(xos_dir, services_dir, [get_models_fn("olt-service", "volt.xproto"),
65 get_models_fn("vsg", "vsg.xproto"),
66 get_models_fn("../profiles/rcord", "rcord.xproto")])
Matteo Scandolo33523412018-04-12 15:21:13 -070067 import synchronizers.new_base.modelaccessor
68 from pull_olts import OLTDevicePullStep, model_accessor
69
70 # import all class names to globals
71 for (k, v) in model_accessor.all_model_classes.items():
72 globals()[k] = v
73
74 self.sync_step = OLTDevicePullStep
75
76 # mock volt service
77 self.volt_service = Mock()
78 self.volt_service.id = "volt_service_id"
79 self.volt_service.voltha_url = "voltha_url"
80 self.volt_service.voltha_user = "voltha_user"
81 self.volt_service.voltha_pass = "voltha_pass"
Matteo Scandolo19466a02018-05-16 17:43:39 -070082 self.volt_service.voltha_port = 1234
Matteo Scandolo33523412018-04-12 15:21:13 -070083
84 # mock voltha responses
85 self.devices = {
86 "items": [
87 {
88 "id": "test_id",
89 "type": "simulated_olt",
90 "host_and_port": "172.17.0.1:50060",
91 "admin_state": "ENABLED",
Matteo Scandolod6fce512018-10-16 10:35:29 -070092 "oper_status": "ACTIVE",
93 "serial_number": "serial_number",
Matteo Scandolo33523412018-04-12 15:21:13 -070094 }
95 ]
96 }
97
98 self.logical_devices = {
99 "items": [
100 {
101 "root_device_id": "test_id",
102 "id": "of_id",
103 "datapath_id": "55334486016"
104 }
105 ]
106 }
107
Matteo Scandolo6be6ee92018-05-24 15:07:51 -0700108 self.ports = {
109 "items": [
110 {
111 "label": "PON port",
112 "port_no": 1,
113 "type": "PON_OLT",
114 "admin_state": "ENABLED",
115 "oper_status": "ACTIVE"
116 },
117 {
118 "label": "NNI facing Ethernet port",
119 "port_no": 2,
120 "type": "ETHERNET_NNI",
121 "admin_state": "ENABLED",
122 "oper_status": "ACTIVE"
123 }
124 ]
125 }
126
Matteo Scandolo33523412018-04-12 15:21:13 -0700127 def tearDown(self):
128 sys.path = self.sys_path_save
129
130 @requests_mock.Mocker()
131 def test_missing_volt_service(self, m):
132 self.assertFalse(m.called)
133
134 @requests_mock.Mocker()
135 def test_pull(self, m):
136
137 with patch.object(VOLTService.objects, "all") as olt_service_mock, \
Matteo Scandolo6be6ee92018-05-24 15:07:51 -0700138 patch.object(OLTDevice, "save") as mock_olt_save, \
139 patch.object(PONPort, "save") as mock_pon_save, \
140 patch.object(NNIPort, "save") as mock_nni_save:
Matteo Scandolo33523412018-04-12 15:21:13 -0700141 olt_service_mock.return_value = [self.volt_service]
142
Matteo Scandolo19466a02018-05-16 17:43:39 -0700143 m.get("http://voltha_url:1234/api/v1/devices", status_code=200, json=self.devices)
Matteo Scandolo6be6ee92018-05-24 15:07:51 -0700144 m.get("http://voltha_url:1234/api/v1/devices/test_id/ports", status_code=200, json=self.ports)
Matteo Scandolo19466a02018-05-16 17:43:39 -0700145 m.get("http://voltha_url:1234/api/v1/logical_devices", status_code=200, json=self.logical_devices)
Matteo Scandolo33523412018-04-12 15:21:13 -0700146
147 self.sync_step().pull_records()
148
149 # TODO how to asster this?
150 # self.assertEqual(existing_olt.admin_state, "ENABLED")
151 # self.assertEqual(existing_olt.oper_status, "ACTIVE")
152 # self.assertEqual(existing_olt.volt_service_id, "volt_service_id")
153 # self.assertEqual(existing_olt.device_id, "test_id")
154 # self.assertEqual(existing_olt.of_id, "of_id")
155 # self.assertEqual(existing_olt.dp_id, "of:0000000ce2314000")
156
Matteo Scandolo6be6ee92018-05-24 15:07:51 -0700157 mock_olt_save.assert_called()
158 mock_pon_save.assert_called()
159 mock_nni_save.assert_called()
Matteo Scandolo33523412018-04-12 15:21:13 -0700160
161 @requests_mock.Mocker()
162 def test_pull_existing(self, m):
163
164 existing_olt = Mock()
Scott Baker09798d82019-01-17 08:34:59 -0800165 existing_olt.admin_state = "ENABLED"
Matteo Scandolo33523412018-04-12 15:21:13 -0700166 existing_olt.enacted = 2
167 existing_olt.updated = 1
168
169 with patch.object(VOLTService.objects, "all") as olt_service_mock, \
Matteo Scandolo6be6ee92018-05-24 15:07:51 -0700170 patch.object(OLTDevice.objects, "filter") as mock_get, \
171 patch.object(PONPort, "save") as mock_pon_save, \
172 patch.object(NNIPort, "save") as mock_nni_save, \
173 patch.object(existing_olt, "save") as mock_olt_save:
Matteo Scandolo33523412018-04-12 15:21:13 -0700174 olt_service_mock.return_value = [self.volt_service]
175 mock_get.return_value = [existing_olt]
176
Matteo Scandolo19466a02018-05-16 17:43:39 -0700177 m.get("http://voltha_url:1234/api/v1/devices", status_code=200, json=self.devices)
Matteo Scandolo6be6ee92018-05-24 15:07:51 -0700178 m.get("http://voltha_url:1234/api/v1/devices/test_id/ports", status_code=200, json=self.ports)
Matteo Scandolo19466a02018-05-16 17:43:39 -0700179 m.get("http://voltha_url:1234/api/v1/logical_devices", status_code=200, json=self.logical_devices)
Matteo Scandolo33523412018-04-12 15:21:13 -0700180
181 self.sync_step().pull_records()
182
183 self.assertEqual(existing_olt.admin_state, "ENABLED")
184 self.assertEqual(existing_olt.oper_status, "ACTIVE")
185 self.assertEqual(existing_olt.volt_service_id, "volt_service_id")
186 self.assertEqual(existing_olt.device_id, "test_id")
187 self.assertEqual(existing_olt.of_id, "of_id")
188 self.assertEqual(existing_olt.dp_id, "of:0000000ce2314000")
189
Matteo Scandolo6be6ee92018-05-24 15:07:51 -0700190 mock_olt_save.assert_called()
191 mock_pon_save.assert_called()
192 mock_nni_save.assert_called()
Matteo Scandolo33523412018-04-12 15:21:13 -0700193
194 @requests_mock.Mocker()
195 def test_pull_existing_do_not_sync(self, m):
196 existing_olt = Mock()
197 existing_olt.enacted = 1
198 existing_olt.updated = 2
Matteo Scandolo6be6ee92018-05-24 15:07:51 -0700199 existing_olt.device_id = "test_id"
Matteo Scandolo33523412018-04-12 15:21:13 -0700200
201 with patch.object(VOLTService.objects, "all") as olt_service_mock, \
Matteo Scandolo6be6ee92018-05-24 15:07:51 -0700202 patch.object(OLTDevice.objects, "filter") as mock_get, \
203 patch.object(PONPort, "save") as mock_pon_save, \
204 patch.object(NNIPort, "save") as mock_nni_save, \
205 patch.object(existing_olt, "save") as mock_olt_save:
206
Matteo Scandolo33523412018-04-12 15:21:13 -0700207 olt_service_mock.return_value = [self.volt_service]
Matteo Scandolo6be6ee92018-05-24 15:07:51 -0700208 mock_get.return_value = [existing_olt]
Matteo Scandolo33523412018-04-12 15:21:13 -0700209
Matteo Scandolo19466a02018-05-16 17:43:39 -0700210 m.get("http://voltha_url:1234/api/v1/devices", status_code=200, json=self.devices)
Matteo Scandolo6be6ee92018-05-24 15:07:51 -0700211 m.get("http://voltha_url:1234/api/v1/devices/test_id/ports", status_code=200, json=self.ports)
Matteo Scandolo19466a02018-05-16 17:43:39 -0700212 m.get("http://voltha_url:1234/api/v1/logical_devices", status_code=200, json=self.logical_devices)
Matteo Scandolo33523412018-04-12 15:21:13 -0700213
214 self.sync_step().pull_records()
215
Matteo Scandolo6be6ee92018-05-24 15:07:51 -0700216 mock_olt_save.assert_not_called()
217 mock_pon_save.assert_called()
218 mock_nni_save.assert_called()
Matteo Scandolo33523412018-04-12 15:21:13 -0700219
Matteo Scandoloa4fb0322018-08-20 17:35:29 -0700220 @requests_mock.Mocker()
221 def test_pull_deleted_object(self, m):
222 existing_olt = Mock()
223 existing_olt.enacted = 2
224 existing_olt.updated = 1
225 existing_olt.device_id = "test_id"
226
227 m.get("http://voltha_url:1234/api/v1/devices", status_code=200, json={"items": []})
228
229 with patch.object(VOLTService.objects, "all") as olt_service_mock, \
230 patch.object(OLTDevice.objects, "get_items") as mock_get, \
231 patch.object(existing_olt, "delete") as mock_olt_delete:
232
233 olt_service_mock.return_value = [self.volt_service]
234 mock_get.return_value = [existing_olt]
235
236 self.sync_step().pull_records()
237
238 mock_olt_delete.assert_called()
239
240
Matteo Scandolo33523412018-04-12 15:21:13 -0700241if __name__ == "__main__":
242 unittest.main()