# contact: bdiaz@whitestack.com or glavado@whitestack.com
##
import asyncio
-import datetime
import unittest
from unittest import mock
-from unittest.mock import Mock
+from osm_policy_module.alarming.service import AlarmingService
+from osm_policy_module.autoscaling.service import AutoscalingService
+from osm_policy_module.common.common_db_client import CommonDbClient
from osm_policy_module.core.agent import PolicyModuleAgent
-from osm_policy_module.core.database import DatabaseManager
+from osm_policy_module.core.config import Config
+from osm_policy_module.healing.service import HealingService
class PolicyAgentTest(unittest.TestCase):
def setUp(self):
self.loop = asyncio.new_event_loop()
- asyncio.set_event_loop(None)
- @mock.patch('osm_policy_module.core.agent.CommonDbClient')
- @mock.patch('osm_policy_module.core.agent.MonClient')
- @mock.patch('osm_policy_module.core.agent.LcmClient')
- @mock.patch.object(PolicyModuleAgent, '_configure_scaling_groups')
- def test_handle_instantiated_or_scaled(self, configure_scaling_groups, lcm_client, mon_client, db_client):
+ @mock.patch.object(CommonDbClient, "__init__", lambda *args, **kwargs: None)
+ @mock.patch("osm_policy_module.alarming.service.MonClient")
+ @mock.patch("osm_policy_module.alarming.service.LcmClient")
+ @mock.patch("osm_policy_module.autoscaling.service.MonClient")
+ @mock.patch("osm_policy_module.autoscaling.service.LcmClient")
+ @mock.patch("osm_policy_module.healing.service.MonClient")
+ @mock.patch("osm_policy_module.healing.service.LcmClient")
+ @mock.patch.object(AutoscalingService, "configure_scaling_groups")
+ @mock.patch.object(AlarmingService, "configure_vnf_alarms")
+ @mock.patch.object(HealingService, "configure_healing_alarms")
+ @mock.patch.object(AutoscalingService, "delete_orphaned_alarms")
+ @mock.patch.object(CommonDbClient, "get_nslcmop")
+ def test_handle_instantiated(
+ self,
+ get_nslcmop,
+ delete_orphaned_alarms,
+ configure_healing_alarms,
+ configure_vnf_alarms,
+ configure_scaling_groups,
+ autoscaling_lcm_client,
+ autoscaling_mon_client,
+ alarming_lcm_client,
+ alarming_mon_client,
+ healing_lcm_client,
+ healing_mon_client,
+ ):
async def mock_configure_scaling_groups(nsr_id):
pass
- agent = PolicyModuleAgent(self.loop)
- assert lcm_client.called
- assert mon_client.called
- assert db_client.called
+ async def mock_configure_vnf_alarms(nsr_id):
+ pass
+
+ async def mock_configure_healing_alarms(nsr_id):
+ pass
+
+ async def mock_delete_orphaned_alarms(nsr_id):
+ pass
+
+ config = Config()
+ agent = PolicyModuleAgent(config)
+ assert autoscaling_lcm_client.called
+ assert autoscaling_mon_client.called
+ assert alarming_lcm_client.called
+ assert alarming_mon_client.called
+ assert healing_lcm_client.called
+ assert healing_mon_client.called
content = {
- 'nslcmop_id': 'test_id',
+ "nslcmop_id": "test_id",
}
nslcmop_completed = {
- 'operationState': 'COMPLETED',
- 'nsInstanceId': 'test_nsr_id'
+ "operationState": "COMPLETED",
+ "nsInstanceId": "test_nsr_id",
+ }
+ nslcmop_failed = {"operationState": "FAILED", "nsInstanceId": "test_nsr_id"}
+ configure_scaling_groups.side_effect = mock_configure_scaling_groups
+ configure_vnf_alarms.side_effect = mock_configure_vnf_alarms
+ configure_healing_alarms.side_effect = mock_configure_healing_alarms
+ delete_orphaned_alarms.side_effect = mock_delete_orphaned_alarms
+
+ get_nslcmop.return_value = nslcmop_completed
+ self.loop.run_until_complete(agent._handle_instantiated(content))
+ configure_scaling_groups.assert_called_with("test_nsr_id")
+ configure_scaling_groups.reset_mock()
+
+ get_nslcmop.return_value = nslcmop_failed
+ self.loop.run_until_complete(agent._handle_instantiated(content))
+ configure_scaling_groups.assert_not_called()
+
+ @mock.patch.object(CommonDbClient, "__init__", lambda *args, **kwargs: None)
+ @mock.patch("osm_policy_module.alarming.service.MonClient")
+ @mock.patch("osm_policy_module.alarming.service.LcmClient")
+ @mock.patch("osm_policy_module.autoscaling.service.MonClient")
+ @mock.patch("osm_policy_module.autoscaling.service.LcmClient")
+ @mock.patch("osm_policy_module.healing.service.MonClient")
+ @mock.patch("osm_policy_module.healing.service.LcmClient")
+ @mock.patch.object(HealingService, "configure_healing_alarms")
+ @mock.patch.object(AutoscalingService, "configure_scaling_groups")
+ @mock.patch.object(AlarmingService, "configure_vnf_alarms")
+ @mock.patch.object(HealingService, "delete_healing_alarms")
+ @mock.patch.object(AutoscalingService, "delete_scaling_groups")
+ @mock.patch.object(AlarmingService, "delete_vnf_alarms")
+ def test_handle_policy_update(
+ self,
+ delete_vnf_alarms,
+ delete_scaling_groups,
+ delete_healing_alarms,
+ configure_vnf_alarms,
+ configure_scaling_groups,
+ configure_healing_alarms,
+ healing_lcm_client,
+ healing_mon_client,
+ autoscaling_lcm_client,
+ autoscaling_mon_client,
+ alarming_lcm_client,
+ alarming_mon_client,
+ ):
+ async def mock_delete_scaling_groups(nsr_id, vnf_member_index):
+ pass
+
+ async def mock_delete_vnf_alarms(nsr_id, vnf_member_index):
+ pass
+
+ async def mock_delete_healing_alarms(nsr_id, vnf_member_index):
+ pass
+
+ async def mock_configure_scaling_groups(nsr_id, vnf_member_index):
+ pass
+
+ async def mock_configure_vnf_alarms(nsr_id, vnf_member_index):
+ pass
+
+ async def mock_configure_healing_alarms(nsr_id, vnf_member_index):
+ pass
+
+ config = Config()
+ agent = PolicyModuleAgent(config)
+ assert autoscaling_lcm_client.called
+ assert autoscaling_mon_client.called
+ assert alarming_lcm_client.called
+ assert alarming_mon_client.called
+ assert healing_lcm_client.called
+ assert healing_mon_client.called
+ content = {
+ "nsr_id": "test_nsr_id",
+ "vnf_member_index": "1",
+ "operationState": "COMPLETED",
}
- nslcmop_failed = {
- 'operationState': 'FAILED',
- 'nsInstanceId': 'test_nsr_id'
+ failed_content = {
+ "nsr_id": "test_nsr_id",
+ "vnf_member_index": "1",
+ "operationState": "FAILED",
}
configure_scaling_groups.side_effect = mock_configure_scaling_groups
+ configure_vnf_alarms.side_effect = mock_configure_vnf_alarms
+ configure_healing_alarms.side_effect = mock_configure_healing_alarms
+ delete_scaling_groups.side_effect = mock_delete_scaling_groups
+ delete_vnf_alarms.side_effect = mock_delete_vnf_alarms
+ delete_healing_alarms.side_effect = mock_delete_healing_alarms
- db_client.return_value.get_nslcmop.return_value = nslcmop_completed
- self.loop.run_until_complete(agent._handle_instantiated_or_scaled(content))
- configure_scaling_groups.assert_called_with('test_nsr_id')
+ self.loop.run_until_complete(agent._handle_policy_update(content))
+ configure_scaling_groups.assert_called_with("test_nsr_id", "1")
configure_scaling_groups.reset_mock()
- db_client.return_value.get_nslcmop.return_value = nslcmop_failed
- self.loop.run_until_complete(agent._handle_instantiated_or_scaled(content))
+ self.loop.run_until_complete(agent._handle_policy_update(failed_content))
configure_scaling_groups.assert_not_called()
- @mock.patch('osm_policy_module.core.agent.CommonDbClient')
- @mock.patch('osm_policy_module.core.agent.MonClient')
- @mock.patch('osm_policy_module.core.agent.LcmClient')
- @mock.patch.object(DatabaseManager, 'get_alarm')
- def test_handle_alarm_notification(self, get_alarm, lcm_client, mon_client, db_client):
- async def mock_scale(nsr_id, scaling_group_name, vnf_member_index, action):
+ @mock.patch.object(CommonDbClient, "__init__", lambda *args, **kwargs: None)
+ @mock.patch("osm_policy_module.autoscaling.service.MonClient")
+ @mock.patch("osm_policy_module.autoscaling.service.LcmClient")
+ @mock.patch("osm_policy_module.alarming.service.MonClient")
+ @mock.patch("osm_policy_module.alarming.service.LcmClient")
+ @mock.patch("osm_policy_module.healing.service.MonClient")
+ @mock.patch("osm_policy_module.healing.service.LcmClient")
+ @mock.patch.object(AutoscalingService, "handle_alarm")
+ @mock.patch.object(AlarmingService, "handle_alarm")
+ @mock.patch.object(HealingService, "handle_alarm")
+ def test_handle_alarm_notification(
+ self,
+ healing_handle_alarm,
+ alarming_handle_alarm,
+ autoscaling_handle_alarm,
+ autoscaling_lcm_client,
+ autoscaling_mon_client,
+ alarming_lcm_client,
+ alarming_mon_client,
+ healing_lcm_client,
+ healing_mon_client,
+ ):
+ async def mock_handle_alarm(alarm_uuid, status, payload=None):
pass
- agent = PolicyModuleAgent(self.loop)
- assert lcm_client.called
- assert mon_client.called
- assert db_client.called
+ config = Config()
+ agent = PolicyModuleAgent(config)
+ assert autoscaling_lcm_client.called
+ assert autoscaling_mon_client.called
+ assert alarming_lcm_client.called
+ assert alarming_mon_client.called
+ assert healing_lcm_client.called
+ assert healing_mon_client.called
content = {
- 'notify_details': {
- 'alarm_uuid': 'test_alarm_uuid',
- 'metric_name': 'test_metric_name',
- 'operation': 'test_operation',
- 'threshold_value': 'test_threshold_value',
- 'vdu_name': 'test_vdu_name',
- 'vnf_member_index': 'test_vnf_member_index',
- 'ns_id': 'test_nsr_id'
+ "notify_details": {
+ "alarm_uuid": "test_alarm_uuid",
+ "metric_name": "test_metric_name",
+ "operation": "test_operation",
+ "threshold_value": "test_threshold_value",
+ "vdu_name": "test_vdu_name",
+ "vnf_member_index": "test_vnf_member_index",
+ "ns_id": "test_nsr_id",
+ "status": "alarm",
}
}
- mock_alarm = Mock()
- mock_alarm.vnf_member_index = 1
- mock_alarm.action = 'scale_out'
- mock_scaling_criteria = Mock()
- mock_scaling_policy = Mock()
- mock_scaling_group = Mock()
- mock_scaling_group.nsr_id = 'test_nsr_id'
- mock_scaling_group.name = 'test_name'
- mock_scaling_policy.cooldown_time = 60
- mock_scaling_policy.scaling_group = mock_scaling_group
- mock_scaling_criteria.scaling_policy = mock_scaling_policy
- mock_alarm.scaling_criteria = mock_scaling_criteria
- get_alarm.return_value = mock_alarm
- lcm_client.return_value.scale.side_effect = mock_scale
-
- mock_scaling_policy.last_scale = datetime.datetime.now() - datetime.timedelta(minutes=90)
+ autoscaling_handle_alarm.side_effect = mock_handle_alarm
+ alarming_handle_alarm.side_effect = mock_handle_alarm
+ healing_handle_alarm.side_effect = mock_handle_alarm
self.loop.run_until_complete(agent._handle_alarm_notification(content))
- lcm_client.return_value.scale.assert_called_with('test_nsr_id', 'test_name', 1, 'scale_out')
- lcm_client.return_value.scale.reset_mock()
+ autoscaling_handle_alarm.assert_called_with("test_alarm_uuid", "alarm")
+ alarming_handle_alarm.assert_called_with("test_alarm_uuid", "alarm", content)
+ healing_handle_alarm.assert_called_with("test_alarm_uuid", "alarm")
- mock_scaling_policy.last_scale = datetime.datetime.now()
- self.loop.run_until_complete(agent._handle_alarm_notification(content))
- lcm_client.return_value.scale.assert_not_called()
+ @mock.patch.object(CommonDbClient, "__init__", lambda *args, **kwargs: None)
+ @mock.patch("osm_policy_module.alarming.service.MonClient")
+ @mock.patch("osm_policy_module.alarming.service.LcmClient")
+ @mock.patch("osm_policy_module.autoscaling.service.MonClient")
+ @mock.patch("osm_policy_module.autoscaling.service.LcmClient")
+ @mock.patch.object(AutoscalingService, "delete_scaling_groups")
+ @mock.patch.object(AlarmingService, "delete_vnf_alarms")
+ def test_handle_vnf_terminated(
+ self,
+ delete_vnf_alarms,
+ delete_scaling_groups,
+ autoscaling_lcm_client,
+ autoscaling_mon_client,
+ alarming_lcm_client,
+ alarming_mon_client,
+ ):
+ async def mock_delete_scaling_groups(nsr_id, vnf_member_index):
+ pass
+
+ async def mock_delete_vnf_alarms(nsr_id, vnf_member_index):
+ pass
+
+ config = Config()
+ agent = PolicyModuleAgent(config)
+ assert autoscaling_lcm_client.called
+ assert autoscaling_mon_client.called
+ assert alarming_lcm_client.called
+ assert alarming_mon_client.called
+ content = {
+ "nsr_id": "test_nsr_id",
+ "vnf_member_index": "1",
+ "operationState": "COMPLETED",
+ }
+ failed_content = {
+ "nsr_id": "test_nsr_id",
+ "vnf_member_index": "1",
+ "operationState": "FAILED",
+ }
+ delete_scaling_groups.side_effect = mock_delete_scaling_groups
+ delete_vnf_alarms.side_effect = mock_delete_vnf_alarms
+
+ self.loop.run_until_complete(agent._handle_vnf_terminated(content))
+ delete_scaling_groups.assert_called_with("test_nsr_id", "1")
+ delete_scaling_groups.reset_mock()
+
+ self.loop.run_until_complete(agent._handle_vnf_terminated(failed_content))
+ delete_scaling_groups.assert_not_called()
-if __name__ == '__main__':
+if __name__ == "__main__":
unittest.main()
projects / osm / POL.git / blobdiff