- @mock.patch('osm_policy_module.autoscaling.service.CommonDbClient')
- @mock.patch('osm_policy_module.autoscaling.service.MonClient')
- @mock.patch('osm_policy_module.autoscaling.service.LcmClient')
- @mock.patch('osm_policy_module.alarming.service.CommonDbClient')
- @mock.patch('osm_policy_module.alarming.service.MonClient')
- @mock.patch('osm_policy_module.alarming.service.LcmClient')
- @mock.patch.object(AutoscalingService, 'handle_alarm')
- @mock.patch.object(AlarmingService, 'handle_alarm')
- def test_handle_alarm_notification(self,
- alarming_handle_alarm,
- autoscaling_handle_alarm,
- autoscaling_lcm_client,
- autoscaling_mon_client,
- autoscaling_db_client,
- alarming_lcm_client,
- alarming_mon_client,
- alarming_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(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",
+ }
+ 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
+
+ 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()
+
+ self.loop.run_until_complete(agent._handle_policy_update(failed_content))
+ configure_scaling_groups.assert_not_called()
+
+ @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,
+ ):