1 # -*- coding: utf-8 -*-
3 # Copyright 2018 Whitestack, LLC
4 # *************************************************************
6 # This file is part of OSM Monitoring module
7 # All Rights Reserved to Whitestack, LLC
9 # Licensed under the Apache License, Version 2.0 (the "License"); you may
10 # not use this file except in compliance with the License. You may obtain
11 # a copy of the License at
13 # http://www.apache.org/licenses/LICENSE-2.0
15 # Unless required by applicable law or agreed to in writing, software
16 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
17 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
18 # License for the specific language governing permissions and limitations
21 # For those usages not covered by the Apache License, Version 2.0 please
22 # contact: bdiaz@whitestack.com or glavado@whitestack.com
26 from unittest
import mock
28 from osm_policy_module
.alarming
.service
import AlarmingService
29 from osm_policy_module
.autoscaling
.service
import AutoscalingService
30 from osm_policy_module
.common
.common_db_client
import CommonDbClient
31 from osm_policy_module
.core
.agent
import PolicyModuleAgent
32 from osm_policy_module
.core
.config
import Config
33 from osm_policy_module
.healing
.service
import HealingService
36 class PolicyAgentTest(unittest
.TestCase
):
38 self
.loop
= asyncio
.new_event_loop()
40 @mock.patch
.object(CommonDbClient
, "__init__", lambda *args
, **kwargs
: None)
41 @mock.patch("osm_policy_module.alarming.service.MonClient")
42 @mock.patch("osm_policy_module.alarming.service.LcmClient")
43 @mock.patch("osm_policy_module.autoscaling.service.MonClient")
44 @mock.patch("osm_policy_module.autoscaling.service.LcmClient")
45 @mock.patch("osm_policy_module.healing.service.MonClient")
46 @mock.patch("osm_policy_module.healing.service.LcmClient")
47 @mock.patch
.object(AutoscalingService
, "configure_scaling_groups")
48 @mock.patch
.object(AlarmingService
, "configure_vnf_alarms")
49 @mock.patch
.object(HealingService
, "configure_healing_alarms")
50 @mock.patch
.object(AutoscalingService
, "delete_orphaned_alarms")
51 @mock.patch
.object(CommonDbClient
, "get_nslcmop")
52 def test_handle_instantiated(
55 delete_orphaned_alarms
,
56 configure_healing_alarms
,
58 configure_scaling_groups
,
59 autoscaling_lcm_client
,
60 autoscaling_mon_client
,
66 async def mock_configure_scaling_groups(nsr_id
):
69 async def mock_configure_vnf_alarms(nsr_id
):
72 async def mock_configure_healing_alarms(nsr_id
):
75 async def mock_delete_orphaned_alarms(nsr_id
):
79 agent
= PolicyModuleAgent(config
, self
.loop
)
80 assert autoscaling_lcm_client
.called
81 assert autoscaling_mon_client
.called
82 assert alarming_lcm_client
.called
83 assert alarming_mon_client
.called
84 assert healing_lcm_client
.called
85 assert healing_mon_client
.called
87 "nslcmop_id": "test_id",
90 "operationState": "COMPLETED",
91 "nsInstanceId": "test_nsr_id",
93 nslcmop_failed
= {"operationState": "FAILED", "nsInstanceId": "test_nsr_id"}
94 configure_scaling_groups
.side_effect
= mock_configure_scaling_groups
95 configure_vnf_alarms
.side_effect
= mock_configure_vnf_alarms
96 configure_healing_alarms
.side_effect
= mock_configure_healing_alarms
97 delete_orphaned_alarms
.side_effect
= mock_delete_orphaned_alarms
99 get_nslcmop
.return_value
= nslcmop_completed
100 self
.loop
.run_until_complete(agent
._handle
_instantiated
(content
))
101 configure_scaling_groups
.assert_called_with("test_nsr_id")
102 configure_scaling_groups
.reset_mock()
104 get_nslcmop
.return_value
= nslcmop_failed
105 self
.loop
.run_until_complete(agent
._handle
_instantiated
(content
))
106 configure_scaling_groups
.assert_not_called()
108 @mock.patch
.object(CommonDbClient
, "__init__", lambda *args
, **kwargs
: None)
109 @mock.patch("osm_policy_module.alarming.service.MonClient")
110 @mock.patch("osm_policy_module.alarming.service.LcmClient")
111 @mock.patch("osm_policy_module.autoscaling.service.MonClient")
112 @mock.patch("osm_policy_module.autoscaling.service.LcmClient")
113 @mock.patch
.object(AutoscalingService
, "configure_scaling_groups")
114 @mock.patch
.object(AlarmingService
, "configure_vnf_alarms")
115 @mock.patch
.object(AutoscalingService
, "delete_scaling_groups")
116 @mock.patch
.object(AlarmingService
, "delete_vnf_alarms")
117 def test_handle_policy_update(
120 delete_scaling_groups
,
121 configure_vnf_alarms
,
122 configure_scaling_groups
,
123 autoscaling_lcm_client
,
124 autoscaling_mon_client
,
128 async def mock_delete_scaling_groups(nsr_id
, vnf_member_index
):
131 async def mock_delete_vnf_alarms(nsr_id
, vnf_member_index
):
134 async def mock_configure_scaling_groups(nsr_id
, vnf_member_index
):
137 async def mock_configure_vnf_alarms(nsr_id
, vnf_member_index
):
141 agent
= PolicyModuleAgent(config
, self
.loop
)
142 assert autoscaling_lcm_client
.called
143 assert autoscaling_mon_client
.called
144 assert alarming_lcm_client
.called
145 assert alarming_mon_client
.called
147 "nsr_id": "test_nsr_id",
148 "vnf_member_index": "1",
149 "operationState": "COMPLETED"
152 "nsr_id": "test_nsr_id",
153 "vnf_member_index": "1",
154 "operationState": "FAILED"
156 configure_scaling_groups
.side_effect
= mock_configure_scaling_groups
157 configure_vnf_alarms
.side_effect
= mock_configure_vnf_alarms
158 delete_scaling_groups
.side_effect
= mock_delete_scaling_groups
159 delete_vnf_alarms
.side_effect
= mock_delete_vnf_alarms
161 self
.loop
.run_until_complete(agent
._handle
_policy
_update
(content
))
162 configure_scaling_groups
.assert_called_with("test_nsr_id", "1")
163 configure_scaling_groups
.reset_mock()
165 self
.loop
.run_until_complete(agent
._handle
_policy
_update
(failed_content
))
166 configure_scaling_groups
.assert_not_called()
168 @mock.patch
.object(CommonDbClient
, "__init__", lambda *args
, **kwargs
: None)
169 @mock.patch("osm_policy_module.autoscaling.service.MonClient")
170 @mock.patch("osm_policy_module.autoscaling.service.LcmClient")
171 @mock.patch("osm_policy_module.alarming.service.MonClient")
172 @mock.patch("osm_policy_module.alarming.service.LcmClient")
173 @mock.patch("osm_policy_module.healing.service.MonClient")
174 @mock.patch("osm_policy_module.healing.service.LcmClient")
175 @mock.patch
.object(AutoscalingService
, "handle_alarm")
176 @mock.patch
.object(AlarmingService
, "handle_alarm")
177 @mock.patch
.object(HealingService
, "handle_alarm")
178 def test_handle_alarm_notification(
180 healing_handle_alarm
,
181 alarming_handle_alarm
,
182 autoscaling_handle_alarm
,
183 autoscaling_lcm_client
,
184 autoscaling_mon_client
,
190 async def mock_handle_alarm(alarm_uuid
, status
, payload
=None):
194 agent
= PolicyModuleAgent(config
, self
.loop
)
195 assert autoscaling_lcm_client
.called
196 assert autoscaling_mon_client
.called
197 assert alarming_lcm_client
.called
198 assert alarming_mon_client
.called
199 assert healing_lcm_client
.called
200 assert healing_mon_client
.called
203 "alarm_uuid": "test_alarm_uuid",
204 "metric_name": "test_metric_name",
205 "operation": "test_operation",
206 "threshold_value": "test_threshold_value",
207 "vdu_name": "test_vdu_name",
208 "vnf_member_index": "test_vnf_member_index",
209 "ns_id": "test_nsr_id",
213 autoscaling_handle_alarm
.side_effect
= mock_handle_alarm
214 alarming_handle_alarm
.side_effect
= mock_handle_alarm
215 healing_handle_alarm
.side_effect
= mock_handle_alarm
217 self
.loop
.run_until_complete(agent
._handle
_alarm
_notification
(content
))
218 autoscaling_handle_alarm
.assert_called_with("test_alarm_uuid", "alarm")
219 alarming_handle_alarm
.assert_called_with("test_alarm_uuid", "alarm", content
)
220 healing_handle_alarm
.assert_called_with("test_alarm_uuid", "alarm")
222 @mock.patch
.object(CommonDbClient
, "__init__", lambda *args
, **kwargs
: None)
223 @mock.patch("osm_policy_module.alarming.service.MonClient")
224 @mock.patch("osm_policy_module.alarming.service.LcmClient")
225 @mock.patch("osm_policy_module.autoscaling.service.MonClient")
226 @mock.patch("osm_policy_module.autoscaling.service.LcmClient")
227 @mock.patch
.object(AutoscalingService
, "delete_scaling_groups")
228 @mock.patch
.object(AlarmingService
, "delete_vnf_alarms")
229 def test_handle_vnf_terminated(
232 delete_scaling_groups
,
233 autoscaling_lcm_client
,
234 autoscaling_mon_client
,
238 async def mock_delete_scaling_groups(nsr_id
, vnf_member_index
):
241 async def mock_delete_vnf_alarms(nsr_id
, vnf_member_index
):
245 agent
= PolicyModuleAgent(config
, self
.loop
)
246 assert autoscaling_lcm_client
.called
247 assert autoscaling_mon_client
.called
248 assert alarming_lcm_client
.called
249 assert alarming_mon_client
.called
251 "nsr_id": "test_nsr_id",
252 "vnf_member_index": "1",
253 "operationState": "COMPLETED"
256 "nsr_id": "test_nsr_id",
257 "vnf_member_index": "1",
258 "operationState": "FAILED"
260 delete_scaling_groups
.side_effect
= mock_delete_scaling_groups
261 delete_vnf_alarms
.side_effect
= mock_delete_vnf_alarms
263 self
.loop
.run_until_complete(agent
._handle
_vnf
_terminated
(content
))
264 delete_scaling_groups
.assert_called_with("test_nsr_id", "1")
265 delete_scaling_groups
.reset_mock()
267 self
.loop
.run_until_complete(agent
._handle
_vnf
_terminated
(failed_content
))
268 delete_scaling_groups
.assert_not_called()
271 if __name__
== "__main__":