blob: 8e15202536e030c7886be339aff36f975c072aa3 [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
Matteo Scandoloea529092018-09-11 16:36:39 -070067
Matteo Scandoload0c1752018-08-09 15:47:16 -070068 self.policy = AttWorkflowDriverServiceInstancePolicy()
69 self.si = AttWorkflowDriverServiceInstance()
70 self.si.owner = AttWorkflowDriverService()
Matteo Scandoloea529092018-09-11 16:36:39 -070071 self.si.serial_number = "BRCM1234"
Matteo Scandoload0c1752018-08-09 15:47:16 -070072
73 def tearDown(self):
74 sys.path = self.sys_path_save
Matteo Scandoload0c1752018-08-09 15:47:16 -070075
Matteo Scandoloc6ac74a2018-09-14 08:14:51 -070076 def test_update_onu(self):
77
78 onu = ONUDevice(
79 serial_number="BRCM1234",
80 admin_state="ENABLED"
81 )
82 with patch.object(ONUDevice.objects, "get_items") as get_onu, \
83 patch.object(onu, "save") as onu_save:
84 get_onu.return_value = [onu]
85
86 self.policy.update_onu("brcm1234", "ENABLED")
87 onu_save.assert_not_called()
88
89 self.policy.update_onu("brcm1234", "DISABLED")
90 self.assertEqual(onu.admin_state, "DISABLED")
91 onu_save.assert_called_with(always_update_timestamp=True)
92
93
Matteo Scandoload0c1752018-08-09 15:47:16 -070094 def test_enable_onu(self):
Matteo Scandoloea529092018-09-11 16:36:39 -070095 from helpers import AttHelpers
96 with patch.object(AttHelpers, "validate_onu") as validate_onu, \
97 patch.object(self.policy, "update_onu") as update_onu, \
98 patch.object(self.si, "save") as save_si:
99 validate_onu.return_value = [True, "valid onu"]
Matteo Scandoload0c1752018-08-09 15:47:16 -0700100
Matteo Scandoloea529092018-09-11 16:36:39 -0700101 self.policy.validate_onu_state(self.si)
Matteo Scandoload0c1752018-08-09 15:47:16 -0700102
Matteo Scandoloea529092018-09-11 16:36:39 -0700103 update_onu.assert_called_once()
104 update_onu.assert_called_with("BRCM1234", "ENABLED")
Matteo Scandoload0c1752018-08-09 15:47:16 -0700105
Matteo Scandoloea529092018-09-11 16:36:39 -0700106 self.assertIn("valid onu", self.si.status_message)
Matteo Scandoload0c1752018-08-09 15:47:16 -0700107
Matteo Scandoloea529092018-09-11 16:36:39 -0700108 def test_disable_onu(self):
109 from helpers import AttHelpers
110 with patch.object(AttHelpers, "validate_onu") as validate_onu, \
111 patch.object(self.policy, "update_onu") as update_onu, \
112 patch.object(self.si, "save") as save_si:
113 validate_onu.return_value = [False, "invalid onu"]
Matteo Scandoload0c1752018-08-09 15:47:16 -0700114
Matteo Scandoloea529092018-09-11 16:36:39 -0700115 self.policy.validate_onu_state(self.si)
116
117 update_onu.assert_called_once()
118 update_onu.assert_called_with("BRCM1234", "DISABLED")
119
120 self.assertIn("invalid onu", self.si.status_message)
121
122 def test_handle_update_validate_onu(self):
123 """
124 Testing that handle_update calls validate_onu with the correct parameters
125 when necessary
126 """
127 with patch.object(self.policy, "validate_onu_state") as validate_onu_state, \
128 patch.object(self.policy, "update_onu") as update_onu, \
129 patch.object(self.policy, "get_subscriber") as get_subscriber:
130 update_onu.return_value = None
131 get_subscriber.return_value = None
132
133 self.si.onu_state = "AWAITING"
Matteo Scandoload0c1752018-08-09 15:47:16 -0700134 self.policy.handle_update(self.si)
Matteo Scandoloea529092018-09-11 16:36:39 -0700135 validate_onu_state.assert_called_with(self.si)
Matteo Scandoload0c1752018-08-09 15:47:16 -0700136
Matteo Scandoloea529092018-09-11 16:36:39 -0700137 self.si.onu_state = "ENABLED"
Matteo Scandoload0c1752018-08-09 15:47:16 -0700138 self.policy.handle_update(self.si)
Matteo Scandoloea529092018-09-11 16:36:39 -0700139 validate_onu_state.assert_called_with(self.si)
Matteo Scandoload0c1752018-08-09 15:47:16 -0700140
Matteo Scandoloea529092018-09-11 16:36:39 -0700141 self.si.onu_state = "DISABLED"
142 self.policy.handle_update(self.si)
143 self.assertEqual(validate_onu_state.call_count, 2)
Matteo Scandoload0c1752018-08-09 15:47:16 -0700144
Matteo Scandoloea529092018-09-11 16:36:39 -0700145 def test_get_subscriber(self):
146
147 sub = RCORDSubscriber(
148 onu_device="BRCM1234"
149 )
150
151 with patch.object(RCORDSubscriber.objects, "get_items") as get_subscribers:
152 get_subscribers.return_value = [sub]
153
154 res = self.policy.get_subscriber("BRCM1234")
155 self.assertEqual(res, sub)
156
157 res = self.policy.get_subscriber("brcm1234")
158 self.assertEqual(res, sub)
159
160 res = self.policy.get_subscriber("foo")
161 self.assertEqual(res, None)
162
163 def test_update_subscriber(self):
164
165 sub = RCORDSubscriber(
166 onu_device="BRCM1234"
167 )
168
169 self.si.status_message = "some content"
170
171 with patch.object(sub, "save") as sub_save:
172 self.si.authentication_state = "AWAITING"
173 self.policy.update_subscriber(sub, self.si)
174 self.assertEqual(sub.status, "awaiting-auth")
175 self.assertIn("Awaiting Authentication", self.si.status_message)
176 sub_save.assert_called()
177 sub_save.reset_mock()
Matteo Scandoloc6ac74a2018-09-14 08:14:51 -0700178 sub.status = None
Matteo Scandoloea529092018-09-11 16:36:39 -0700179
180 self.si.authentication_state = "REQUESTED"
181 self.policy.update_subscriber(sub, self.si)
182 self.assertEqual(sub.status, "awaiting-auth")
183 self.assertIn("Authentication requested", self.si.status_message)
184 sub_save.assert_called()
185 sub_save.reset_mock()
Matteo Scandoloc6ac74a2018-09-14 08:14:51 -0700186 sub.status = None
Matteo Scandoloea529092018-09-11 16:36:39 -0700187
188 self.si.authentication_state = "STARTED"
189 self.policy.update_subscriber(sub, self.si)
190 self.assertEqual(sub.status, "awaiting-auth")
191 self.assertIn("Authentication started", self.si.status_message)
192 sub_save.assert_called()
193 sub_save.reset_mock()
Matteo Scandoloc6ac74a2018-09-14 08:14:51 -0700194 sub.status = None
Matteo Scandoloea529092018-09-11 16:36:39 -0700195
196 self.si.authentication_state = "APPROVED"
197 self.policy.update_subscriber(sub, self.si)
198 self.assertEqual(sub.status, "enabled")
199 self.assertIn("Authentication succeded", self.si.status_message)
200 sub_save.assert_called()
201 sub_save.reset_mock()
Matteo Scandoloc6ac74a2018-09-14 08:14:51 -0700202 sub.status = None
Matteo Scandoloea529092018-09-11 16:36:39 -0700203
204 self.si.authentication_state = "DENIED"
205 self.policy.update_subscriber(sub, self.si)
206 self.assertEqual(sub.status, "auth-failed")
207 self.assertIn("Authentication denied", self.si.status_message)
208 sub_save.assert_called()
209 sub_save.reset_mock()
Matteo Scandoloc6ac74a2018-09-14 08:14:51 -0700210 sub.status = None
211
212 def test_update_subscriber_not(self):
213 sub = RCORDSubscriber(
214 onu_device="BRCM1234"
215 )
216
217 with patch.object(sub, "save") as sub_save:
218 sub.status = "awaiting-auth"
219 self.si.authentication_state = "AWAITING"
220 self.policy.update_subscriber(sub, self.si)
221 sub_save.assert_not_called()
222
223 sub.status = "awaiting-auth"
224 self.si.authentication_state = "REQUESTED"
225 self.policy.update_subscriber(sub, self.si)
226 sub_save.assert_not_called()
227
228 sub.status = "awaiting-auth"
229 self.si.authentication_state = "STARTED"
230 self.policy.update_subscriber(sub, self.si)
231 sub_save.assert_not_called()
232
233 sub.status = "enabled"
234 self.si.authentication_state = "APPROVED"
235 self.policy.update_subscriber(sub, self.si)
236 sub_save.assert_not_called()
237
238 sub.status = "auth-failed"
239 self.si.authentication_state = "DENIED"
240 self.policy.update_subscriber(sub, self.si)
241 sub_save.assert_not_called()
242
Matteo Scandoloea529092018-09-11 16:36:39 -0700243
244 def test_handle_update_subscriber(self):
Matteo Scandoloe8c33d62018-08-16 14:37:24 -0700245 self.si.onu_state = "DISABLED"
Matteo Scandoload0c1752018-08-09 15:47:16 -0700246
Matteo Scandoloea529092018-09-11 16:36:39 -0700247 sub = RCORDSubscriber(
248 onu_device="BRCM1234"
Matteo Scandoloe8c33d62018-08-16 14:37:24 -0700249 )
250
Matteo Scandoloea529092018-09-11 16:36:39 -0700251 with patch.object(self.policy, "get_subscriber") as get_subscriber, \
Matteo Scandolob8da43d2018-09-12 15:52:16 -0700252 patch.object(self.policy, "update_onu") as update_onu, \
Matteo Scandoloea529092018-09-11 16:36:39 -0700253 patch.object(self.policy, "update_subscriber") as update_subscriber:
Matteo Scandoload0c1752018-08-09 15:47:16 -0700254
Matteo Scandoloea529092018-09-11 16:36:39 -0700255 get_subscriber.return_value = None
Matteo Scandoload0c1752018-08-09 15:47:16 -0700256 self.policy.handle_update(self.si)
Matteo Scandoloea529092018-09-11 16:36:39 -0700257 self.assertEqual(update_subscriber.call_count, 0)
Matteo Scandoload0c1752018-08-09 15:47:16 -0700258
Matteo Scandoloea529092018-09-11 16:36:39 -0700259 get_subscriber.return_value = sub
Matteo Scandoload0c1752018-08-09 15:47:16 -0700260 self.policy.handle_update(self.si)
Matteo Scandoloea529092018-09-11 16:36:39 -0700261 update_subscriber.assert_called_with(sub, self.si)
262
Matteo Scandoload0c1752018-08-09 15:47:16 -0700263
Matteo Scandoload0c1752018-08-09 15:47:16 -0700264if __name__ == '__main__':
265 unittest.main()
266