osm_policy_module/tests/unit/healing/test_healing_service.py

   1 # -*- coding: utf-8 -*-
   2
   3 # Copyright 2018 Whitestack, LLC
   4 # *************************************************************
   5
   6 # This file is part of OSM Monitoring module
   7 # All Rights Reserved to Whitestack, LLC
   8
   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
  12
  13 #         http://www.apache.org/licenses/LICENSE-2.0
  14
  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
  19 # under the License.
  20
  21 # For those usages not covered by the Apache License, Version 2.0 please
  22 # contact: bdiaz@whitestack.com or glavado@whitestack.com
  23 ##
  24 import asyncio
  25 import datetime
  26 from unittest import TestCase, mock
  27
  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
  34
  35
  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):
  40     def setUp(self):
  41         self.config = Config()
  42         self.loop = asyncio.new_event_loop()
  43
  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
  50
  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()
  55
  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()
  60
  61         service = HealingService(self.config)
  62         self.loop.run_until_complete(
  63             service.update_alarm_status("test_uuid", "insufficient_data")
  64         )
  65         self.assertEqual(mock_alarm.last_status, "insufficient_data")
  66         mock_alarm.save.assert_called_with()
  67
  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(
  78             "test_nsr_id",
  79             "test_vnfinstance_id",
  80             "test_vdur_name",
  81             "test_vdu_id",
  82             "test_vnf_member_index",
  83             "test_heal_type",
  84             "test_day1",
  85             "test_count_index",
  86         )
  87
  88     def _build_mock_alarm(
  89         self,
  90         last_status="alarm",
  91         last_heal=datetime.datetime.min,
  92         cooldown_time=10,
  93     ):
  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"
 107         return mock_alarm