blob: 3bbc4ec178e378c6405b3f7a146107e45958ca70 [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"),
Matteo Scandoloe04a4882019-02-14 10:22:24 -080039 ("rcord", "rcord.xproto"),
Matteo Scandolo35207b72019-05-10 08:46:48 -070040 ("onos-service", "onos.xproto")])
Matteo Scandolo04287a82018-10-02 15:25:26 -070041
Scott Baker47b47302019-01-30 16:55:07 -080042 import xossynchronizer.modelaccessor
43 import mock_modelaccessor
44 reload(mock_modelaccessor) # in case nose2 loaded it in a previous test
45 reload(xossynchronizer.modelaccessor) # in case nose2 loaded it in a previous test
Matteo Scandolo04287a82018-10-02 15:25:26 -070046
Scott Baker47b47302019-01-30 16:55:07 -080047 from xossynchronizer.modelaccessor import model_accessor
Matteo Scandolo04287a82018-10-02 15:25:26 -070048 from mock_modelaccessor import MockObjectList
49
50 from kubernetes_event import KubernetesPodDetailsEventStep
51
52 # import all class names to globals
53 for (k, v) in model_accessor.all_model_classes.items():
54 globals()[k] = v
55
Scott Baker47b47302019-01-30 16:55:07 -080056 self.model_accessor = model_accessor
Matteo Scandolo04287a82018-10-02 15:25:26 -070057 self.event_step = KubernetesPodDetailsEventStep
58
59 self.onos = ONOSService(name="myonos",
60 id=1111,
61 rest_hostname = "onos-url",
62 rest_port = "8181",
63 rest_username = "karaf",
64 rest_password = "karaf",
65 backend_code=1,
66 backend_status="succeeded")
67
68 self.fcservice = VOLTService(name="myoltservice",
69 id=1112,
70 backend_code=1,
71 backend_status="succeeded",
72 subscriber_services=[self.onos])
73
74 self.fcsi1 = VOLTServiceInstance(name="myfcsi1",
75 owner=self.fcservice,
76 backend_code=1,
77 backend_status="succeeded")
78
79 self.fcsi2 = VOLTServiceInstance(name="myfcsi2",
80 owner=self.fcservice,
81 backend_code=1,
82 backend_status="succeeded")
83
84 self.fcservice.service_instances = MockObjectList([self.fcsi1, self.fcsi2])
85
86 self.log = Mock()
87
88 def tearDown(self):
89 sys.path = self.sys_path_save
90
91 def test_process_event(self):
92 with patch.object(VOLTService.objects, "get_items") as fcservice_objects, \
93 patch.object(Service.objects, "get_items") as service_objects, \
94 patch.object(VOLTServiceInstance, "save", autospec=True) as fcsi_save:
95 fcservice_objects.return_value = [self.fcservice]
96 service_objects.return_value = [self.onos, self.fcservice]
97
98 event_dict = {"status": "created",
99 "labels": {"xos_service": "myonos"}}
100 event = Mock()
101 event.value = json.dumps(event_dict)
102
Scott Baker47b47302019-01-30 16:55:07 -0800103 step = self.event_step(model_accessor=self.model_accessor, log=self.log)
Matteo Scandolo04287a82018-10-02 15:25:26 -0700104 step.process_event(event)
105
106 self.assertEqual(self.fcsi1.backend_code, 0)
107 self.assertEqual(self.fcsi1.backend_status, "resynchronize due to kubernetes event")
108
109 self.assertEqual(self.fcsi2.backend_code, 0)
110 self.assertEqual(self.fcsi2.backend_status, "resynchronize due to kubernetes event")
111
112 fcsi_save.assert_has_calls([call(self.fcsi1, update_fields=["updated", "backend_code", "backend_status"],
113 always_update_timestamp=True),
114 call(self.fcsi2, update_fields=["updated", "backend_code", "backend_status"],
115 always_update_timestamp=True)])
116
117 def test_process_event_unknownstatus(self):
118 with patch.object(VOLTService.objects, "get_items") as fcservice_objects, \
119 patch.object(Service.objects, "get_items") as service_objects, \
120 patch.object(VOLTServiceInstance, "save") as fcsi_save:
121 fcservice_objects.return_value = [self.fcservice]
122 service_objects.return_value = [self.onos, self.fcservice]
123
124 event_dict = {"status": "something_else",
125 "labels": {"xos_service": "myonos"}}
126 event = Mock()
127 event.value = json.dumps(event_dict)
128
Scott Baker47b47302019-01-30 16:55:07 -0800129 step = self.event_step(model_accessor=self.model_accessor, log=self.log)
Matteo Scandolo04287a82018-10-02 15:25:26 -0700130 step.process_event(event)
131
132 self.assertEqual(self.fcsi1.backend_code, 1)
133 self.assertEqual(self.fcsi1.backend_status, "succeeded")
134
135 self.assertEqual(self.fcsi2.backend_code, 1)
136 self.assertEqual(self.fcsi2.backend_status, "succeeded")
137
138 fcsi_save.assert_not_called()
139
140 def test_process_event_unknownservice(self):
141 with patch.object(VOLTService.objects, "get_items") as fcservice_objects, \
142 patch.object(Service.objects, "get_items") as service_objects, \
143 patch.object(VOLTServiceInstance, "save") as fcsi_save:
144 fcservice_objects.return_value = [self.fcservice]
145 service_objects.return_value = [self.onos, self.fcservice]
146
147 event_dict = {"status": "created",
148 "labels": {"xos_service": "something_else"}}
149 event = Mock()
150 event.value = json.dumps(event_dict)
151
Scott Baker47b47302019-01-30 16:55:07 -0800152 step = self.event_step(model_accessor=self.model_accessor, log=self.log)
Matteo Scandolo04287a82018-10-02 15:25:26 -0700153 step.process_event(event)
154
155 self.assertEqual(self.fcsi1.backend_code, 1)
156 self.assertEqual(self.fcsi1.backend_status, "succeeded")
157
158 self.assertEqual(self.fcsi2.backend_code, 1)
159 self.assertEqual(self.fcsi2.backend_status, "succeeded")
160
161 fcsi_save.assert_not_called()
162
163 def test_process_event_nolabels(self):
164 with patch.object(VOLTService.objects, "get_items") as fcservice_objects, \
165 patch.object(Service.objects, "get_items") as service_objects, \
166 patch.object(VOLTServiceInstance, "save") as fcsi_save:
167 fcservice_objects.return_value = [self.fcservice]
168 service_objects.return_value = [self.onos, self.fcservice]
169
170 event_dict = {"status": "created"}
171 event = Mock()
172 event.value = json.dumps(event_dict)
173
Scott Baker47b47302019-01-30 16:55:07 -0800174 step = self.event_step(model_accessor=self.model_accessor, log=self.log)
Matteo Scandolo04287a82018-10-02 15:25:26 -0700175 step.process_event(event)
176
177 self.assertEqual(self.fcsi1.backend_code, 1)
178 self.assertEqual(self.fcsi1.backend_status, "succeeded")
179
180 self.assertEqual(self.fcsi2.backend_code, 1)
181 self.assertEqual(self.fcsi2.backend_status, "succeeded")
182
183 fcsi_save.assert_not_called()
184
185if __name__ == '__main__':
186 unittest.main()
187
188
189