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.core.agent import PolicyModuleAgent
  31 from osm_policy_module.core.config import Config
  32
  33
  34 class PolicyAgentTest(unittest.TestCase):
  35     def setUp(self):
  36         self.loop = asyncio.new_event_loop()
  37
  38     @mock.patch('osm_policy_module.alarming.service.CommonDbClient')
  39     @mock.patch('osm_policy_module.alarming.service.MonClient')
  40     @mock.patch('osm_policy_module.alarming.service.LcmClient')
  41     @mock.patch('osm_policy_module.autoscaling.service.CommonDbClient')
  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     def test_handle_instantiated(self,
  48                                  delete_orphaned_alarms,
  49                                  configure_vnf_alarms,
  50                                  configure_scaling_groups,
  51                                  autoscaling_lcm_client,
  52                                  autoscaling_mon_client,
  53                                  autoscaling_db_client,
  54                                  alarming_lcm_client,
  55                                  alarming_mon_client,
  56                                  alarming_db_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 autoscaling_db_client.called
  71         assert alarming_lcm_client.called
  72         assert alarming_mon_client.called
  73         assert alarming_db_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 = {
  82             'operationState': 'FAILED',
  83             'nsInstanceId': 'test_nsr_id'
  84         }
  85         configure_scaling_groups.side_effect = mock_configure_scaling_groups
  86         configure_vnf_alarms.side_effect = mock_configure_vnf_alarms
  87         delete_orphaned_alarms.side_effect = mock_delete_orphaned_alarms
  88
  89         autoscaling_db_client.return_value.get_nslcmop.return_value = nslcmop_completed
  90         self.loop.run_until_complete(agent._handle_instantiated(content))
  91         configure_scaling_groups.assert_called_with('test_nsr_id')
  92         configure_scaling_groups.reset_mock()
  93
  94         autoscaling_db_client.return_value.get_nslcmop.return_value = nslcmop_failed
  95         self.loop.run_until_complete(agent._handle_instantiated(content))
  96         configure_scaling_groups.assert_not_called()
  97
  98     @mock.patch('osm_policy_module.autoscaling.service.CommonDbClient')
  99     @mock.patch('osm_policy_module.autoscaling.service.MonClient')
 100     @mock.patch('osm_policy_module.autoscaling.service.LcmClient')
 101     @mock.patch('osm_policy_module.alarming.service.CommonDbClient')
 102     @mock.patch('osm_policy_module.alarming.service.MonClient')
 103     @mock.patch('osm_policy_module.alarming.service.LcmClient')
 104     @mock.patch.object(AutoscalingService, 'handle_alarm')
 105     @mock.patch.object(AlarmingService, 'handle_alarm')
 106     def test_handle_alarm_notification(self,
 107                                        alarming_handle_alarm,
 108                                        autoscaling_handle_alarm,
 109                                        autoscaling_lcm_client,
 110                                        autoscaling_mon_client,
 111                                        autoscaling_db_client,
 112                                        alarming_lcm_client,
 113                                        alarming_mon_client,
 114                                        alarming_db_client):
 115         async def mock_handle_alarm(alarm_uuid, status, payload=None):
 116             pass
 117
 118         config = Config()
 119         agent = PolicyModuleAgent(config, self.loop)
 120         assert autoscaling_lcm_client.called
 121         assert autoscaling_mon_client.called
 122         assert autoscaling_db_client.called
 123         assert alarming_lcm_client.called
 124         assert alarming_mon_client.called
 125         assert alarming_db_client.called
 126         content = {
 127             'notify_details': {
 128                 'alarm_uuid': 'test_alarm_uuid',
 129                 'metric_name': 'test_metric_name',
 130                 'operation': 'test_operation',
 131                 'threshold_value': 'test_threshold_value',
 132                 'vdu_name': 'test_vdu_name',
 133                 'vnf_member_index': 'test_vnf_member_index',
 134                 'ns_id': 'test_nsr_id',
 135                 'status': 'alarm'
 136             }
 137         }
 138         autoscaling_handle_alarm.side_effect = mock_handle_alarm
 139         alarming_handle_alarm.side_effect = mock_handle_alarm
 140
 141         self.loop.run_until_complete(agent._handle_alarm_notification(content))
 142         autoscaling_handle_alarm.assert_called_with('test_alarm_uuid', 'alarm')
 143         alarming_handle_alarm.assert_called_with('test_alarm_uuid', 'alarm', content)
 144
 145
 146 if __name__ == '__main__':
 147     unittest.main()