blob: 5894f5e10968b27d9034e41935f5cb8a8384833b [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
Matteo Scandolo33523412018-04-12 15:21:13 -070021test_path=os.path.abspath(os.path.dirname(os.path.realpath(__file__)))
Matteo Scandolo33523412018-04-12 15:21:13 -070022
23class TestSyncOLTDevice(unittest.TestCase):
24
25 def setUp(self):
26 global DeferredException
27
28 self.sys_path_save = sys.path
Matteo Scandolo33523412018-04-12 15:21:13 -070029
30 # Setting up the config module
31 from xosconfig import Config
Matteo Scandolof7ebb112018-09-18 16:17:22 -070032 config = os.path.join(test_path, "../test_config.yaml")
Matteo Scandolo33523412018-04-12 15:21:13 -070033 Config.clear()
34 Config.init(config, "synchronizer-config-schema.yaml")
35 # END Setting up the config module
36
Scott Baker47b47302019-01-30 16:55:07 -080037 from xossynchronizer.mock_modelaccessor_build import mock_modelaccessor_config
38 mock_modelaccessor_config(test_path, [("olt-service", "volt.xproto"),
Matteo Scandolo35207b72019-05-10 08:46:48 -070039 ("rcord", "rcord.xproto")])
Matteo Scandolo19466a02018-05-16 17:43:39 -070040
Scott Baker47b47302019-01-30 16:55:07 -080041 import xossynchronizer.modelaccessor
42 reload(xossynchronizer.modelaccessor) # in case nose2 loaded it in a previous test
43
44 from xossynchronizer.modelaccessor import model_accessor
45 self.model_accessor = model_accessor
46
47 from pull_olts import OLTDevicePullStep
Matteo Scandolo33523412018-04-12 15:21:13 -070048
49 # import all class names to globals
50 for (k, v) in model_accessor.all_model_classes.items():
51 globals()[k] = v
52
53 self.sync_step = OLTDevicePullStep
54
55 # mock volt service
56 self.volt_service = Mock()
57 self.volt_service.id = "volt_service_id"
58 self.volt_service.voltha_url = "voltha_url"
59 self.volt_service.voltha_user = "voltha_user"
60 self.volt_service.voltha_pass = "voltha_pass"
Matteo Scandolo19466a02018-05-16 17:43:39 -070061 self.volt_service.voltha_port = 1234
Matteo Scandolo33523412018-04-12 15:21:13 -070062
63 # mock voltha responses
64 self.devices = {
65 "items": [
66 {
67 "id": "test_id",
68 "type": "simulated_olt",
69 "host_and_port": "172.17.0.1:50060",
70 "admin_state": "ENABLED",
Matteo Scandolod6fce512018-10-16 10:35:29 -070071 "oper_status": "ACTIVE",
72 "serial_number": "serial_number",
Matteo Scandolo33523412018-04-12 15:21:13 -070073 }
74 ]
75 }
76
77 self.logical_devices = {
78 "items": [
79 {
80 "root_device_id": "test_id",
81 "id": "of_id",
82 "datapath_id": "55334486016"
83 }
84 ]
85 }
86
Matteo Scandolo6be6ee92018-05-24 15:07:51 -070087 self.ports = {
88 "items": [
89 {
90 "label": "PON port",
91 "port_no": 1,
92 "type": "PON_OLT",
93 "admin_state": "ENABLED",
94 "oper_status": "ACTIVE"
95 },
96 {
97 "label": "NNI facing Ethernet port",
98 "port_no": 2,
99 "type": "ETHERNET_NNI",
100 "admin_state": "ENABLED",
101 "oper_status": "ACTIVE"
102 }
103 ]
104 }
105
Matteo Scandolo33523412018-04-12 15:21:13 -0700106 def tearDown(self):
107 sys.path = self.sys_path_save
108
109 @requests_mock.Mocker()
110 def test_missing_volt_service(self, m):
111 self.assertFalse(m.called)
112
113 @requests_mock.Mocker()
Matteo Scandolo778fef72019-04-22 13:32:36 -0700114 def test_pull_host_and_port(self, m):
Matteo Scandolo33523412018-04-12 15:21:13 -0700115
116 with patch.object(VOLTService.objects, "all") as olt_service_mock, \
Matteo Scandolo778fef72019-04-22 13:32:36 -0700117 patch.object(OLTDevice, "save", autospec=True) as mock_olt_save, \
Matteo Scandolo6be6ee92018-05-24 15:07:51 -0700118 patch.object(PONPort, "save") as mock_pon_save, \
119 patch.object(NNIPort, "save") as mock_nni_save:
Matteo Scandolo33523412018-04-12 15:21:13 -0700120 olt_service_mock.return_value = [self.volt_service]
121
Matteo Scandolo19466a02018-05-16 17:43:39 -0700122 m.get("http://voltha_url:1234/api/v1/devices", status_code=200, json=self.devices)
Matteo Scandolo6be6ee92018-05-24 15:07:51 -0700123 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 -0700124 m.get("http://voltha_url:1234/api/v1/logical_devices", status_code=200, json=self.logical_devices)
Matteo Scandolo33523412018-04-12 15:21:13 -0700125
Scott Baker47b47302019-01-30 16:55:07 -0800126 self.sync_step(model_accessor=self.model_accessor).pull_records()
Matteo Scandolo33523412018-04-12 15:21:13 -0700127
Matteo Scandolo33523412018-04-12 15:21:13 -0700128
Matteo Scandolo778fef72019-04-22 13:32:36 -0700129 saved_olts = mock_olt_save.call_args_list
130 simulated_olt = saved_olts[0][0][0]
131 self.assertEqual(len(saved_olts), 1)
132
133 self.assertEqual(simulated_olt.admin_state, "ENABLED")
134 self.assertEqual(simulated_olt.oper_status, "ACTIVE")
135 self.assertEqual(simulated_olt.volt_service_id, "volt_service_id")
136 self.assertEqual(simulated_olt.device_id, "test_id")
137 self.assertEqual(simulated_olt.of_id, "of_id")
138 self.assertEqual(simulated_olt.dp_id, "of:0000000ce2314000")
139
140
Matteo Scandolo6be6ee92018-05-24 15:07:51 -0700141 mock_pon_save.assert_called()
142 mock_nni_save.assert_called()
Matteo Scandolo33523412018-04-12 15:21:13 -0700143
144 @requests_mock.Mocker()
Matteo Scandolo778fef72019-04-22 13:32:36 -0700145 def test_pull_mac_address(self, m):
146 devices = {
147 "items": [
148 {
149 'id': 'tibit_id',
150 'type': 'tibit_olt',
151 'mac_address': '70:b3:d5:52:30:6f',
152 'admin_state': 'ENABLED',
153 'oper_status': 'ACTIVE',
154 'serial_number': 'OLT-70b3d552306f',
155 }
156 ]
157 }
158
159 logical_devices = {
160 "items": [
161 {
162 "root_device_id": "tibit_id",
163 "id": "of_id",
164 "datapath_id": "55334486017"
165 }
166 ]
167 }
168
169 with patch.object(VOLTService.objects, "all") as olt_service_mock, \
170 patch.object(OLTDevice, "save", autospec=True) as mock_olt_save, \
171 patch.object(PONPort, "save") as mock_pon_save, \
172 patch.object(NNIPort, "save") as mock_nni_save:
173 olt_service_mock.return_value = [self.volt_service]
174
175 m.get("http://voltha_url:1234/api/v1/devices", status_code=200, json=devices)
176 m.get("http://voltha_url:1234/api/v1/devices/tibit_id/ports", status_code=200, json=self.ports)
177 m.get("http://voltha_url:1234/api/v1/logical_devices", status_code=200, json=logical_devices)
178
179
180 self.sync_step(model_accessor=self.model_accessor).pull_records()
181
182
183 saved_olts = mock_olt_save.call_args_list
184 tibit_olt = saved_olts[0][0][0]
185 self.assertEqual(len(saved_olts), 1)
186
187 self.assertEqual(tibit_olt.admin_state, "ENABLED")
188 self.assertEqual(tibit_olt.oper_status, "ACTIVE")
189 self.assertEqual(tibit_olt.volt_service_id, "volt_service_id")
190 self.assertEqual(tibit_olt.device_id, "tibit_id")
191 self.assertEqual(tibit_olt.of_id, "of_id")
192 self.assertEqual(tibit_olt.dp_id, "of:0000000ce2314001")
193
194 mock_pon_save.assert_called()
195 mock_nni_save.assert_called()
196
197
198 @requests_mock.Mocker()
Matteo Scandolo33523412018-04-12 15:21:13 -0700199 def test_pull_existing(self, m):
200
201 existing_olt = Mock()
Scott Baker09798d82019-01-17 08:34:59 -0800202 existing_olt.admin_state = "ENABLED"
Matteo Scandolo33523412018-04-12 15:21:13 -0700203 existing_olt.enacted = 2
204 existing_olt.updated = 1
205
206 with patch.object(VOLTService.objects, "all") as olt_service_mock, \
Matteo Scandolo6be6ee92018-05-24 15:07:51 -0700207 patch.object(OLTDevice.objects, "filter") as mock_get, \
208 patch.object(PONPort, "save") as mock_pon_save, \
209 patch.object(NNIPort, "save") as mock_nni_save, \
210 patch.object(existing_olt, "save") as mock_olt_save:
Matteo Scandolo33523412018-04-12 15:21:13 -0700211 olt_service_mock.return_value = [self.volt_service]
212 mock_get.return_value = [existing_olt]
213
Matteo Scandolo19466a02018-05-16 17:43:39 -0700214 m.get("http://voltha_url:1234/api/v1/devices", status_code=200, json=self.devices)
Matteo Scandolo6be6ee92018-05-24 15:07:51 -0700215 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 -0700216 m.get("http://voltha_url:1234/api/v1/logical_devices", status_code=200, json=self.logical_devices)
Matteo Scandolo33523412018-04-12 15:21:13 -0700217
Scott Baker47b47302019-01-30 16:55:07 -0800218 self.sync_step(model_accessor=self.model_accessor).pull_records()
Matteo Scandolo33523412018-04-12 15:21:13 -0700219
220 self.assertEqual(existing_olt.admin_state, "ENABLED")
221 self.assertEqual(existing_olt.oper_status, "ACTIVE")
222 self.assertEqual(existing_olt.volt_service_id, "volt_service_id")
223 self.assertEqual(existing_olt.device_id, "test_id")
224 self.assertEqual(existing_olt.of_id, "of_id")
225 self.assertEqual(existing_olt.dp_id, "of:0000000ce2314000")
226
Andy Bavier00c573c2019-02-08 16:19:11 -0700227 # mock_olt_save.assert_called()
Matteo Scandolo6be6ee92018-05-24 15:07:51 -0700228 mock_pon_save.assert_called()
229 mock_nni_save.assert_called()
Matteo Scandolo33523412018-04-12 15:21:13 -0700230
231 @requests_mock.Mocker()
232 def test_pull_existing_do_not_sync(self, m):
233 existing_olt = Mock()
234 existing_olt.enacted = 1
235 existing_olt.updated = 2
Matteo Scandolo6be6ee92018-05-24 15:07:51 -0700236 existing_olt.device_id = "test_id"
Matteo Scandolo33523412018-04-12 15:21:13 -0700237
238 with patch.object(VOLTService.objects, "all") as olt_service_mock, \
Matteo Scandolo6be6ee92018-05-24 15:07:51 -0700239 patch.object(OLTDevice.objects, "filter") as mock_get, \
240 patch.object(PONPort, "save") as mock_pon_save, \
241 patch.object(NNIPort, "save") as mock_nni_save, \
242 patch.object(existing_olt, "save") as mock_olt_save:
243
Matteo Scandolo33523412018-04-12 15:21:13 -0700244 olt_service_mock.return_value = [self.volt_service]
Matteo Scandolo6be6ee92018-05-24 15:07:51 -0700245 mock_get.return_value = [existing_olt]
Matteo Scandolo33523412018-04-12 15:21:13 -0700246
Matteo Scandolo19466a02018-05-16 17:43:39 -0700247 m.get("http://voltha_url:1234/api/v1/devices", status_code=200, json=self.devices)
Matteo Scandolo6be6ee92018-05-24 15:07:51 -0700248 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 -0700249 m.get("http://voltha_url:1234/api/v1/logical_devices", status_code=200, json=self.logical_devices)
Matteo Scandolo33523412018-04-12 15:21:13 -0700250
Scott Baker47b47302019-01-30 16:55:07 -0800251 self.sync_step(model_accessor=self.model_accessor).pull_records()
Matteo Scandolo33523412018-04-12 15:21:13 -0700252
Matteo Scandolo6be6ee92018-05-24 15:07:51 -0700253 mock_olt_save.assert_not_called()
254 mock_pon_save.assert_called()
255 mock_nni_save.assert_called()
Matteo Scandolo33523412018-04-12 15:21:13 -0700256
Matteo Scandoloa4fb0322018-08-20 17:35:29 -0700257 @requests_mock.Mocker()
258 def test_pull_deleted_object(self, m):
259 existing_olt = Mock()
260 existing_olt.enacted = 2
261 existing_olt.updated = 1
262 existing_olt.device_id = "test_id"
263
264 m.get("http://voltha_url:1234/api/v1/devices", status_code=200, json={"items": []})
265
266 with patch.object(VOLTService.objects, "all") as olt_service_mock, \
267 patch.object(OLTDevice.objects, "get_items") as mock_get, \
268 patch.object(existing_olt, "delete") as mock_olt_delete:
269
270 olt_service_mock.return_value = [self.volt_service]
271 mock_get.return_value = [existing_olt]
272
Scott Baker47b47302019-01-30 16:55:07 -0800273 self.sync_step(model_accessor=self.model_accessor).pull_records()
Matteo Scandoloa4fb0322018-08-20 17:35:29 -0700274
275 mock_olt_delete.assert_called()
276
Vijaykumar Kushwaha377d1da2019-03-20 10:29:42 +0530277#[SEBA-367] Unit test for blank response recieved from Voltha
278
279 @requests_mock.Mocker()
280 def test_blank_response_received(self, m):
281
282 m.get("http://voltha_url:1234/api/v1/devices", status_code=200, text="")
283 with patch.object(VOLTService.objects, "all") as olt_service_mock, \
284 patch.object(PONPort, "save") as mock_pon_save, \
285 patch.object(NNIPort, "save") as mock_nni_save, \
286 patch.object(OLTDevice.objects, "get_items") as mock_get:
287
288 olt_service_mock.return_value = [self.volt_service]
289
290 self.sync_step(model_accessor=self.model_accessor).pull_records()
291
292 olt_service_mock.assert_called()
293 mock_pon_save.assert_not_called()
294 mock_nni_save.assert_not_called()
295 mock_get.assert_not_called()
296
297#[SEBA-367] Unit test for invalid json received from Voltha
298
299 @requests_mock.Mocker()
300 def test_invalid_json(self, m):
301
302 m.get("http://voltha_url:1234/api/v1/devices", status_code=200, text="{\"items\" : [host_and_port }")
303 with patch.object(VOLTService.objects, "all") as olt_service_mock, \
304 patch.object(PONPort, "save") as mock_pon_save, \
305 patch.object(NNIPort, "save") as mock_nni_save, \
306 patch.object(OLTDevice.objects, "get_items") as mock_get:
307
308 olt_service_mock.return_value = [self.volt_service]
309
310 self.sync_step(model_accessor=self.model_accessor).pull_records()
311
312 olt_service_mock.assert_called()
313 mock_pon_save.assert_not_called()
314 mock_nni_save.assert_not_called()
315 mock_get.assert_not_called()
Matteo Scandoloa4fb0322018-08-20 17:35:29 -0700316
Matteo Scandolo33523412018-04-12 15:21:13 -0700317if __name__ == "__main__":
Vijaykumar Kushwaha377d1da2019-03-20 10:29:42 +0530318 unittest.main()
319