blob: 774efdacbe5dcaefaee7f37402a77dabd44d78ae [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
Matteo Scandoloe8c33d62018-08-16 14:37:24 -070087 def test_defer_update(self):
Matteo Scandoload0c1752018-08-09 15:47:16 -070088 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
Matteo Scandoloe8c33d62018-08-16 14:37:24 -070094 with self.assertRaises(Exception) as e:
95 self.policy.handle_update(self.si)
96
97 self.assertEqual(e.exception.message, "MODEL_POLICY: deferring handle_update for AttWorkflowDriverServiceInstance 98052 as not validated yet")
Matteo Scandoload0c1752018-08-09 15:47:16 -070098 subscriber_save.assert_not_called()
99 onu_save.assert_not_called()
100
101 def test_disable_onu(self):
102 self.si.valid = "invalid"
103 self.si.serial_number = "BRCM1234"
104 self.si.backend_code = 1
Matteo Scandoloe8c33d62018-08-16 14:37:24 -0700105 self.si.onu_state = "ENABLED"
Matteo Scandoload0c1752018-08-09 15:47:16 -0700106
107 onu = ONUDevice(
108 serial_number=self.si.serial_number
109 )
110
111 with patch.object(ONUDevice.objects, "get_items") as onu_objects, \
112 patch.object(RCORDSubscriber, "save") as subscriber_save, \
113 patch.object(ONUDevice, "save") as onu_save:
114
115 onu_objects.return_value = [onu]
116
117 self.policy.handle_update(self.si)
118 subscriber_save.assert_not_called()
119 self.assertEqual(onu.admin_state, "DISABLED")
120 onu_save.assert_called()
121
122 def test_enable_onu(self):
123 self.si.valid = "valid"
124 self.si.serial_number = "BRCM1234"
125 self.si.c_tag = None
126 self.si.backend_code = 1
Matteo Scandoloe8c33d62018-08-16 14:37:24 -0700127 self.si.onu_state = "ENABLED"
Matteo Scandoload0c1752018-08-09 15:47:16 -0700128
129 onu = ONUDevice(
130 serial_number=self.si.serial_number,
131 admin_state="DISABLED"
132 )
133
134 subscriber = RCORDSubscriber(
135 onu_device=self.si.serial_number,
136 status='pre-provisioned'
137 )
138
139 with patch.object(ONUDevice.objects, "get_items") as onu_objects, \
140 patch.object(RCORDSubscriber.objects, "get_items") as subscriber_objects, \
141 patch.object(ONUDevice, "save") as onu_save:
142
143 onu_objects.return_value = [onu]
144 subscriber_objects.return_value = [subscriber]
145
146 self.policy.handle_update(self.si)
147 self.assertEqual(onu.admin_state, "ENABLED")
148 onu_save.assert_called()
149
150 def test_do_not_create_subscriber(self):
151 self.si.valid = "valid"
152 self.si.backend_code = 1
153 self.si.serial_number = "BRCM1234"
154 self.si.authentication_state = "DENIEND"
Matteo Scandoloe8c33d62018-08-16 14:37:24 -0700155 self.si.onu_state = "ENABLED"
Matteo Scandoload0c1752018-08-09 15:47:16 -0700156
157 onu = ONUDevice(
158 serial_number=self.si.serial_number,
159 admin_state="DISABLED"
160 )
161
162 with patch.object(ONUDevice.objects, "get_items") as onu_objects, \
163 patch.object(RCORDSubscriber, "save", autospec=True) as subscriber_save, \
164 patch.object(ONUDevice, "save") as onu_save:
165
166 onu_objects.return_value = [onu]
167
168 self.policy.handle_update(self.si)
169
170 self.assertEqual(onu.admin_state, "ENABLED")
171 onu_save.assert_called()
172 self.assertEqual(subscriber_save.call_count, 0)
173
Matteo Scandoloe8c33d62018-08-16 14:37:24 -0700174 def test_subscriber_awaiting_status_onu_state_disabled(self):
Matteo Scandoload0c1752018-08-09 15:47:16 -0700175 self.si.valid = "valid"
Matteo Scandoload0c1752018-08-09 15:47:16 -0700176 self.si.backend_code = 1
Matteo Scandoloe8c33d62018-08-16 14:37:24 -0700177 self.si.serial_number = "BRCM1234"
178 self.si.onu_state = "DISABLED"
Matteo Scandoload0c1752018-08-09 15:47:16 -0700179
180 onu = ONUDevice(
181 serial_number=self.si.serial_number,
Matteo Scandoloe8c33d62018-08-16 14:37:24 -0700182 admin_state="DISABLED"
183 )
184
185 subscriber = RCORDSubscriber(
186 onu_device=self.si.serial_number,
187 status='enabled'
Matteo Scandoload0c1752018-08-09 15:47:16 -0700188 )
189
190 with patch.object(ONUDevice.objects, "get_items") as onu_objects, \
Matteo Scandoloe8c33d62018-08-16 14:37:24 -0700191 patch.object(RCORDSubscriber.objects, "get_items") as subscriber_objects, \
192 patch.object(RCORDSubscriber, "save") as subscriber_save:
Matteo Scandoload0c1752018-08-09 15:47:16 -0700193 onu_objects.return_value = [onu]
Matteo Scandoloe8c33d62018-08-16 14:37:24 -0700194 subscriber_objects.return_value = [subscriber]
Matteo Scandoload0c1752018-08-09 15:47:16 -0700195
196 self.policy.handle_update(self.si)
Matteo Scandoloe8c33d62018-08-16 14:37:24 -0700197 self.assertEqual(subscriber.status, "awaiting-auth")
198 subscriber_save.assert_called()
Matteo Scandoload0c1752018-08-09 15:47:16 -0700199
Matteo Scandoloe8c33d62018-08-16 14:37:24 -0700200 def test_subscriber_enable_status_auth_state_approved(self):
Matteo Scandoload0c1752018-08-09 15:47:16 -0700201 self.si.valid = "valid"
Matteo Scandoload0c1752018-08-09 15:47:16 -0700202 self.si.backend_code = 1
Matteo Scandoloe8c33d62018-08-16 14:37:24 -0700203 self.si.serial_number = "BRCM1234"
204 self.si.onu_state = "ENABLED"
Matteo Scandoload0c1752018-08-09 15:47:16 -0700205 self.si.authentication_state = "APPROVED"
206
207 onu = ONUDevice(
208 serial_number=self.si.serial_number,
209 admin_state="ENABLED"
210 )
211
Matteo Scandoloe8c33d62018-08-16 14:37:24 -0700212 subscriber = RCORDSubscriber(
213 onu_device=self.si.serial_number,
214 status='awaiting-auth'
215 )
Matteo Scandoload0c1752018-08-09 15:47:16 -0700216
Matteo Scandoloe8c33d62018-08-16 14:37:24 -0700217 with patch.object(ONUDevice.objects, "get_items") as onu_objects, \
218 patch.object(RCORDSubscriber.objects, "get_items") as subscriber_objects, \
219 patch.object(RCORDSubscriber, "save") as subscriber_save:
Matteo Scandoload0c1752018-08-09 15:47:16 -0700220 onu_objects.return_value = [onu]
Matteo Scandoloe8c33d62018-08-16 14:37:24 -0700221 subscriber_objects.return_value = [subscriber]
Matteo Scandoload0c1752018-08-09 15:47:16 -0700222
223 self.policy.handle_update(self.si)
Matteo Scandoloe8c33d62018-08-16 14:37:24 -0700224 self.assertEqual(subscriber.status, "enabled")
225 subscriber_save.assert_called()
Matteo Scandoload0c1752018-08-09 15:47:16 -0700226
Matteo Scandoload0c1752018-08-09 15:47:16 -0700227if __name__ == '__main__':
228 unittest.main()
229