blob: ae2915d08a524ebc0162ec9c2bae5ef518647c4d [file] [log] [blame]
Scott Bakerb2e8aa12018-08-15 16:04:41 -07001# 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
Scott Bakerb2e8aa12018-08-15 16:04:41 -070023test_path=os.path.abspath(os.path.dirname(os.path.realpath(__file__)))
Scott Bakerb2e8aa12018-08-15 16:04:41 -070024
25class TestKubernetesEvent(unittest.TestCase):
26
27 def setUp(self):
28 global DeferredException
29
30 self.sys_path_save = sys.path
Scott Bakerb2e8aa12018-08-15 16:04:41 -070031
32 # Setting up the config module
33 from xosconfig import Config
34 config = os.path.join(test_path, "../test_config.yaml")
35 Config.clear()
36 Config.init(config, "synchronizer-config-schema.yaml")
37 # END Setting up the config module
38
Scott Baker382366d2019-02-04 10:58:43 -080039 from xossynchronizer.mock_modelaccessor_build import mock_modelaccessor_config
40 mock_modelaccessor_config(test_path, [("fabric", "fabric.xproto"),
41 ("onos-service", "onos.xproto")])
Scott Bakerb2e8aa12018-08-15 16:04:41 -070042
Scott Baker382366d2019-02-04 10:58:43 -080043 import xossynchronizer.modelaccessor
44 import mock_modelaccessor
45 reload(mock_modelaccessor) # in case nose2 loaded it in a previous test
46 reload(xossynchronizer.modelaccessor) # in case nose2 loaded it in a previous test
47
48 from xossynchronizer.modelaccessor import model_accessor
49 self.model_accessor = model_accessor
50
51 from mock_modelaccessor import MockObjectList
Scott Bakerb2e8aa12018-08-15 16:04:41 -070052 from kubernetes_event import KubernetesPodDetailsEventStep
53
54 # import all class names to globals
55 for (k, v) in model_accessor.all_model_classes.items():
56 globals()[k] = v
57
58 self.event_step = KubernetesPodDetailsEventStep
59
60 self.onos = ONOSService(name="myonos",
61 id=1111,
62 rest_hostname = "onos-url",
63 rest_port = "8181",
64 rest_username = "karaf",
65 rest_password = "karaf",
66 backend_code=1,
67 backend_status="succeeded")
68
69 self.fabric_service = FabricService(name="fabric",
70 id=1112,
71 backend_code=1,
72 backend_status="succeeded",
73 provider_services=[self.onos])
74
75 self.switch = Switch(name="switch1",
76 backend_code=1,
77 backend_status="succeeded")
78
79 self.port1 = SwitchPort(name="switch1port1",
80 switch=self.switch,
81 backend_code=1,
82 backend_status="succeeded")
83
84 self.port2 = SwitchPort(name="switch1port2",
85 switch=self.switch,
86 backend_code=1,
87 backend_status="succeeded")
88
89 self.switch.ports = MockObjectList([self.port1, self.port2])
90
91 self.log = Mock()
92
93 def tearDown(self):
94 sys.path = self.sys_path_save
95
96 def test_process_event(self):
97 with patch.object(FabricService.objects, "get_items") as fabric_service_objects, \
98 patch.object(Service.objects, "get_items") as service_objects, \
99 patch.object(Switch.objects, "get_items") as switch_objects, \
100 patch.object(Switch, "save", autospec=True) as switch_save, \
101 patch.object(SwitchPort, "save", autospec=True) as switchport_save:
102 fabric_service_objects.return_value = [self.fabric_service]
103 service_objects.return_value = [self.onos, self.fabric_service]
104 switch_objects.return_value = [self.switch]
105
106 event_dict = {"status": "created",
107 "labels": {"xos_service": "myonos"}}
108 event = Mock()
109 event.value = json.dumps(event_dict)
110
Scott Baker382366d2019-02-04 10:58:43 -0800111 step = self.event_step(model_accessor=self.model_accessor, log=self.log)
Scott Bakerb2e8aa12018-08-15 16:04:41 -0700112 step.process_event(event)
113
114 self.assertEqual(self.switch.backend_code, 0)
115 self.assertEqual(self.switch.backend_status, "resynchronize due to kubernetes event")
116
117 switch_save.assert_called_with(self.switch, update_fields=["updated", "backend_code", "backend_status"],
118 always_update_timestamp=True)
119
120 self.assertEqual(self.port1.backend_code, 0)
121 self.assertEqual(self.port1.backend_status, "resynchronize due to kubernetes event")
122
123 self.assertEqual(self.port2.backend_code, 0)
124 self.assertEqual(self.port2.backend_status, "resynchronize due to kubernetes event")
125
126 switchport_save.assert_has_calls([call(self.port1, update_fields=["updated", "backend_code", "backend_status"],
127 always_update_timestamp=True),
128 call(self.port2, update_fields=["updated", "backend_code", "backend_status"],
129 always_update_timestamp=True)])
130
131 def test_process_event_unknownstatus(self):
132 with patch.object(FabricService.objects, "get_items") as fabric_service_objects, \
133 patch.object(Service.objects, "get_items") as service_objects, \
134 patch.object(Switch.objects, "get_items") as switch_objects, \
135 patch.object(Switch, "save") as switch_save, \
136 patch.object(SwitchPort, "save") as switchport_save:
137 fabric_service_objects.return_value = [self.fabric_service]
138 service_objects.return_value = [self.onos, self.fabric_service]
139 switch_objects.return_value = [self.switch]
140
141 event_dict = {"status": "something_else",
142 "labels": {"xos_service": "myonos"}}
143 event = Mock()
144 event.value = json.dumps(event_dict)
145
Scott Baker382366d2019-02-04 10:58:43 -0800146 step = self.event_step(model_accessor=self.model_accessor, log=self.log)
Scott Bakerb2e8aa12018-08-15 16:04:41 -0700147 step.process_event(event)
148
149 self.assertEqual(self.switch.backend_code, 1)
150 self.assertEqual(self.switch.backend_status, "succeeded")
151
152 switch_save.assert_not_called()
153
154 self.assertEqual(self.port1.backend_code, 1)
155 self.assertEqual(self.port1.backend_status, "succeeded")
156
157 self.assertEqual(self.port2.backend_code, 1)
158 self.assertEqual(self.port2.backend_status, "succeeded")
159
160 switchport_save.assert_not_called()
161
162 def test_process_event_unknownservice(self):
163 with patch.object(FabricService.objects, "get_items") as fabric_service_objects, \
164 patch.object(Service.objects, "get_items") as service_objects, \
165 patch.object(Switch.objects, "get_items") as switch_objects, \
166 patch.object(Switch, "save") as switch_save, \
167 patch.object(SwitchPort, "save") as switchport_save:
168 fabric_service_objects.return_value = [self.fabric_service]
169 service_objects.return_value = [self.onos, self.fabric_service]
170 switch_objects.return_value = [self.switch]
171
172 event_dict = {"status": "created",
173 "labels": {"xos_service": "something_else"}}
174 event = Mock()
175 event.value = json.dumps(event_dict)
176
Scott Baker382366d2019-02-04 10:58:43 -0800177 step = self.event_step(model_accessor=self.model_accessor, log=self.log)
Scott Bakerb2e8aa12018-08-15 16:04:41 -0700178 step.process_event(event)
179
180 self.assertEqual(self.switch.backend_code, 1)
181 self.assertEqual(self.switch.backend_status, "succeeded")
182
183 switch_save.assert_not_called()
184
185 self.assertEqual(self.port1.backend_code, 1)
186 self.assertEqual(self.port1.backend_status, "succeeded")
187
188 self.assertEqual(self.port2.backend_code, 1)
189 self.assertEqual(self.port2.backend_status, "succeeded")
190
191 switchport_save.assert_not_called()
192
193 def test_process_event_nolabels(self):
194 with patch.object(FabricService.objects, "get_items") as fabric_service_objects, \
195 patch.object(Service.objects, "get_items") as service_objects, \
196 patch.object(Switch.objects, "get_items") as switch_objects, \
197 patch.object(Switch, "save") as switch_save, \
198 patch.object(SwitchPort, "save") as switchport_save:
199 fabric_service_objects.return_value = [self.fabric_service]
200 service_objects.return_value = [self.onos, self.fabric_service]
201 switch_objects.return_value = [self.switch]
202
203 event_dict = {"status": "created"}
204 event = Mock()
205 event.value = json.dumps(event_dict)
206
Scott Baker382366d2019-02-04 10:58:43 -0800207 step = self.event_step(model_accessor=self.model_accessor, log=self.log)
Scott Bakerb2e8aa12018-08-15 16:04:41 -0700208 step.process_event(event)
209
210 self.assertEqual(self.switch.backend_code, 1)
211 self.assertEqual(self.switch.backend_status, "succeeded")
212
213 switch_save.assert_not_called()
214
215 self.assertEqual(self.port1.backend_code, 1)
216 self.assertEqual(self.port1.backend_status, "succeeded")
217
218 self.assertEqual(self.port2.backend_code, 1)
219 self.assertEqual(self.port2.backend_status, "succeeded")
220
221 switchport_save.assert_not_called()
222
223if __name__ == '__main__':
224 unittest.main()
225
226
227