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