blob: 18c81749c1ad2aa772c1d34db0121692a54b6cb6 [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
76 def test_enable_onu(self):
Matteo Scandoloea529092018-09-11 16:36:39 -070077 from helpers import AttHelpers
78 with patch.object(AttHelpers, "validate_onu") as validate_onu, \
79 patch.object(self.policy, "update_onu") as update_onu, \
80 patch.object(self.si, "save") as save_si:
81 validate_onu.return_value = [True, "valid onu"]
Matteo Scandoload0c1752018-08-09 15:47:16 -070082
Matteo Scandoloea529092018-09-11 16:36:39 -070083 self.policy.validate_onu_state(self.si)
Matteo Scandoload0c1752018-08-09 15:47:16 -070084
Matteo Scandoloea529092018-09-11 16:36:39 -070085 update_onu.assert_called_once()
86 update_onu.assert_called_with("BRCM1234", "ENABLED")
Matteo Scandoload0c1752018-08-09 15:47:16 -070087
Matteo Scandoloea529092018-09-11 16:36:39 -070088 self.assertIn("valid onu", self.si.status_message)
Matteo Scandoload0c1752018-08-09 15:47:16 -070089
Matteo Scandoloea529092018-09-11 16:36:39 -070090 def test_disable_onu(self):
91 from helpers import AttHelpers
92 with patch.object(AttHelpers, "validate_onu") as validate_onu, \
93 patch.object(self.policy, "update_onu") as update_onu, \
94 patch.object(self.si, "save") as save_si:
95 validate_onu.return_value = [False, "invalid onu"]
Matteo Scandoload0c1752018-08-09 15:47:16 -070096
Matteo Scandoloea529092018-09-11 16:36:39 -070097 self.policy.validate_onu_state(self.si)
98
99 update_onu.assert_called_once()
100 update_onu.assert_called_with("BRCM1234", "DISABLED")
101
102 self.assertIn("invalid onu", self.si.status_message)
103
104 def test_handle_update_validate_onu(self):
105 """
106 Testing that handle_update calls validate_onu with the correct parameters
107 when necessary
108 """
109 with patch.object(self.policy, "validate_onu_state") as validate_onu_state, \
110 patch.object(self.policy, "update_onu") as update_onu, \
111 patch.object(self.policy, "get_subscriber") as get_subscriber:
112 update_onu.return_value = None
113 get_subscriber.return_value = None
114
115 self.si.onu_state = "AWAITING"
Matteo Scandoload0c1752018-08-09 15:47:16 -0700116 self.policy.handle_update(self.si)
Matteo Scandoloea529092018-09-11 16:36:39 -0700117 validate_onu_state.assert_called_with(self.si)
Matteo Scandoload0c1752018-08-09 15:47:16 -0700118
Matteo Scandoloea529092018-09-11 16:36:39 -0700119 self.si.onu_state = "ENABLED"
Matteo Scandoload0c1752018-08-09 15:47:16 -0700120 self.policy.handle_update(self.si)
Matteo Scandoloea529092018-09-11 16:36:39 -0700121 validate_onu_state.assert_called_with(self.si)
Matteo Scandoload0c1752018-08-09 15:47:16 -0700122
Matteo Scandoloea529092018-09-11 16:36:39 -0700123 self.si.onu_state = "DISABLED"
124 self.policy.handle_update(self.si)
125 self.assertEqual(validate_onu_state.call_count, 2)
Matteo Scandoload0c1752018-08-09 15:47:16 -0700126
Matteo Scandoloea529092018-09-11 16:36:39 -0700127 def test_get_subscriber(self):
128
129 sub = RCORDSubscriber(
130 onu_device="BRCM1234"
131 )
132
133 with patch.object(RCORDSubscriber.objects, "get_items") as get_subscribers:
134 get_subscribers.return_value = [sub]
135
136 res = self.policy.get_subscriber("BRCM1234")
137 self.assertEqual(res, sub)
138
139 res = self.policy.get_subscriber("brcm1234")
140 self.assertEqual(res, sub)
141
142 res = self.policy.get_subscriber("foo")
143 self.assertEqual(res, None)
144
145 def test_update_subscriber(self):
146
147 sub = RCORDSubscriber(
148 onu_device="BRCM1234"
149 )
150
151 self.si.status_message = "some content"
152
153 with patch.object(sub, "save") as sub_save:
154 self.si.authentication_state = "AWAITING"
155 self.policy.update_subscriber(sub, self.si)
156 self.assertEqual(sub.status, "awaiting-auth")
157 self.assertIn("Awaiting Authentication", self.si.status_message)
158 sub_save.assert_called()
159 sub_save.reset_mock()
160
161 self.si.authentication_state = "REQUESTED"
162 self.policy.update_subscriber(sub, self.si)
163 self.assertEqual(sub.status, "awaiting-auth")
164 self.assertIn("Authentication requested", self.si.status_message)
165 sub_save.assert_called()
166 sub_save.reset_mock()
167
168 self.si.authentication_state = "STARTED"
169 self.policy.update_subscriber(sub, self.si)
170 self.assertEqual(sub.status, "awaiting-auth")
171 self.assertIn("Authentication started", self.si.status_message)
172 sub_save.assert_called()
173 sub_save.reset_mock()
174
175 self.si.authentication_state = "APPROVED"
176 self.policy.update_subscriber(sub, self.si)
177 self.assertEqual(sub.status, "enabled")
178 self.assertIn("Authentication succeded", self.si.status_message)
179 sub_save.assert_called()
180 sub_save.reset_mock()
181
182 self.si.authentication_state = "DENIED"
183 self.policy.update_subscriber(sub, self.si)
184 self.assertEqual(sub.status, "auth-failed")
185 self.assertIn("Authentication denied", self.si.status_message)
186 sub_save.assert_called()
187 sub_save.reset_mock()
188
189 def test_handle_update_subscriber(self):
Matteo Scandoloe8c33d62018-08-16 14:37:24 -0700190 self.si.onu_state = "DISABLED"
Matteo Scandoload0c1752018-08-09 15:47:16 -0700191
Matteo Scandoloea529092018-09-11 16:36:39 -0700192 sub = RCORDSubscriber(
193 onu_device="BRCM1234"
Matteo Scandoloe8c33d62018-08-16 14:37:24 -0700194 )
195
Matteo Scandoloea529092018-09-11 16:36:39 -0700196 with patch.object(self.policy, "get_subscriber") as get_subscriber, \
197 patch.object(self.policy, "update_subscriber") as update_subscriber:
Matteo Scandoload0c1752018-08-09 15:47:16 -0700198
Matteo Scandoloea529092018-09-11 16:36:39 -0700199 get_subscriber.return_value = None
Matteo Scandoload0c1752018-08-09 15:47:16 -0700200 self.policy.handle_update(self.si)
Matteo Scandoloea529092018-09-11 16:36:39 -0700201 self.assertEqual(update_subscriber.call_count, 0)
Matteo Scandoload0c1752018-08-09 15:47:16 -0700202
Matteo Scandoloea529092018-09-11 16:36:39 -0700203 get_subscriber.return_value = sub
Matteo Scandoload0c1752018-08-09 15:47:16 -0700204 self.policy.handle_update(self.si)
Matteo Scandoloea529092018-09-11 16:36:39 -0700205 update_subscriber.assert_called_with(sub, self.si)
206
Matteo Scandoload0c1752018-08-09 15:47:16 -0700207
Matteo Scandoload0c1752018-08-09 15:47:16 -0700208if __name__ == '__main__':
209 unittest.main()
210