blob: bec5098d9486c8b75223a714183d5e715998c2ce [file] [log] [blame]
Matteo Scandolo04287a82018-10-02 15:25:26 -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
Matteo Scandolo04287a82018-10-02 15:25:26 -070023test_path=os.path.abspath(os.path.dirname(os.path.realpath(__file__)))
Matteo Scandolo04287a82018-10-02 15:25:26 -070024
25class TestKubernetesEvent(unittest.TestCase):
26
27 def setUp(self):
28 global DeferredException
29
30 self.sys_path_save = sys.path
Matteo Scandolo04287a82018-10-02 15:25:26 -070031
32 config = os.path.join(test_path, "../test_config.yaml")
33 from xosconfig import Config
34 Config.clear()
35 Config.init(config, 'synchronizer-config-schema.yaml')
36
Scott Baker47b47302019-01-30 16:55:07 -080037 from xossynchronizer.mock_modelaccessor_build import mock_modelaccessor_config
38 mock_modelaccessor_config(test_path, [("olt-service", "volt.xproto"),
39 ("vsg", "vsg.xproto"),
Matteo Scandoloe04a4882019-02-14 10:22:24 -080040 ("rcord", "rcord.xproto"),
Scott Baker47b47302019-01-30 16:55:07 -080041 ("onos-service", "onos.xproto"),])
Matteo Scandolo04287a82018-10-02 15:25:26 -070042
Scott Baker47b47302019-01-30 16:55:07 -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
Matteo Scandolo04287a82018-10-02 15:25:26 -070047
Scott Baker47b47302019-01-30 16:55:07 -080048 from xossynchronizer.modelaccessor import model_accessor
Matteo Scandolo04287a82018-10-02 15:25:26 -070049 from mock_modelaccessor import MockObjectList
50
51 from kubernetes_event import KubernetesPodDetailsEventStep
52
53 # import all class names to globals
54 for (k, v) in model_accessor.all_model_classes.items():
55 globals()[k] = v
56
Scott Baker47b47302019-01-30 16:55:07 -080057 self.model_accessor = model_accessor
Matteo Scandolo04287a82018-10-02 15:25:26 -070058 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.fcservice = VOLTService(name="myoltservice",
70 id=1112,
71 backend_code=1,
72 backend_status="succeeded",
73 subscriber_services=[self.onos])
74
75 self.fcsi1 = VOLTServiceInstance(name="myfcsi1",
76 owner=self.fcservice,
77 backend_code=1,
78 backend_status="succeeded")
79
80 self.fcsi2 = VOLTServiceInstance(name="myfcsi2",
81 owner=self.fcservice,
82 backend_code=1,
83 backend_status="succeeded")
84
85 self.fcservice.service_instances = MockObjectList([self.fcsi1, self.fcsi2])
86
87 self.log = Mock()
88
89 def tearDown(self):
90 sys.path = self.sys_path_save
91
92 def test_process_event(self):
93 with patch.object(VOLTService.objects, "get_items") as fcservice_objects, \
94 patch.object(Service.objects, "get_items") as service_objects, \
95 patch.object(VOLTServiceInstance, "save", autospec=True) as fcsi_save:
96 fcservice_objects.return_value = [self.fcservice]
97 service_objects.return_value = [self.onos, self.fcservice]
98
99 event_dict = {"status": "created",
100 "labels": {"xos_service": "myonos"}}
101 event = Mock()
102 event.value = json.dumps(event_dict)
103
Scott Baker47b47302019-01-30 16:55:07 -0800104 step = self.event_step(model_accessor=self.model_accessor, log=self.log)
Matteo Scandolo04287a82018-10-02 15:25:26 -0700105 step.process_event(event)
106
107 self.assertEqual(self.fcsi1.backend_code, 0)
108 self.assertEqual(self.fcsi1.backend_status, "resynchronize due to kubernetes event")
109
110 self.assertEqual(self.fcsi2.backend_code, 0)
111 self.assertEqual(self.fcsi2.backend_status, "resynchronize due to kubernetes event")
112
113 fcsi_save.assert_has_calls([call(self.fcsi1, update_fields=["updated", "backend_code", "backend_status"],
114 always_update_timestamp=True),
115 call(self.fcsi2, update_fields=["updated", "backend_code", "backend_status"],
116 always_update_timestamp=True)])
117
118 def test_process_event_unknownstatus(self):
119 with patch.object(VOLTService.objects, "get_items") as fcservice_objects, \
120 patch.object(Service.objects, "get_items") as service_objects, \
121 patch.object(VOLTServiceInstance, "save") as fcsi_save:
122 fcservice_objects.return_value = [self.fcservice]
123 service_objects.return_value = [self.onos, self.fcservice]
124
125 event_dict = {"status": "something_else",
126 "labels": {"xos_service": "myonos"}}
127 event = Mock()
128 event.value = json.dumps(event_dict)
129
Scott Baker47b47302019-01-30 16:55:07 -0800130 step = self.event_step(model_accessor=self.model_accessor, log=self.log)
Matteo Scandolo04287a82018-10-02 15:25:26 -0700131 step.process_event(event)
132
133 self.assertEqual(self.fcsi1.backend_code, 1)
134 self.assertEqual(self.fcsi1.backend_status, "succeeded")
135
136 self.assertEqual(self.fcsi2.backend_code, 1)
137 self.assertEqual(self.fcsi2.backend_status, "succeeded")
138
139 fcsi_save.assert_not_called()
140
141 def test_process_event_unknownservice(self):
142 with patch.object(VOLTService.objects, "get_items") as fcservice_objects, \
143 patch.object(Service.objects, "get_items") as service_objects, \
144 patch.object(VOLTServiceInstance, "save") as fcsi_save:
145 fcservice_objects.return_value = [self.fcservice]
146 service_objects.return_value = [self.onos, self.fcservice]
147
148 event_dict = {"status": "created",
149 "labels": {"xos_service": "something_else"}}
150 event = Mock()
151 event.value = json.dumps(event_dict)
152
Scott Baker47b47302019-01-30 16:55:07 -0800153 step = self.event_step(model_accessor=self.model_accessor, log=self.log)
Matteo Scandolo04287a82018-10-02 15:25:26 -0700154 step.process_event(event)
155
156 self.assertEqual(self.fcsi1.backend_code, 1)
157 self.assertEqual(self.fcsi1.backend_status, "succeeded")
158
159 self.assertEqual(self.fcsi2.backend_code, 1)
160 self.assertEqual(self.fcsi2.backend_status, "succeeded")
161
162 fcsi_save.assert_not_called()
163
164 def test_process_event_nolabels(self):
165 with patch.object(VOLTService.objects, "get_items") as fcservice_objects, \
166 patch.object(Service.objects, "get_items") as service_objects, \
167 patch.object(VOLTServiceInstance, "save") as fcsi_save:
168 fcservice_objects.return_value = [self.fcservice]
169 service_objects.return_value = [self.onos, self.fcservice]
170
171 event_dict = {"status": "created"}
172 event = Mock()
173 event.value = json.dumps(event_dict)
174
Scott Baker47b47302019-01-30 16:55:07 -0800175 step = self.event_step(model_accessor=self.model_accessor, log=self.log)
Matteo Scandolo04287a82018-10-02 15:25:26 -0700176 step.process_event(event)
177
178 self.assertEqual(self.fcsi1.backend_code, 1)
179 self.assertEqual(self.fcsi1.backend_status, "succeeded")
180
181 self.assertEqual(self.fcsi2.backend_code, 1)
182 self.assertEqual(self.fcsi2.backend_status, "succeeded")
183
184 fcsi_save.assert_not_called()
185
186if __name__ == '__main__':
187 unittest.main()
188
189
190