blob: a34fedccd36bebe7ec0b842616e1e0964fb74d93 [file] [log] [blame]
Matteo Scandoload0c1752018-08-09 15:47:16 -07001
2# Copyright 2017-present Open Networking Foundation
3#
4# Licensed under the Apache License, Version 2.0 (the "License");
5# you may not use this file except in compliance with the License.
6# You may obtain a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15
16
17import unittest
18from mock import patch, call, Mock, PropertyMock
19
20import os, sys
21
22test_path=os.path.abspath(os.path.dirname(os.path.realpath(__file__)))
23service_dir=os.path.join(test_path, "../../../..")
24xos_dir=os.path.join(test_path, "../../..")
25if not os.path.exists(os.path.join(test_path, "new_base")):
26 xos_dir=os.path.join(test_path, "../../../../../../orchestration/xos/xos")
27 services_dir=os.path.join(xos_dir, "../../xos_services")
28
29def get_models_fn(service_name, xproto_name):
30 name = os.path.join(service_name, "xos", "synchronizer", "models", xproto_name)
31 if os.path.exists(os.path.join(services_dir, name)):
32 return name
33 raise Exception("Unable to find service=%s xproto=%s" % (service_name, xproto_name))
34
35class TestModelPolicyAttWorkflowDriverServiceInstance(unittest.TestCase):
36 def setUp(self):
37
38 self.sys_path_save = sys.path
39 sys.path.append(xos_dir)
40 sys.path.append(os.path.join(xos_dir, 'synchronizers', 'new_base'))
41
42 config = os.path.join(test_path, "../test_config.yaml")
43 from xosconfig import Config
44 Config.clear()
45 Config.init(config, 'synchronizer-config-schema.yaml')
46
47 from synchronizers.new_base.mock_modelaccessor_build import build_mock_modelaccessor
48 build_mock_modelaccessor(xos_dir, services_dir, [
49 get_models_fn("att-workflow-driver", "att-workflow-driver.xproto"),
50 get_models_fn("olt-service", "volt.xproto"),
51 get_models_fn("../profiles/rcord", "rcord.xproto")
52 ])
53
54 import synchronizers.new_base.modelaccessor
55 from model_policy_att_workflow_driver_serviceinstance import AttWorkflowDriverServiceInstancePolicy, model_accessor
56
57 from mock_modelaccessor import MockObjectList
58
59 # import all class names to globals
60 for (k, v) in model_accessor.all_model_classes.items():
61 globals()[k] = v
62
63 # Some of the functions we call have side-effects. For example, creating a VSGServiceInstance may lead to creation of
64 # tags. Ideally, this wouldn't happen, but it does. So make sure we reset the world.
65 model_accessor.reset_all_object_stores()
66
67 self.policy = AttWorkflowDriverServiceInstancePolicy()
68 self.si = AttWorkflowDriverServiceInstance()
69 self.si.owner = AttWorkflowDriverService()
70
71 def tearDown(self):
72 sys.path = self.sys_path_save
73 self.si = None
74
75 def test_not_synced(self):
76 self.si.valid = "awaiting"
77 self.si.backend_code = 0
78
79 with patch.object(RCORDSubscriber, "save") as subscriber_save, \
80 patch.object(ONUDevice, "save") as onu_save:
81
82 with self.assertRaises(Exception) as e:
83 self.policy.handle_update(self.si)
84
85 self.assertIn("has not been synced yet", e.exception.message)
86
87 def test_skip_update(self):
88 self.si.valid = "awaiting"
89 self.si.backend_code = 1
90
91 with patch.object(RCORDSubscriber, "save") as subscriber_save, \
92 patch.object(ONUDevice, "save") as onu_save:
93
94 self.policy.handle_update(self.si)
95 subscriber_save.assert_not_called()
96 onu_save.assert_not_called()
97
98 def test_disable_onu(self):
99 self.si.valid = "invalid"
100 self.si.serial_number = "BRCM1234"
101 self.si.backend_code = 1
102
103 onu = ONUDevice(
104 serial_number=self.si.serial_number
105 )
106
107 with patch.object(ONUDevice.objects, "get_items") as onu_objects, \
108 patch.object(RCORDSubscriber, "save") as subscriber_save, \
109 patch.object(ONUDevice, "save") as onu_save:
110
111 onu_objects.return_value = [onu]
112
113 self.policy.handle_update(self.si)
114 subscriber_save.assert_not_called()
115 self.assertEqual(onu.admin_state, "DISABLED")
116 onu_save.assert_called()
117
118 def test_enable_onu(self):
119 self.si.valid = "valid"
120 self.si.serial_number = "BRCM1234"
121 self.si.c_tag = None
122 self.si.backend_code = 1
123
124 onu = ONUDevice(
125 serial_number=self.si.serial_number,
126 admin_state="DISABLED"
127 )
128
129 subscriber = RCORDSubscriber(
130 onu_device=self.si.serial_number,
131 status='pre-provisioned'
132 )
133
134 with patch.object(ONUDevice.objects, "get_items") as onu_objects, \
135 patch.object(RCORDSubscriber.objects, "get_items") as subscriber_objects, \
136 patch.object(ONUDevice, "save") as onu_save:
137
138 onu_objects.return_value = [onu]
139 subscriber_objects.return_value = [subscriber]
140
141 self.policy.handle_update(self.si)
142 self.assertEqual(onu.admin_state, "ENABLED")
143 onu_save.assert_called()
144
145 def test_do_not_create_subscriber(self):
146 self.si.valid = "valid"
147 self.si.backend_code = 1
148 self.si.serial_number = "BRCM1234"
149 self.si.authentication_state = "DENIEND"
150 self.si.owner.leaf_model.create_on_discovery = False
151
152 onu = ONUDevice(
153 serial_number=self.si.serial_number,
154 admin_state="DISABLED"
155 )
156
157 with patch.object(ONUDevice.objects, "get_items") as onu_objects, \
158 patch.object(RCORDSubscriber, "save", autospec=True) as subscriber_save, \
159 patch.object(ONUDevice, "save") as onu_save:
160
161 onu_objects.return_value = [onu]
162
163 self.policy.handle_update(self.si)
164
165 self.assertEqual(onu.admin_state, "ENABLED")
166 onu_save.assert_called()
167 self.assertEqual(subscriber_save.call_count, 0)
168
169 def test_create_subscriber(self):
170 self.si.valid = "valid"
171 self.si.serial_number = "BRCM1234"
172 self.si.backend_code = 1
173
174 onu = ONUDevice(
175 serial_number=self.si.serial_number,
176 admin_state="ENABLED"
177 )
178
179 with patch.object(ONUDevice.objects, "get_items") as onu_objects, \
180 patch.object(RCORDSubscriber, "save", autospec=True) as subscriber_save, \
181 patch.object(ONUDevice, "save") as onu_save:
182
183 onu_objects.return_value = [onu]
184
185 self.policy.handle_update(self.si)
186 self.assertEqual(subscriber_save.call_count, 1)
187
188 subscriber = subscriber_save.call_args[0][0]
189 self.assertEqual(subscriber.onu_device, self.si.serial_number)
190
191 onu_save.assert_not_called()
192
193 def test_create_subscriber_no_create_on_discovery(self):
194 """
195 test_create_subscriber_no_create_on_discovery
196 When si.owner.create_on_discovery = False we still need to create the subscriber after authentication
197 """
198
199 self.si.valid = "valid"
200 self.si.serial_number = "BRCM1234"
201 self.si.backend_code = 1
202 self.si.owner.leaf_model.create_on_discovery = False
203 self.si.authentication_state = "APPROVED"
204
205 onu = ONUDevice(
206 serial_number=self.si.serial_number,
207 admin_state="ENABLED"
208 )
209
210 with patch.object(ONUDevice.objects, "get_items") as onu_objects, \
211 patch.object(RCORDSubscriber, "save", autospec=True) as subscriber_save, \
212 patch.object(ONUDevice, "save") as onu_save:
213
214 onu_objects.return_value = [onu]
215
216 self.policy.handle_update(self.si)
217 self.assertEqual(subscriber_save.call_count, 1)
218
219 subscriber = subscriber_save.call_args[0][0]
220 self.assertEqual(subscriber.onu_device, self.si.serial_number)
221
222 onu_save.assert_not_called()
223
224 def test_create_subscriber_with_ctag(self):
225 self.si.valid = "valid"
226 self.si.serial_number = "BRCM1234"
227 self.si.c_tag = 111
228 self.si.backend_code = 1
229
230 onu = ONUDevice(
231 serial_number=self.si.serial_number,
232 admin_state="ENABLED"
233 )
234
235 with patch.object(ONUDevice.objects, "get_items") as onu_objects, \
236 patch.object(RCORDSubscriber, "save", autospec=True) as subscriber_save, \
237 patch.object(ONUDevice, "save") as onu_save:
238
239 onu_objects.return_value = [onu]
240
241 self.policy.handle_update(self.si)
242 self.assertEqual(subscriber_save.call_count, 1)
243
244 subscriber = subscriber_save.call_args[0][0]
245 self.assertEqual(subscriber.onu_device, self.si.serial_number)
246 self.assertEqual(subscriber.c_tag, self.si.c_tag)
247
248 onu_save.assert_not_called()
249
250 def _test_add_c_tag_to_pre_provisioned_subscriber(self):
251 self.si.valid = "valid"
252 self.si.serial_number = "BRCM1234"
253 self.si.c_tag = 111
254 self.si.backend_code = 1
255
256 onu = ONUDevice(
257 serial_number=self.si.serial_number,
258 admin_state="ENABLED"
259 )
260
261 subscriber = RCORDSubscriber(
262 onu_device=self.si.serial_number,
263 )
264
265 with patch.object(ONUDevice.objects, "get_items") as onu_objects, \
266 patch.object(RCORDSubscriber.objects, "get_items") as subscriber_objects, \
267 patch.object(RCORDSubscriber, "save", autospec=True) as subscriber_save, \
268 patch.object(ONUDevice, "save") as onu_save:
269
270 onu_objects.return_value = [onu]
271 subscriber_objects.return_value = [subscriber]
272
273 self.policy.handle_update(self.si)
274 self.assertEqual(subscriber_save.call_count, 1)
275
276 subscriber = subscriber_save.call_args[0][0]
277 self.assertEqual(subscriber.onu_device, self.si.serial_number)
278 self.assertEqual(subscriber.c_tag, self.si.c_tag)
279
280 onu_save.assert_not_called()
281
282if __name__ == '__main__':
283 unittest.main()
284