osm_policy_module/tests/unit/core/test_policy_agent.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 unittest
  26 from unittest import mock
  27
  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
  34
  35 class PolicyAgentTest(unittest.TestCase):
  36     def setUp(self):
  37         self.loop = asyncio.new_event_loop()
  38
  39     @mock.patch.object(CommonDbClient, "__init__", lambda *args, **kwargs: None)
  40     @mock.patch("osm_policy_module.alarming.service.MonClient")
  41     @mock.patch("osm_policy_module.alarming.service.LcmClient")
  42     @mock.patch("osm_policy_module.autoscaling.service.MonClient")
  43     @mock.patch("osm_policy_module.autoscaling.service.LcmClient")
  44     @mock.patch.object(AutoscalingService, "configure_scaling_groups")
  45     @mock.patch.object(AlarmingService, "configure_vnf_alarms")
  46     @mock.patch.object(AutoscalingService, "delete_orphaned_alarms")
  47     @mock.patch.object(CommonDbClient, "get_nslcmop")
  48     def test_handle_instantiated(
  49         self,
  50         get_nslcmop,
  51         delete_orphaned_alarms,
  52         configure_vnf_alarms,
  53         configure_scaling_groups,
  54         autoscaling_lcm_client,
  55         autoscaling_mon_client,
  56         alarming_lcm_client,
  57         alarming_mon_client,
  58     ):
  59         async def mock_configure_scaling_groups(nsr_id):
  60             pass
  61
  62         async def mock_configure_vnf_alarms(nsr_id):
  63             pass
  64
  65         async def mock_delete_orphaned_alarms(nsr_id):
  66             pass
  67
  68         config = Config()
  69         agent = PolicyModuleAgent(config, self.loop)
  70         assert autoscaling_lcm_client.called
  71         assert autoscaling_mon_client.called
  72         assert alarming_lcm_client.called
  73         assert alarming_mon_client.called
  74         content = {
  75             "nslcmop_id": "test_id",
  76         }
  77         nslcmop_completed = {
  78             "operationState": "COMPLETED",
  79             "nsInstanceId": "test_nsr_id",
  80         }
  81         nslcmop_failed = {"operationState": "FAILED", "nsInstanceId": "test_nsr_id"}
  82         configure_scaling_groups.side_effect = mock_configure_scaling_groups
  83         configure_vnf_alarms.side_effect = mock_configure_vnf_alarms
  84         delete_orphaned_alarms.side_effect = mock_delete_orphaned_alarms
  85
  86         get_nslcmop.return_value = nslcmop_completed
  87         self.loop.run_until_complete(agent._handle_instantiated(content))
  88         configure_scaling_groups.assert_called_with("test_nsr_id")
  89         configure_scaling_groups.reset_mock()
  90
  91         get_nslcmop.return_value = nslcmop_failed
  92         self.loop.run_until_complete(agent._handle_instantiated(content))
  93         configure_scaling_groups.assert_not_called()
  94
  95     @mock.patch.object(CommonDbClient, "__init__", lambda *args, **kwargs: None)
  96     @mock.patch("osm_policy_module.autoscaling.service.MonClient")
  97     @mock.patch("osm_policy_module.autoscaling.service.LcmClient")
  98     @mock.patch("osm_policy_module.alarming.service.MonClient")
  99     @mock.patch("osm_policy_module.alarming.service.LcmClient")
 100     @mock.patch.object(AutoscalingService, "handle_alarm")
 101     @mock.patch.object(AlarmingService, "handle_alarm")
 102     def test_handle_alarm_notification(
 103         self,
 104         alarming_handle_alarm,
 105         autoscaling_handle_alarm,
 106         autoscaling_lcm_client,
 107         autoscaling_mon_client,
 108         alarming_lcm_client,
 109         alarming_mon_client,
 110     ):
 111         async def mock_handle_alarm(alarm_uuid, status, payload=None):
 112             pass
 113
 114         config = Config()
 115         agent = PolicyModuleAgent(config, self.loop)
 116         assert autoscaling_lcm_client.called
 117         assert autoscaling_mon_client.called
 118         assert alarming_lcm_client.called
 119         assert alarming_mon_client.called
 120         content = {
 121             "notify_details": {
 122                 "alarm_uuid": "test_alarm_uuid",
 123                 "metric_name": "test_metric_name",
 124                 "operation": "test_operation",
 125                 "threshold_value": "test_threshold_value",
 126                 "vdu_name": "test_vdu_name",
 127                 "vnf_member_index": "test_vnf_member_index",
 128                 "ns_id": "test_nsr_id",
 129                 "status": "alarm",
 130             }
 131         }
 132         autoscaling_handle_alarm.side_effect = mock_handle_alarm
 133         alarming_handle_alarm.side_effect = mock_handle_alarm
 134
 135         self.loop.run_until_complete(agent._handle_alarm_notification(content))
 136         autoscaling_handle_alarm.assert_called_with("test_alarm_uuid", "alarm")
 137         alarming_handle_alarm.assert_called_with("test_alarm_uuid", "alarm", content)
 138
 139     @mock.patch.object(CommonDbClient, "__init__", lambda *args, **kwargs: None)
 140     @mock.patch("osm_policy_module.alarming.service.MonClient")
 141     @mock.patch("osm_policy_module.alarming.service.LcmClient")
 142     @mock.patch("osm_policy_module.autoscaling.service.MonClient")
 143     @mock.patch("osm_policy_module.autoscaling.service.LcmClient")
 144     @mock.patch.object(AutoscalingService, "delete_scaling_groups")
 145     @mock.patch.object(AlarmingService, "delete_vnf_alarms")
 146     def test_handle_vnf_terminated(
 147         self,
 148         delete_vnf_alarms,
 149         delete_scaling_groups,
 150         autoscaling_lcm_client,
 151         autoscaling_mon_client,
 152         alarming_lcm_client,
 153         alarming_mon_client,
 154     ):
 155         async def mock_delete_scaling_groups(nsr_id, vnf_member_index):
 156             pass
 157
 158         async def mock_delete_vnf_alarms(nsr_id, vnf_member_index):
 159             pass
 160
 161         config = Config()
 162         agent = PolicyModuleAgent(config, self.loop)
 163         assert autoscaling_lcm_client.called
 164         assert autoscaling_mon_client.called
 165         assert alarming_lcm_client.called
 166         assert alarming_mon_client.called
 167         content = {
 168             "nsr_id": "test_nsr_id",
 169             "vnf_member_index": "1",
 170             "operationState": "COMPLETED"
 171         }
 172         failed_content = {
 173             "nsr_id": "test_nsr_id",
 174             "vnf_member_index": "1",
 175             "operationState": "FAILED"
 176         }
 177         delete_scaling_groups.side_effect = mock_delete_scaling_groups
 178         delete_vnf_alarms.side_effect = mock_delete_vnf_alarms
 179
 180         self.loop.run_until_complete(agent._handle_vnf_terminated(content))
 181         delete_scaling_groups.assert_called_with("test_nsr_id", "1")
 182         delete_scaling_groups.reset_mock()
 183
 184         self.loop.run_until_complete(agent._handle_vnf_terminated(failed_content))
 185         delete_scaling_groups.assert_not_called()
 186
 187
 188 if __name__ == "__main__":
 189     unittest.main()