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(self,
  49                                  get_nslcmop,
  50                                  delete_orphaned_alarms,
  51                                  configure_vnf_alarms,
  52                                  configure_scaling_groups,
  53                                  autoscaling_lcm_client,
  54                                  autoscaling_mon_client,
  55                                  alarming_lcm_client,
  56                                  alarming_mon_client):
  57         async def mock_configure_scaling_groups(nsr_id):
  58             pass
  59
  60         async def mock_configure_vnf_alarms(nsr_id):
  61             pass
  62
  63         async def mock_delete_orphaned_alarms(nsr_id):
  64             pass
  65
  66         config = Config()
  67         agent = PolicyModuleAgent(config, self.loop)
  68         assert autoscaling_lcm_client.called
  69         assert autoscaling_mon_client.called
  70         assert alarming_lcm_client.called
  71         assert alarming_mon_client.called
  72         content = {
  73             'nslcmop_id': 'test_id',
  74         }
  75         nslcmop_completed = {
  76             'operationState': 'COMPLETED',
  77             'nsInstanceId': 'test_nsr_id'
  78         }
  79         nslcmop_failed = {
  80             'operationState': 'FAILED',
  81             'nsInstanceId': 'test_nsr_id'
  82         }
  83         configure_scaling_groups.side_effect = mock_configure_scaling_groups
  84         configure_vnf_alarms.side_effect = mock_configure_vnf_alarms
  85         delete_orphaned_alarms.side_effect = mock_delete_orphaned_alarms
  86
  87         get_nslcmop.return_value = nslcmop_completed
  88         self.loop.run_until_complete(agent._handle_instantiated(content))
  89         configure_scaling_groups.assert_called_with('test_nsr_id')
  90         configure_scaling_groups.reset_mock()
  91
  92         get_nslcmop.return_value = nslcmop_failed
  93         self.loop.run_until_complete(agent._handle_instantiated(content))
  94         configure_scaling_groups.assert_not_called()
  95
  96     @mock.patch.object(CommonDbClient, "__init__", lambda *args, **kwargs: None)
  97     @mock.patch('osm_policy_module.autoscaling.service.MonClient')
  98     @mock.patch('osm_policy_module.autoscaling.service.LcmClient')
  99     @mock.patch('osm_policy_module.alarming.service.MonClient')
 100     @mock.patch('osm_policy_module.alarming.service.LcmClient')
 101     @mock.patch.object(AutoscalingService, 'handle_alarm')
 102     @mock.patch.object(AlarmingService, 'handle_alarm')
 103     def test_handle_alarm_notification(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         async def mock_handle_alarm(alarm_uuid, status, payload=None):
 111             pass
 112
 113         config = Config()
 114         agent = PolicyModuleAgent(config, self.loop)
 115         assert autoscaling_lcm_client.called
 116         assert autoscaling_mon_client.called
 117         assert alarming_lcm_client.called
 118         assert alarming_mon_client.called
 119         content = {
 120             'notify_details': {
 121                 'alarm_uuid': 'test_alarm_uuid',
 122                 'metric_name': 'test_metric_name',
 123                 'operation': 'test_operation',
 124                 'threshold_value': 'test_threshold_value',
 125                 'vdu_name': 'test_vdu_name',
 126                 'vnf_member_index': 'test_vnf_member_index',
 127                 'ns_id': 'test_nsr_id',
 128                 'status': 'alarm'
 129             }
 130         }
 131         autoscaling_handle_alarm.side_effect = mock_handle_alarm
 132         alarming_handle_alarm.side_effect = mock_handle_alarm
 133
 134         self.loop.run_until_complete(agent._handle_alarm_notification(content))
 135         autoscaling_handle_alarm.assert_called_with('test_alarm_uuid', 'alarm')
 136         alarming_handle_alarm.assert_called_with('test_alarm_uuid', 'alarm', content)
 137
 138
 139 if __name__ == '__main__':
 140     unittest.main()