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 TestCase
, mock
28 from osm_policy_module
.autoscaling
.service
import HealingService
29 from osm_policy_module
.common
.common_db_client
import CommonDbClient
30 from osm_policy_module
.common
.lcm_client
import LcmClient
31 from osm_policy_module
.common
.mon_client
import MonClient
32 from osm_policy_module
.core
.config
import Config
33 from osm_policy_module
.core
.database
import HealingActionRepository
36 @mock.patch
.object(LcmClient
, "__init__", lambda *args
, **kwargs
: None)
37 @mock.patch
.object(MonClient
, "__init__", lambda *args
, **kwargs
: None)
38 @mock.patch
.object(CommonDbClient
, "__init__", lambda *args
, **kwargs
: None)
39 class TestHealscalingService(TestCase
):
41 self
.config
= Config()
42 self
.loop
= asyncio
.new_event_loop()
44 @mock.patch
.object(HealingActionRepository
, "get")
45 @mock.patch("osm_policy_module.core.database.db")
46 def test_update_alarm_status(self
, database
, get_alarm
):
47 mock_alarm
= mock
.Mock()
48 mock_alarm
.last_status
= "insufficient_data"
49 get_alarm
.return_value
= mock_alarm
51 service
= HealingService(self
.config
)
52 self
.loop
.run_until_complete(service
.update_alarm_status("test_uuid", "alarm"))
53 self
.assertEqual(mock_alarm
.last_status
, "alarm")
54 mock_alarm
.save
.assert_called_with()
56 service
= HealingService(self
.config
)
57 self
.loop
.run_until_complete(service
.update_alarm_status("test_uuid", "ok"))
58 self
.assertEqual(mock_alarm
.last_status
, "ok")
59 mock_alarm
.save
.assert_called_with()
61 service
= HealingService(self
.config
)
62 self
.loop
.run_until_complete(
63 service
.update_alarm_status("test_uuid", "insufficient_data")
65 self
.assertEqual(mock_alarm
.last_status
, "insufficient_data")
66 mock_alarm
.save
.assert_called_with()
68 @mock.patch
.object(HealingActionRepository
, "list")
69 @mock.patch
.object(HealingActionRepository
, "get")
70 @mock.patch(LcmClient
, "heal")
71 @mock.patch("osm_policy_module.core.database.db")
72 def test_handle_alarm(self
, database
, heal
, get_alarm
, list_alarms
):
73 mock_alarm
= self
._build
_mock
_alarm
("test_id", status
="alarm")
74 get_alarm
.return_value
= mock_alarm
75 service
= HealingService(self
.config
)
76 self
.loop
.run_until_complete(service
.handle_alarm("test_id", "alarm"))
77 heal
.assert_called_with(
79 "test_vnfinstance_id",
82 "test_vnf_member_index",
88 def _build_mock_alarm(
91 last_heal
=datetime
.datetime
.min,
94 mock_alarm
= mock
.Mock()
95 mock_alarm
.last_status
= last_status
96 mock_alarm
.vnf_member_index
= "1"
97 mock_alarm
.last_heal
= last_heal
98 mock_alarm
.cooldown_time
= cooldown_time
99 mock_alarm
.nsr_id
= "test_nsr_id"
100 mock_alarm
.vnfinstance_id
= "test_vnfinstance_id"
101 mock_alarm
.vdur_name
= "test_vdur_name"
102 mock_alarm
.vdu_id
= "test_vdu_id"
103 mock_alarm
.vnf_member_index
= "test_vnf_member_index"
104 mock_alarm
.heal_type
= "test_heal_type"
105 mock_alarm
.day1
= "test_day1"
106 mock_alarm
.count_index
= "test_count_index"