blob: 611e084fcd5f1988352b5e6de35863fd8d91a2c0 [file] [log] [blame]
Matteo Scandolod44ca992018-05-17 15:02:10 -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
Scott Baker22b46c52018-11-15 15:15:29 -080015import functools
Matteo Scandolod44ca992018-05-17 15:02:10 -070016import unittest
17from mock import patch, call, Mock, PropertyMock
18import requests_mock
19
20import os, sys
21
Matteo Scandolod44ca992018-05-17 15:02:10 -070022test_path=os.path.abspath(os.path.dirname(os.path.realpath(__file__)))
Matteo Scandolod44ca992018-05-17 15:02:10 -070023
24class TestPullONUDevice(unittest.TestCase):
25
26 def setUp(self):
27 global DeferredException
28
29 self.sys_path_save = sys.path
Matteo Scandolod44ca992018-05-17 15:02:10 -070030
31 # Setting up the config module
32 from xosconfig import Config
Matteo Scandolof7ebb112018-09-18 16:17:22 -070033 config = os.path.join(test_path, "../test_config.yaml")
Matteo Scandolod44ca992018-05-17 15:02:10 -070034 Config.clear()
35 Config.init(config, "synchronizer-config-schema.yaml")
36 # END Setting up the config module
37
Scott Baker47b47302019-01-30 16:55:07 -080038 from xossynchronizer.mock_modelaccessor_build import mock_modelaccessor_config
39 mock_modelaccessor_config(test_path, [("olt-service", "volt.xproto"),
40 ("vsg", "vsg.xproto"),
Matteo Scandoloe04a4882019-02-14 10:22:24 -080041 ("rcord", "rcord.xproto"),])
Matteo Scandolod44ca992018-05-17 15:02:10 -070042
Scott Baker47b47302019-01-30 16:55:07 -080043 import xossynchronizer.modelaccessor
44 reload(xossynchronizer.modelaccessor) # in case nose2 loaded it in a previous test
45
46 from xossynchronizer.modelaccessor import model_accessor
47 self.model_accessor = model_accessor
48
49 from pull_onus import ONUDevicePullStep
Matteo Scandolod44ca992018-05-17 15:02:10 -070050
51 # import all class names to globals
52 for (k, v) in model_accessor.all_model_classes.items():
53 globals()[k] = v
54
55 self.sync_step = ONUDevicePullStep
56
57 # mock volt service
58 self.volt_service = Mock()
59 self.volt_service.id = "volt_service_id"
60 self.volt_service.voltha_url = "voltha_url"
61 self.volt_service.voltha_user = "voltha_user"
62 self.volt_service.voltha_pass = "voltha_pass"
63 self.volt_service.voltha_port = 1234
64
65 # mock OLTDevice
66 self.olt = Mock()
67 self.olt.id = 1
68
Scott Baker22b46c52018-11-15 15:15:29 -080069 # second mock OLTDevice
70 self.olt2 = Mock()
71 self.olt2.id = 2
72
Matteo Scandolo6be6ee92018-05-24 15:07:51 -070073 # mock pon port
74 self.pon_port = Mock()
75 self.pon_port.id = 1
76
Scott Baker22b46c52018-11-15 15:15:29 -080077 # mock pon port
78 self.pon_port2 = Mock()
79 self.pon_port2.id = 2
80
Matteo Scandolod44ca992018-05-17 15:02:10 -070081 # mock voltha responses
82 self.devices = {
83 "items": [
84 {
85 "id": "0001130158f01b2d",
86 "type": "broadcom_onu",
87 "vendor": "Broadcom",
88 "serial_number": "BRCM22222222",
89 "vendor_id": "BRCM",
90 "adapter": "broadcom_onu",
91 "vlan": 0,
92 "admin_state": "ENABLED",
93 "oper_status": "ACTIVE",
94 "connect_status": "REACHABLE",
Scott Baker3db0eef2019-01-15 11:56:41 -080095 "reason": "starting-omci",
Matteo Scandolo6be6ee92018-05-24 15:07:51 -070096 "parent_id": "00010fc93996afea",
97 "parent_port_no": 1
Matteo Scandolod44ca992018-05-17 15:02:10 -070098 }
99 ]
100 }
101
Scott Baker22b46c52018-11-15 15:15:29 -0800102 self.two_devices = {
103 "items": [
104 {
105 "id": "0001130158f01b2d",
106 "type": "broadcom_onu",
107 "vendor": "Broadcom",
108 "serial_number": "BRCM22222222",
109 "vendor_id": "BRCM",
110 "adapter": "broadcom_onu",
111 "vlan": 0,
112 "admin_state": "ENABLED",
113 "oper_status": "ACTIVE",
114 "connect_status": "REACHABLE",
Scott Baker3db0eef2019-01-15 11:56:41 -0800115 "reason": "starting-omci",
Scott Baker22b46c52018-11-15 15:15:29 -0800116 "parent_id": "00010fc93996afea",
117 "parent_port_no": 1
118 },
119 {
120 "id": "0001130158f01b2e",
121 "type": "broadcom_onu",
122 "vendor": "Broadcom",
123 "serial_number": "BRCM22222223",
124 "vendor_id": "BRCM",
125 "adapter": "broadcom_onu",
126 "vlan": 0,
127 "admin_state": "ENABLED",
128 "oper_status": "ACTIVE",
129 "connect_status": "REACHABLE",
Scott Baker3db0eef2019-01-15 11:56:41 -0800130 "reason": "omci-admin-lock",
Scott Baker22b46c52018-11-15 15:15:29 -0800131 "parent_id": "00010fc93996afeb",
132 "parent_port_no": 1
133 }
134 ],
135 }
136
Matteo Scandolod8ed60e2018-06-18 17:00:57 -0700137 # TODO add ports
138 self.ports = {
139 "items": []
140 }
141
Matteo Scandolod44ca992018-05-17 15:02:10 -0700142 def tearDown(self):
143 sys.path = self.sys_path_save
144
145 @requests_mock.Mocker()
146 def test_missing_volt_service(self, m):
147 self.assertFalse(m.called)
148
149 @requests_mock.Mocker()
Matteo Scandolo6be6ee92018-05-24 15:07:51 -0700150 def test_pull(self, m):
Matteo Scandolod44ca992018-05-17 15:02:10 -0700151
152 with patch.object(VOLTService.objects, "all") as olt_service_mock, \
153 patch.object(OLTDevice.objects, "get") as mock_olt_device, \
Matteo Scandolo6be6ee92018-05-24 15:07:51 -0700154 patch.object(PONPort.objects, "get") as mock_pon_port, \
Matteo Scandolod8ed60e2018-06-18 17:00:57 -0700155 patch.object(ONUDevice, "save", autospec=True) as mock_save:
Matteo Scandolod44ca992018-05-17 15:02:10 -0700156 olt_service_mock.return_value = [self.volt_service]
Matteo Scandolo6be6ee92018-05-24 15:07:51 -0700157 mock_pon_port.return_value = self.pon_port
Matteo Scandolod44ca992018-05-17 15:02:10 -0700158 mock_olt_device.return_value = self.olt
159
160 m.get("http://voltha_url:1234/api/v1/devices", status_code=200, json=self.devices)
Matteo Scandolod8ed60e2018-06-18 17:00:57 -0700161 m.get("http://voltha_url:1234/api/v1/devices/0001130158f01b2d/ports", status_code=200, json=self.ports)
Matteo Scandolod44ca992018-05-17 15:02:10 -0700162
Scott Baker47b47302019-01-30 16:55:07 -0800163 self.sync_step(model_accessor=self.model_accessor).pull_records()
Matteo Scandolod44ca992018-05-17 15:02:10 -0700164
Matteo Scandolod8ed60e2018-06-18 17:00:57 -0700165 saved_onu = mock_save.call_args[0][0]
Matteo Scandolod44ca992018-05-17 15:02:10 -0700166
Matteo Scandolod8ed60e2018-06-18 17:00:57 -0700167 self.assertEqual(saved_onu.admin_state, "ENABLED")
168 self.assertEqual(saved_onu.oper_status, "ACTIVE")
169 self.assertEqual(saved_onu.connect_status, "REACHABLE")
Scott Baker3db0eef2019-01-15 11:56:41 -0800170 self.assertEqual(saved_onu.reason, "starting-omci")
Matteo Scandolod8ed60e2018-06-18 17:00:57 -0700171 self.assertEqual(saved_onu.device_type, "broadcom_onu")
172 self.assertEqual(saved_onu.vendor, "Broadcom")
173 self.assertEqual(saved_onu.device_id, "0001130158f01b2d")
Matteo Scandolod44ca992018-05-17 15:02:10 -0700174
Matteo Scandolod8ed60e2018-06-18 17:00:57 -0700175 self.assertEqual(mock_save.call_count, 1)
Matteo Scandolod44ca992018-05-17 15:02:10 -0700176
Scott Baker22b46c52018-11-15 15:15:29 -0800177 @requests_mock.Mocker()
178 def test_pull_bad_pon(self, m):
179
180 def olt_side_effect(device_id):
181 # fail the first onu device
182 if device_id=="00010fc93996afea":
183 return self.olt
184 else:
185 return self.olt2
186
187 def pon_port_side_effect(mock_pon_port, port_no, olt_device_id):
188 # fail the first onu device
189 if olt_device_id==1:
190 raise IndexError()
191 return self.pon_port2
192
193 with patch.object(VOLTService.objects, "all") as olt_service_mock, \
194 patch.object(OLTDevice.objects, "get") as mock_olt_device, \
195 patch.object(PONPort.objects, "get") as mock_pon_port, \
196 patch.object(ONUDevice, "save", autospec=True) as mock_save:
197 olt_service_mock.return_value = [self.volt_service]
198 mock_pon_port.side_effect = functools.partial(pon_port_side_effect, self.pon_port)
199 mock_olt_device.side_effect = olt_side_effect
200
201 m.get("http://voltha_url:1234/api/v1/devices", status_code=200, json=self.two_devices)
202 m.get("http://voltha_url:1234/api/v1/devices/0001130158f01b2d/ports", status_code=200, json=self.ports)
203 m.get("http://voltha_url:1234/api/v1/devices/0001130158f01b2e/ports", status_code=200, json=self.ports)
204
Scott Baker47b47302019-01-30 16:55:07 -0800205 self.sync_step(model_accessor=self.model_accessor).pull_records()
Scott Baker22b46c52018-11-15 15:15:29 -0800206
207 self.assertEqual(mock_save.call_count, 1)
208 saved_onu = mock_save.call_args[0][0]
209
210 # we should get the second onu in self.two_onus
211
212 self.assertEqual(saved_onu.admin_state, "ENABLED")
213 self.assertEqual(saved_onu.oper_status, "ACTIVE")
214 self.assertEqual(saved_onu.connect_status, "REACHABLE")
Scott Baker3db0eef2019-01-15 11:56:41 -0800215 self.assertEqual(saved_onu.reason, "omci-admin-lock")
Scott Baker22b46c52018-11-15 15:15:29 -0800216 self.assertEqual(saved_onu.device_type, "broadcom_onu")
217 self.assertEqual(saved_onu.vendor, "Broadcom")
218 self.assertEqual(saved_onu.device_id, "0001130158f01b2e")
219
220 self.assertEqual(mock_save.call_count, 1)
221
222 @requests_mock.Mocker()
223 def test_pull_bad_olt(self, m):
224
225 def olt_side_effect(device_id):
226 # fail the first onu device
227 if device_id=="00010fc93996afea":
228 raise IndexError()
229 else:
230 return self.olt2
231
232 with patch.object(VOLTService.objects, "all") as olt_service_mock, \
233 patch.object(OLTDevice.objects, "get") as mock_olt_device, \
234 patch.object(PONPort.objects, "get") as mock_pon_port, \
235 patch.object(ONUDevice, "save", autospec=True) as mock_save:
236 olt_service_mock.return_value = [self.volt_service]
237 mock_pon_port.return_value = self.pon_port2
238 mock_olt_device.side_effect = olt_side_effect
239
240 m.get("http://voltha_url:1234/api/v1/devices", status_code=200, json=self.two_devices)
241 m.get("http://voltha_url:1234/api/v1/devices/0001130158f01b2d/ports", status_code=200, json=self.ports)
242 m.get("http://voltha_url:1234/api/v1/devices/0001130158f01b2e/ports", status_code=200, json=self.ports)
243
Scott Baker47b47302019-01-30 16:55:07 -0800244 self.sync_step(model_accessor=self.model_accessor).pull_records()
Scott Baker22b46c52018-11-15 15:15:29 -0800245
246 self.assertEqual(mock_save.call_count, 1)
247 saved_onu = mock_save.call_args[0][0]
248
249 # we should get the second onu in self.two_onus
250
251 self.assertEqual(saved_onu.admin_state, "ENABLED")
252 self.assertEqual(saved_onu.oper_status, "ACTIVE")
253 self.assertEqual(saved_onu.connect_status, "REACHABLE")
Scott Baker3db0eef2019-01-15 11:56:41 -0800254 self.assertEqual(saved_onu.reason, "omci-admin-lock")
Scott Baker22b46c52018-11-15 15:15:29 -0800255 self.assertEqual(saved_onu.device_type, "broadcom_onu")
256 self.assertEqual(saved_onu.vendor, "Broadcom")
257 self.assertEqual(saved_onu.device_id, "0001130158f01b2e")
258
259 self.assertEqual(mock_save.call_count, 1)
260
Vijaykumar Kushwaha377d1da2019-03-20 10:29:42 +0530261#[SEBA-367] Unit test for blank response recieved from Voltha
262
263 @requests_mock.Mocker()
264 def test_blank_response_received(self, m):
265
266 m.get("http://voltha_url:1234/api/v1/devices", status_code=200, text="")
267 with patch.object(VOLTService.objects, "all") as olt_service_mock, \
268 patch.object(PONPort.objects, "get") as mock_pon_port, \
269 patch.object(OLTDevice.objects, "get") as mock_get, \
270 patch.object(ONUDevice, "save", autospec=True) as mock_save:
271
272 olt_service_mock.return_value = [self.volt_service]
273
274 self.sync_step(model_accessor=self.model_accessor).pull_records()
275
276 olt_service_mock.assert_called()
277 mock_pon_port.assert_not_called()
278 mock_get.assert_not_called()
279 self.assertEqual(mock_save.call_count, 0)
280
281#[SEBA-367] Unit test for invalid json received from Voltha
282
283 @requests_mock.Mocker()
284 def test_invalid_json(self, m):
285
286 m.get("http://voltha_url:1234/api/v1/devices", status_code=200, text="{\"items\" : [host_and_port}")
287 with patch.object(VOLTService.objects, "all") as olt_service_mock, \
288 patch.object(PONPort.objects, "get") as mock_pon_port, \
289 patch.object(OLTDevice.objects, "get") as mock_get, \
290 patch.object(ONUDevice, "save", autospec=True) as mock_save:
291
292 olt_service_mock.return_value = [self.volt_service]
293
294 self.sync_step(model_accessor=self.model_accessor).pull_records()
295
296 olt_service_mock.assert_called()
297 mock_pon_port.assert_not_called()
298 mock_get.assert_not_called()
299 self.assertEqual(mock_save.call_count, 0)
Matteo Scandolod44ca992018-05-17 15:02:10 -0700300
301if __name__ == "__main__":
Vijaykumar Kushwaha377d1da2019-03-20 10:29:42 +0530302 unittest.main()