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.alarming.service.MonClient")
  97     @mock.patch("osm_policy_module.alarming.service.LcmClient")
  98     @mock.patch("osm_policy_module.autoscaling.service.MonClient")
  99     @mock.patch("osm_policy_module.autoscaling.service.LcmClient")
 100     @mock.patch.object(AutoscalingService, "configure_scaling_groups")
 101     @mock.patch.object(AlarmingService, "configure_vnf_alarms")
 102     @mock.patch.object(AutoscalingService, "delete_scaling_groups")
 103     @mock.patch.object(AlarmingService, "delete_vnf_alarms")
 104     def test_handle_policy_update(
 105         self,
 106         delete_vnf_alarms,
 107         delete_scaling_groups,
 108         configure_vnf_alarms,
 109         configure_scaling_groups,
 110         autoscaling_lcm_client,
 111         autoscaling_mon_client,
 112         alarming_lcm_client,
 113         alarming_mon_client,
 114     ):
 115         async def mock_delete_scaling_groups(nsr_id, vnf_member_index):
 116             pass
 117
 118         async def mock_delete_vnf_alarms(nsr_id, vnf_member_index):
 119             pass
 120
 121         async def mock_configure_scaling_groups(nsr_id, vnf_member_index):
 122             pass
 123
 124         async def mock_configure_vnf_alarms(nsr_id, vnf_member_index):
 125             pass
 126
 127         config = Config()
 128         agent = PolicyModuleAgent(config, self.loop)
 129         assert autoscaling_lcm_client.called
 130         assert autoscaling_mon_client.called
 131         assert alarming_lcm_client.called
 132         assert alarming_mon_client.called
 133         content = {
 134             "nsr_id": "test_nsr_id",
 135             "vnf_member_index": "1",
 136             "operationState": "COMPLETED"
 137         }
 138         failed_content = {
 139             "nsr_id": "test_nsr_id",
 140             "vnf_member_index": "1",
 141             "operationState": "FAILED"
 142         }
 143         configure_scaling_groups.side_effect = mock_configure_scaling_groups
 144         configure_vnf_alarms.side_effect = mock_configure_vnf_alarms
 145         delete_scaling_groups.side_effect = mock_delete_scaling_groups
 146         delete_vnf_alarms.side_effect = mock_delete_vnf_alarms
 147
 148         self.loop.run_until_complete(agent._handle_policy_update(content))
 149         configure_scaling_groups.assert_called_with("test_nsr_id", "1")
 150         configure_scaling_groups.reset_mock()
 151
 152         self.loop.run_until_complete(agent._handle_policy_update(failed_content))
 153         configure_scaling_groups.assert_not_called()
 154
 155     @mock.patch.object(CommonDbClient, "__init__", lambda *args, **kwargs: None)
 156     @mock.patch("osm_policy_module.autoscaling.service.MonClient")
 157     @mock.patch("osm_policy_module.autoscaling.service.LcmClient")
 158     @mock.patch("osm_policy_module.alarming.service.MonClient")
 159     @mock.patch("osm_policy_module.alarming.service.LcmClient")
 160     @mock.patch.object(AutoscalingService, "handle_alarm")
 161     @mock.patch.object(AlarmingService, "handle_alarm")
 162     def test_handle_alarm_notification(
 163         self,
 164         alarming_handle_alarm,
 165         autoscaling_handle_alarm,
 166         autoscaling_lcm_client,
 167         autoscaling_mon_client,
 168         alarming_lcm_client,
 169         alarming_mon_client,
 170     ):
 171         async def mock_handle_alarm(alarm_uuid, status, payload=None):
 172             pass
 173
 174         config = Config()
 175         agent = PolicyModuleAgent(config, self.loop)
 176         assert autoscaling_lcm_client.called
 177         assert autoscaling_mon_client.called
 178         assert alarming_lcm_client.called
 179         assert alarming_mon_client.called
 180         content = {
 181             "notify_details": {
 182                 "alarm_uuid": "test_alarm_uuid",
 183                 "metric_name": "test_metric_name",
 184                 "operation": "test_operation",
 185                 "threshold_value": "test_threshold_value",
 186                 "vdu_name": "test_vdu_name",
 187                 "vnf_member_index": "test_vnf_member_index",
 188                 "ns_id": "test_nsr_id",
 189                 "status": "alarm",
 190             }
 191         }
 192         autoscaling_handle_alarm.side_effect = mock_handle_alarm
 193         alarming_handle_alarm.side_effect = mock_handle_alarm
 194
 195         self.loop.run_until_complete(agent._handle_alarm_notification(content))
 196         autoscaling_handle_alarm.assert_called_with("test_alarm_uuid", "alarm")
 197         alarming_handle_alarm.assert_called_with("test_alarm_uuid", "alarm", content)
 198
 199     @mock.patch.object(CommonDbClient, "__init__", lambda *args, **kwargs: None)
 200     @mock.patch("osm_policy_module.alarming.service.MonClient")
 201     @mock.patch("osm_policy_module.alarming.service.LcmClient")
 202     @mock.patch("osm_policy_module.autoscaling.service.MonClient")
 203     @mock.patch("osm_policy_module.autoscaling.service.LcmClient")
 204     @mock.patch.object(AutoscalingService, "delete_scaling_groups")
 205     @mock.patch.object(AlarmingService, "delete_vnf_alarms")
 206     def test_handle_vnf_terminated(
 207         self,
 208         delete_vnf_alarms,
 209         delete_scaling_groups,
 210         autoscaling_lcm_client,
 211         autoscaling_mon_client,
 212         alarming_lcm_client,
 213         alarming_mon_client,
 214     ):
 215         async def mock_delete_scaling_groups(nsr_id, vnf_member_index):
 216             pass
 217
 218         async def mock_delete_vnf_alarms(nsr_id, vnf_member_index):
 219             pass
 220
 221         config = Config()
 222         agent = PolicyModuleAgent(config, self.loop)
 223         assert autoscaling_lcm_client.called
 224         assert autoscaling_mon_client.called
 225         assert alarming_lcm_client.called
 226         assert alarming_mon_client.called
 227         content = {
 228             "nsr_id": "test_nsr_id",
 229             "vnf_member_index": "1",
 230             "operationState": "COMPLETED"
 231         }
 232         failed_content = {
 233             "nsr_id": "test_nsr_id",
 234             "vnf_member_index": "1",
 235             "operationState": "FAILED"
 236         }
 237         delete_scaling_groups.side_effect = mock_delete_scaling_groups
 238         delete_vnf_alarms.side_effect = mock_delete_vnf_alarms
 239
 240         self.loop.run_until_complete(agent._handle_vnf_terminated(content))
 241         delete_scaling_groups.assert_called_with("test_nsr_id", "1")
 242         delete_scaling_groups.reset_mock()
 243
 244         self.loop.run_until_complete(agent._handle_vnf_terminated(failed_content))
 245         delete_scaling_groups.assert_not_called()
 246
 247
 248 if __name__ == "__main__":
 249     unittest.main()