blob: 13ec5a4fa1b171dbd27008e9ebdaa1d6d4dcf512 [file] [log] [blame]
Matteo Scandolocc94e902018-05-22 15:25:25 -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
Scott Bakerb1671512019-04-01 19:16:54 -070016import os
17import sys
Matteo Scandolocc94e902018-05-22 15:25:25 -070018from mock import patch, Mock, MagicMock
19
Scott Bakerb1671512019-04-01 19:16:54 -070020test_path = os.path.abspath(os.path.dirname(os.path.realpath(__file__)))
21service_dir = os.path.join(test_path, "../../../..")
22xos_dir = os.path.join(test_path, "../../..")
Matteo Scandolo0813b882018-07-16 14:38:01 -040023if not os.path.exists(os.path.join(test_path, "new_base")):
Scott Bakerb1671512019-04-01 19:16:54 -070024 xos_dir = os.path.join(test_path, "../../../../../../orchestration/xos/xos")
25 services_dir = os.path.join(xos_dir, "../../xos_services")
Matteo Scandolo0813b882018-07-16 14:38:01 -040026
Matteo Scandolocc94e902018-05-22 15:25:25 -070027# mocking XOS exception, as they're based in Django
Scott Bakerb1671512019-04-01 19:16:54 -070028
29
Matteo Scandolocc94e902018-05-22 15:25:25 -070030class Exceptions:
31 XOSValidationError = Exception
32 XOSProgrammingError = Exception
33 XOSPermissionDenied = Exception
vigneshethiraj4d3e5972019-04-04 19:22:14 +053034 XOSConfigurationError = Exception
Matteo Scandolocc94e902018-05-22 15:25:25 -070035
Scott Bakerb1671512019-04-01 19:16:54 -070036
Matteo Scandolocc94e902018-05-22 15:25:25 -070037class XOS:
38 exceptions = Exceptions
39
Scott Bakerb1671512019-04-01 19:16:54 -070040
Matteo Scandolocc94e902018-05-22 15:25:25 -070041class TestRCORDModels(unittest.TestCase):
42 def setUp(self):
Matteo Scandolo0813b882018-07-16 14:38:01 -040043
44 self.sys_path_save = sys.path
45 sys.path.append(xos_dir)
46
Matteo Scandolocc94e902018-05-22 15:25:25 -070047 self.xos = XOS
48
49 self.models_decl = Mock()
50 self.models_decl.RCORDSubscriber_decl = MagicMock
51 self.models_decl.RCORDSubscriber_decl.save = Mock()
52 self.models_decl.RCORDSubscriber_decl.objects = Mock()
53 self.models_decl.RCORDSubscriber_decl.objects.filter.return_value = []
54
Luca Preted6700ba2018-09-12 16:40:49 -070055 self.models_decl.RCORDIpAddress_decl = MagicMock
56 self.models_decl.RCORDIpAddress_decl.save = Mock()
57 self.models_decl.RCORDIpAddress_decl.objects = Mock()
58 self.models_decl.RCORDIpAddress_decl.objects.filter.return_value = []
59
Matteo Scandolocc94e902018-05-22 15:25:25 -070060 modules = {
61 'xos.exceptions': self.xos.exceptions,
62 'models_decl': self.models_decl
63 }
64
65 self.module_patcher = patch.dict('sys.modules', modules)
66 self.module_patcher.start()
67
Luca Preted6700ba2018-09-12 16:40:49 -070068 from models import RCORDSubscriber, RCORDIpAddress
Matteo Scandolocc94e902018-05-22 15:25:25 -070069
Matteo Scandoloa2b2e222019-06-18 18:32:43 -070070 self.volt = Mock(name="vOLT")
71 self.volt.leaf_model.name = "vOLT"
72 self.volt.get_olt_technology_from_unu_sn.return_value = "xgspon"
73
74
Matteo Scandolocc94e902018-05-22 15:25:25 -070075 self.rcord_subscriber_class = RCORDSubscriber
76
77 self.rcord_subscriber = RCORDSubscriber()
Matteo Scandolo10d94512019-03-04 15:54:43 -080078 self.rcord_subscriber.deleted = False
Scott Bakerb1671512019-04-01 19:16:54 -070079 self.rcord_subscriber.id = None # this is a new model
Matteo Scandoloaa37b422018-05-23 15:36:56 -070080 self.rcord_subscriber.is_new = True
Matteo Scandolocc94e902018-05-22 15:25:25 -070081 self.rcord_subscriber.onu_device = "BRCM1234"
Daniele Morocca9e792020-02-19 17:45:53 -080082 self.rcord_subscriber.onu_device_uni = "BRCM1234-1"
Matteo Scandoloaa37b422018-05-23 15:36:56 -070083 self.rcord_subscriber.c_tag = 111
Matteo Scandoloc348b3f2018-07-29 09:35:11 -070084 self.rcord_subscriber.s_tag = 222
Luca Preted6700ba2018-09-12 16:40:49 -070085 self.rcord_subscriber.ips = Mock()
86 self.rcord_subscriber.ips.all.return_value = []
Matteo Scandolocc94e902018-05-22 15:25:25 -070087 self.rcord_subscriber.mac_address = "00:AA:00:00:00:01"
88 self.rcord_subscriber.owner.leaf_model.access = "voltha"
89 self.rcord_subscriber.owner.provider_services = [self.volt]
vigneshethiraj4d3e5972019-04-04 19:22:14 +053090 self.rcord_subscriber.list_of_unused_c_tags_for_s_tag = []
Matteo Scandolocc94e902018-05-22 15:25:25 -070091
Matteo Scandoloa2b2e222019-06-18 18:32:43 -070092
Luca Preted6700ba2018-09-12 16:40:49 -070093 self.rcord_ip = RCORDIpAddress()
Scott Bakerb1671512019-04-01 19:16:54 -070094 self.rcord_ip.subscriber = 1
Luca Preted6700ba2018-09-12 16:40:49 -070095
Matteo Scandoloa2b2e222019-06-18 18:32:43 -070096 # TODO add a test for validate_tech_profile_id
97
Matteo Scandolo0813b882018-07-16 14:38:01 -040098 def tearDown(self):
99 sys.path = self.sys_path_save
Matteo Scandolocc94e902018-05-22 15:25:25 -0700100
101 def test_save(self):
102 self.rcord_subscriber.save()
103 self.models_decl.RCORDSubscriber_decl.save.assert_called()
104
Luca Preted6700ba2018-09-12 16:40:49 -0700105 def _test_validate_ipv4_address(self):
106 self.rcord_ip.ip = "192.168.0."
Matteo Scandolocc94e902018-05-22 15:25:25 -0700107 with self.assertRaises(Exception) as e:
Luca Preted6700ba2018-09-12 16:40:49 -0700108 self.rcord_ip.save()
Matteo Scandolocc94e902018-05-22 15:25:25 -0700109
Luca Preted6700ba2018-09-12 16:40:49 -0700110 self.assertEqual(e.exception.message, "The IP specified is not valid: 192.168.0.")
111 self.models_decl.RCORDIpAddress.save.assert_not_called()
112
113 def test_validate_ipv6_address(self):
114 self.rcord_ip.ip = "2001:0db8:85a3:0000:0000:8a2e:03"
115 with self.assertRaises(Exception) as e:
116 self.rcord_ip.save()
117
118 self.assertEqual(e.exception.message, "The IP specified is not valid: 2001:0db8:85a3:0000:0000:8a2e:03")
119 self.models_decl.RCORDIpAddress.save.assert_not_called()
Matteo Scandolocc94e902018-05-22 15:25:25 -0700120
121 def test_validate_mac_address(self):
122 self.rcord_subscriber.mac_address = "invalid"
123 with self.assertRaises(Exception) as e:
124 self.rcord_subscriber.save()
125
Luca Preted6700ba2018-09-12 16:40:49 -0700126 self.assertEqual(e.exception.message, "The MAC address specified is not valid: invalid")
Matteo Scandolocc94e902018-05-22 15:25:25 -0700127 self.models_decl.RCORDSubscriber_decl.save.assert_not_called()
128
129 def test_valid_onu_device(self):
130 self.rcord_subscriber.save()
131 self.models_decl.RCORDSubscriber_decl.save.assert_called()
132
133 def test_invalid_onu_device(self):
134 self.volt.leaf_model.has_access_device.return_value = False
135 with self.assertRaises(Exception) as e:
136 self.rcord_subscriber.save()
137
138 self.assertEqual(e.exception.message, "The onu_device you specified (BRCM1234) does not exists")
139 self.models_decl.RCORDSubscriber_decl.save.assert_not_called()
140
Matteo Scandolo10d94512019-03-04 15:54:43 -0800141 def test_missing_onu_device_on_delete(self):
142 self.volt.leaf_model.has_access_device.return_value = False
143 self.rcord_subscriber.deleted = True
144 self.rcord_subscriber.save()
145 self.models_decl.RCORDSubscriber_decl.save.assert_called()
146
Matteo Scandoloa1875602018-10-16 07:35:04 -0700147 def test_validate_c_tag_pass(self):
148 """
149 check that other subscriber attached to the same ONU don't have the same c_tag
150 """
151
152 self.models_decl.RCORDSubscriber_decl.objects.filter.return_value = [self.rcord_subscriber]
153
Matteo Scandoloa1875602018-10-16 07:35:04 -0700154 self.rcord_subscriber.save()
155
156 self.models_decl.RCORDSubscriber_decl.save.assert_called()
157
158 def test_validate_c_tag_fail(self):
Matteo Scandolocc94e902018-05-22 15:25:25 -0700159 """
160 check that other subscriber attached to the same ONU don't have the same c_tag
161 """
162
163 s = Mock()
Matteo Scandoloaa37b422018-05-23 15:36:56 -0700164 s.c_tag = 111
Matteo Scandolocc94e902018-05-22 15:25:25 -0700165 s.onu_device = "BRCM1234"
166
Matteo Scandoloa1875602018-10-16 07:35:04 -0700167 self.models_decl.RCORDSubscriber_decl.objects.filter.return_value = [s, self.rcord_subscriber]
Matteo Scandolocc94e902018-05-22 15:25:25 -0700168
169 with self.assertRaises(Exception) as e:
170 self.rcord_subscriber.save()
171
172 self.assertEqual(e.exception.message, "The c_tag you specified (111) has already been used on device BRCM1234")
173 self.models_decl.RCORDSubscriber_decl.save.assert_not_called()
174
HARDIK WINDLASSabe3a9c2019-03-18 16:17:28 +0530175 def test_validate_get_used_s_c_tag_subscriber_id(self):
Matteo Scandoloa1875602018-10-16 07:35:04 -0700176 """
177 check that other subscriber using the same s_tag don't have the same c_tag
178 """
HARDIK WINDLASSabe3a9c2019-03-18 16:17:28 +0530179 s1 = Mock()
180 s1.id = 456
181 s1.c_tag = 999
182 s1.s_tag = 222
183 s1.onu_device = "BRCM1234"
184
185 s2 = Mock()
186 s2.id = 123
187 s2.c_tag = 111
188 s2.s_tag = 222
189 s2.onu_device = "BRCM12345"
190
191 self.rcord_subscriber.get_same_onu_subscribers = Mock()
192 self.rcord_subscriber.get_same_onu_subscribers.return_value = [s1]
193
194 self.rcord_subscriber.get_same_s_c_tag_subscribers = Mock()
195 self.rcord_subscriber.get_same_s_c_tag_subscribers.return_value = [s2]
Matteo Scandoloa1875602018-10-16 07:35:04 -0700196
197 with self.assertRaises(Exception) as e:
198 self.rcord_subscriber.save()
199
Scott Bakerb1671512019-04-01 19:16:54 -0700200 self.assertEqual(e.exception.message,
201 "The c_tag(111) and s_tag(222) pair you specified,has already been used by Subscriber with Id (123)")
Matteo Scandoloa1875602018-10-16 07:35:04 -0700202 self.models_decl.RCORDSubscriber_decl.save.assert_not_called()
203
Matteo Scandoloaa37b422018-05-23 15:36:56 -0700204 def test_validate_c_tag_on_update(self):
205 s = Mock()
206 s.c_tag = 111
207 s.onu_device = "BRCM1234"
208 s.id = 1
209
210 self.models_decl.RCORDSubscriber_decl.objects.filter.return_value = [s]
211
212 self.rcord_subscriber.is_new = False
213 self.rcord_subscriber.id = 1
214 self.rcord_subscriber.save()
215
216 self.models_decl.RCORDSubscriber_decl.save.assert_called()
217
218 def test_validate_c_tag_on_update_fail(self):
219 s = Mock()
220 s.c_tag = 222
221 s.onu_device = "BRCM1234"
222 s.id = 2
223
224 self.models_decl.RCORDSubscriber_decl.objects.filter.return_value = [s]
225
226 self.rcord_subscriber.id = 1
227 self.rcord_subscriber.is_new = False
228 self.rcord_subscriber.c_tag = 222
229 with self.assertRaises(Exception) as e:
230 self.rcord_subscriber.save()
231
232 self.assertEqual(e.exception.message, "The c_tag you specified (222) has already been used on device BRCM1234")
233 self.models_decl.RCORDSubscriber_decl.save.assert_not_called()
Matteo Scandolocc94e902018-05-22 15:25:25 -0700234
235 def test_generate_c_tag(self):
236 s = Mock()
237 s.c_tag = "111"
HARDIK WINDLASSabe3a9c2019-03-18 16:17:28 +0530238 s.s_tag = "223"
Matteo Scandolocc94e902018-05-22 15:25:25 -0700239 s.onu_device = "BRCM1234"
Daniele Morocca9e792020-02-19 17:45:53 -0800240
HARDIK WINDLASSabe3a9c2019-03-18 16:17:28 +0530241 self.rcord_subscriber.get_same_onu_subscribers = Mock()
242 self.rcord_subscriber.get_same_onu_subscribers.return_value = [s]
243
244 self.rcord_subscriber.get_same_s_c_tag_subscribers = Mock()
245 self.rcord_subscriber.get_same_s_c_tag_subscribers.return_value = []
246
Matteo Scandolocc94e902018-05-22 15:25:25 -0700247 self.rcord_subscriber.c_tag = None
248
249 self.rcord_subscriber.save()
250
251 self.models_decl.RCORDSubscriber_decl.save.assert_called()
252 self.assertNotEquals(self.rcord_subscriber.c_tag, "111")
253 self.assertGreater(self.rcord_subscriber.c_tag, 16)
254 self.assertLess(self.rcord_subscriber.c_tag, 4097)
255
vigneshethiraj4d3e5972019-04-04 19:22:14 +0530256 #Testing whether the random generation choses c_tag from the list provided
257 self.rcord_subscriber.c_tag = None
258 self.rcord_subscriber.s_tag = None
259 self.rcord_subscriber.unused_c_tags_for_s_tag = Mock()
260 self.rcord_subscriber.unused_c_tags_for_s_tag.return_value = [18,19]
261
262 self.rcord_subscriber.save()
263 self.models_decl.RCORDSubscriber_decl.save.assert_called()
264 self.assertGreater(self.rcord_subscriber.c_tag, 17)
265 self.assertLess(self.rcord_subscriber.c_tag, 20)
266
267 self.rcord_subscriber.c_tag = None
268 self.rcord_subscriber.s_tag = None
269
270 self.rcord_subscriber.save()
271 self.models_decl.RCORDSubscriber_decl.save.assert_called()
272 self.assertGreater(self.rcord_subscriber.c_tag, 16)
273 self.assertLess(self.rcord_subscriber.c_tag, 4096)
274
275 def test_unused_c_tags_for_s_tag(self):
276 s=[]
277 for i in range(16,4097):
278 d = Mock()
279 d.c_tag = i
280 d.s_tag = "222"
281 d.onu_device = "BRCM1234"
282 s.append(d)
283
284 self.rcord_subscriber.get_same_onu_subscribers = Mock()
285 self.rcord_subscriber.get_same_onu_subscribers.return_value = []
286 self.rcord_subscriber.get_same_s_c_tag_subscribers = Mock()
287 self.rcord_subscriber.get_same_s_c_tag_subscribers.return_value = []
288
289 self.models_decl.RCORDSubscriber_decl.objects.filter.return_value = s
290 self.rcord_subscriber.s_tag = 222
291 self.rcord_subscriber.c_tag = None
292 with self.assertRaises(Exception) as e:
293 self.rcord_subscriber.save()
294 self.assertEqual(e.exception.message, "All the c_tags are exhausted for this s_tag: 222")
295 self.models_decl.RCORDSubscriber_decl.save.assert_not_called()
296
297 def test_unused_s_tags_for_c_tag(self):
298 s=[]
299 for i in range(16,4097):
300 d = Mock()
301 d.s_tag = i
302 d.c_tag = "111"
303 d.onu_device = "BRCM1234"
304 s.append(d)
305
306 self.rcord_subscriber.get_same_onu_subscribers = Mock()
307 self.rcord_subscriber.get_same_onu_subscribers.return_value = []
308 self.rcord_subscriber.get_same_s_c_tag_subscribers = Mock()
309 self.rcord_subscriber.get_same_s_c_tag_subscribers.return_value = []
310
311 self.models_decl.RCORDSubscriber_decl.objects.filter.return_value = s
312 self.rcord_subscriber.c_tag = 111
313 self.rcord_subscriber.s_tag = None
314 with self.assertRaises(Exception) as e:
315 self.rcord_subscriber.save()
316 self.assertEqual(e.exception.message, "All the s_tags are exhausted for this c_tag: 111")
317 self.models_decl.RCORDSubscriber_decl.save.assert_not_called()
318
319
320
Matteo Scandoloc348b3f2018-07-29 09:35:11 -0700321 def test_generate_s_tag(self):
HARDIK WINDLASSabe3a9c2019-03-18 16:17:28 +0530322 self.rcord_subscriber.s_tag = None
Matteo Scandoloc348b3f2018-07-29 09:35:11 -0700323
324 self.rcord_subscriber.save()
325
326 self.models_decl.RCORDSubscriber_decl.save.assert_called()
327 self.assertNotEqual(self.rcord_subscriber.s_tag, None)
328
vigneshethiraj4d3e5972019-04-04 19:22:14 +0530329
Matteo Scandoloc348b3f2018-07-29 09:35:11 -0700330 def test_provisioned_s_stag(self):
331 self.rcord_subscriber.save()
332 self.models_decl.RCORDSubscriber_decl.save.assert_called()
333 self.assertEqual(self.rcord_subscriber.s_tag, 222)
334
335
Matteo Scandolocc94e902018-05-22 15:25:25 -0700336if __name__ == '__main__':
337 unittest.main()