Gitiles
Code Review Sign In
gerrit.opencord.org / fabric / 1537787bd3edc5f55b08d5ce5fa944c622cd7d80 / . / xos / synchronizer / event_steps / test_kubernetes_event.py
blob: 22b0c95ccfcd09665ce04cd983d8049679d91e11 [file] [log] [blame]
Scott Bakerb2e8aa12018-08-15 16:04:41 -0700[diff] [blame]1# 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
16import json
17import functools
18from mock import patch, call, Mock, PropertyMock
19import requests_mock
20
21import os, sys
22
23# Hack to load synchronizer framework
24test_path=os.path.abspath(os.path.dirname(os.path.realpath(__file__)))
25xos_dir=os.path.join(test_path, "../../..")
26if not os.path.exists(os.path.join(test_path, "new_base")):
27 xos_dir=os.path.join(test_path, "../../../../../../orchestration/xos/xos")
28 services_dir = os.path.join(xos_dir, "../../xos_services")
29sys.path.append(xos_dir)
30sys.path.append(os.path.join(xos_dir, 'synchronizers', 'new_base'))
31# END Hack to load synchronizer framework
32
33# generate model from xproto
34def get_models_fn(service_name, xproto_name):
35 name = os.path.join(service_name, "xos", xproto_name)
36 if os.path.exists(os.path.join(services_dir, name)):
37 return name
38 else:
39 name = os.path.join(service_name, "xos", "synchronizer", "models", xproto_name)
40 if os.path.exists(os.path.join(services_dir, name)):
41 return name
42 raise Exception("Unable to find service=%s xproto=%s" % (service_name, xproto_name))
43# END generate model from xproto
44
45class TestKubernetesEvent(unittest.TestCase):
46
47 def setUp(self):
48 global DeferredException
49
50 self.sys_path_save = sys.path
51 sys.path.append(xos_dir)
52 sys.path.append(os.path.join(xos_dir, 'synchronizers', 'new_base'))
53
54 # Setting up the config module
55 from xosconfig import Config
56 config = os.path.join(test_path, "../test_config.yaml")
57 Config.clear()
58 Config.init(config, "synchronizer-config-schema.yaml")
59 # END Setting up the config module
60
61 from synchronizers.new_base.mock_modelaccessor_build import build_mock_modelaccessor
62 build_mock_modelaccessor(xos_dir, services_dir, [
63 get_models_fn("fabric", "fabric.xproto"),
64 get_models_fn("onos-service", "onos.xproto"),
65 ])
66 import synchronizers.new_base.mock_modelaccessor
67 reload(synchronizers.new_base.mock_modelaccessor) # in case nose2 loaded it in a previous test
68 import synchronizers.new_base.modelaccessor
69 reload(synchronizers.new_base.modelaccessor) # in case nose2 loaded it in a previous test
70 from kubernetes_event import model_accessor
71 from mock_modelaccessor import MockObjectList
72
73 from kubernetes_event import KubernetesPodDetailsEventStep
74
75 # import all class names to globals
76 for (k, v) in model_accessor.all_model_classes.items():
77 globals()[k] = v
78
79 self.event_step = KubernetesPodDetailsEventStep
80
81 self.onos = ONOSService(name="myonos",
82 id=1111,
83 rest_hostname = "onos-url",
84 rest_port = "8181",
85 rest_username = "karaf",
86 rest_password = "karaf",
87 backend_code=1,
88 backend_status="succeeded")
89
90 self.fabric_service = FabricService(name="fabric",
91 id=1112,
92 backend_code=1,
93 backend_status="succeeded",
94 provider_services=[self.onos])
95
96 self.switch = Switch(name="switch1",
97 backend_code=1,
98 backend_status="succeeded")
99
100 self.port1 = SwitchPort(name="switch1port1",
101 switch=self.switch,
102 backend_code=1,
103 backend_status="succeeded")
104
105 self.port2 = SwitchPort(name="switch1port2",
106 switch=self.switch,
107 backend_code=1,
108 backend_status="succeeded")
109
110 self.switch.ports = MockObjectList([self.port1, self.port2])
111
112 self.log = Mock()
113
114 def tearDown(self):
115 sys.path = self.sys_path_save
116
117 def test_process_event(self):
118 with patch.object(FabricService.objects, "get_items") as fabric_service_objects, \
119 patch.object(Service.objects, "get_items") as service_objects, \
120 patch.object(Switch.objects, "get_items") as switch_objects, \
121 patch.object(Switch, "save", autospec=True) as switch_save, \
122 patch.object(SwitchPort, "save", autospec=True) as switchport_save:
123 fabric_service_objects.return_value = [self.fabric_service]
124 service_objects.return_value = [self.onos, self.fabric_service]
125 switch_objects.return_value = [self.switch]
126
127 event_dict = {"status": "created",
128 "labels": {"xos_service": "myonos"}}
129 event = Mock()
130 event.value = json.dumps(event_dict)
131
132 step = self.event_step(log=self.log)
133 step.process_event(event)
134
135 self.assertEqual(self.switch.backend_code, 0)
136 self.assertEqual(self.switch.backend_status, "resynchronize due to kubernetes event")
137
138 switch_save.assert_called_with(self.switch, update_fields=["updated", "backend_code", "backend_status"],
139 always_update_timestamp=True)
140
141 self.assertEqual(self.port1.backend_code, 0)
142 self.assertEqual(self.port1.backend_status, "resynchronize due to kubernetes event")
143
144 self.assertEqual(self.port2.backend_code, 0)
145 self.assertEqual(self.port2.backend_status, "resynchronize due to kubernetes event")
146
147 switchport_save.assert_has_calls([call(self.port1, update_fields=["updated", "backend_code", "backend_status"],
148 always_update_timestamp=True),
149 call(self.port2, update_fields=["updated", "backend_code", "backend_status"],
150 always_update_timestamp=True)])
151
152 def test_process_event_unknownstatus(self):
153 with patch.object(FabricService.objects, "get_items") as fabric_service_objects, \
154 patch.object(Service.objects, "get_items") as service_objects, \
155 patch.object(Switch.objects, "get_items") as switch_objects, \
156 patch.object(Switch, "save") as switch_save, \
157 patch.object(SwitchPort, "save") as switchport_save:
158 fabric_service_objects.return_value = [self.fabric_service]
159 service_objects.return_value = [self.onos, self.fabric_service]
160 switch_objects.return_value = [self.switch]
161
162 event_dict = {"status": "something_else",
163 "labels": {"xos_service": "myonos"}}
164 event = Mock()
165 event.value = json.dumps(event_dict)
166
167 step = self.event_step(log=self.log)
168 step.process_event(event)
169
170 self.assertEqual(self.switch.backend_code, 1)
171 self.assertEqual(self.switch.backend_status, "succeeded")
172
173 switch_save.assert_not_called()
174
175 self.assertEqual(self.port1.backend_code, 1)
176 self.assertEqual(self.port1.backend_status, "succeeded")
177
178 self.assertEqual(self.port2.backend_code, 1)
179 self.assertEqual(self.port2.backend_status, "succeeded")
180
181 switchport_save.assert_not_called()
182
183 def test_process_event_unknownservice(self):
184 with patch.object(FabricService.objects, "get_items") as fabric_service_objects, \
185 patch.object(Service.objects, "get_items") as service_objects, \
186 patch.object(Switch.objects, "get_items") as switch_objects, \
187 patch.object(Switch, "save") as switch_save, \
188 patch.object(SwitchPort, "save") as switchport_save:
189 fabric_service_objects.return_value = [self.fabric_service]
190 service_objects.return_value = [self.onos, self.fabric_service]
191 switch_objects.return_value = [self.switch]
192
193 event_dict = {"status": "created",
194 "labels": {"xos_service": "something_else"}}
195 event = Mock()
196 event.value = json.dumps(event_dict)
197
198 step = self.event_step(log=self.log)
199 step.process_event(event)
200
201 self.assertEqual(self.switch.backend_code, 1)
202 self.assertEqual(self.switch.backend_status, "succeeded")
203
204 switch_save.assert_not_called()
205
206 self.assertEqual(self.port1.backend_code, 1)
207 self.assertEqual(self.port1.backend_status, "succeeded")
208
209 self.assertEqual(self.port2.backend_code, 1)
210 self.assertEqual(self.port2.backend_status, "succeeded")
211
212 switchport_save.assert_not_called()
213
214 def test_process_event_nolabels(self):
215 with patch.object(FabricService.objects, "get_items") as fabric_service_objects, \
216 patch.object(Service.objects, "get_items") as service_objects, \
217 patch.object(Switch.objects, "get_items") as switch_objects, \
218 patch.object(Switch, "save") as switch_save, \
219 patch.object(SwitchPort, "save") as switchport_save:
220 fabric_service_objects.return_value = [self.fabric_service]
221 service_objects.return_value = [self.onos, self.fabric_service]
222 switch_objects.return_value = [self.switch]
223
224 event_dict = {"status": "created"}
225 event = Mock()
226 event.value = json.dumps(event_dict)
227
228 step = self.event_step(log=self.log)
229 step.process_event(event)
230
231 self.assertEqual(self.switch.backend_code, 1)
232 self.assertEqual(self.switch.backend_status, "succeeded")
233
234 switch_save.assert_not_called()
235
236 self.assertEqual(self.port1.backend_code, 1)
237 self.assertEqual(self.port1.backend_status, "succeeded")
238
239 self.assertEqual(self.port2.backend_code, 1)
240 self.assertEqual(self.port2.backend_status, "succeeded")
241
242 switchport_save.assert_not_called()
243
244if __name__ == '__main__':
245 unittest.main()
246
247
248