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 datetime
  26 import unittest
  27 from unittest import mock
  28 from unittest.mock import Mock
  29
  30 from osm_policy_module.core.agent import PolicyModuleAgent
  31 from osm_policy_module.core.config import Config
  32 from osm_policy_module.core.database import DatabaseManager
  33
  34
  35 class PolicyAgentTest(unittest.TestCase):
  36     def setUp(self):
  37         self.loop = asyncio.new_event_loop()
  38
  39     @mock.patch('osm_policy_module.core.agent.CommonDbClient')
  40     @mock.patch('osm_policy_module.core.agent.MonClient')
  41     @mock.patch('osm_policy_module.core.agent.LcmClient')
  42     @mock.patch.object(PolicyModuleAgent, '_configure_scaling_groups')
  43     @mock.patch.object(PolicyModuleAgent, '_delete_orphaned_alarms')
  44     def test_handle_instantiated(self, delete_orphaned_alarms, configure_scaling_groups, lcm_client,
  45                                  mon_client, db_client):
  46         async def mock_configure_scaling_groups(nsr_id):
  47             pass
  48
  49         async def mock_delete_orphaned_alarms(nsr_id):
  50             pass
  51
  52         config = Config()
  53         agent = PolicyModuleAgent(config, self.loop)
  54         assert lcm_client.called
  55         assert mon_client.called
  56         assert db_client.called
  57         content = {
  58             'nslcmop_id': 'test_id',
  59         }
  60         nslcmop_completed = {
  61             'operationState': 'COMPLETED',
  62             'nsInstanceId': 'test_nsr_id'
  63         }
  64         nslcmop_failed = {
  65             'operationState': 'FAILED',
  66             'nsInstanceId': 'test_nsr_id'
  67         }
  68         configure_scaling_groups.side_effect = mock_configure_scaling_groups
  69         delete_orphaned_alarms.side_effect = mock_delete_orphaned_alarms
  70
  71         db_client.return_value.get_nslcmop.return_value = nslcmop_completed
  72         self.loop.run_until_complete(agent._handle_instantiated(content))
  73         configure_scaling_groups.assert_called_with('test_nsr_id')
  74         configure_scaling_groups.reset_mock()
  75
  76         db_client.return_value.get_nslcmop.return_value = nslcmop_failed
  77         self.loop.run_until_complete(agent._handle_instantiated(content))
  78         configure_scaling_groups.assert_not_called()
  79
  80     @mock.patch('osm_policy_module.core.agent.CommonDbClient')
  81     @mock.patch('osm_policy_module.core.agent.MonClient')
  82     @mock.patch('osm_policy_module.core.agent.LcmClient')
  83     @mock.patch.object(DatabaseManager, 'get_alarm')
  84     def test_handle_alarm_notification(self, get_alarm, lcm_client, mon_client, db_client):
  85         async def mock_scale(nsr_id, scaling_group_name, vnf_member_index, action):
  86             pass
  87
  88         config = Config()
  89         agent = PolicyModuleAgent(config, self.loop)
  90         assert lcm_client.called
  91         assert mon_client.called
  92         assert db_client.called
  93         content = {
  94             'notify_details': {
  95                 'alarm_uuid': 'test_alarm_uuid',
  96                 'metric_name': 'test_metric_name',
  97                 'operation': 'test_operation',
  98                 'threshold_value': 'test_threshold_value',
  99                 'vdu_name': 'test_vdu_name',
 100                 'vnf_member_index': 'test_vnf_member_index',
 101                 'ns_id': 'test_nsr_id'
 102             }
 103         }
 104         mock_alarm = Mock()
 105         mock_alarm.vnf_member_index = 1
 106         mock_alarm.action = 'scale_out'
 107         mock_scaling_criteria = Mock()
 108         mock_scaling_policy = Mock()
 109         mock_scaling_group = Mock()
 110         mock_scaling_group.nsr_id = 'test_nsr_id'
 111         mock_scaling_group.name = 'test_name'
 112         mock_scaling_policy.cooldown_time = 60
 113         mock_scaling_policy.scaling_group = mock_scaling_group
 114         mock_scaling_criteria.scaling_policy = mock_scaling_policy
 115         mock_alarm.scaling_criteria = mock_scaling_criteria
 116         get_alarm.return_value = mock_alarm
 117         lcm_client.return_value.scale.side_effect = mock_scale
 118
 119         mock_scaling_policy.last_scale = datetime.datetime.now() - datetime.timedelta(minutes=90)
 120
 121         self.loop.run_until_complete(agent._handle_alarm_notification(content))
 122         lcm_client.return_value.scale.assert_called_with('test_nsr_id', 'test_name', 1, 'scale_out')
 123         lcm_client.return_value.scale.reset_mock()
 124
 125         mock_scaling_policy.last_scale = datetime.datetime.now()
 126         self.loop.run_until_complete(agent._handle_alarm_notification(content))
 127         lcm_client.return_value.scale.assert_not_called()
 128
 129
 130 if __name__ == '__main__':
 131     unittest.main()