self.loop = asyncio.new_event_loop()
@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, 'configure_scaling_groups')
- @mock.patch.object(AlarmingService, 'configure_vnf_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_vnf_alarms,
- configure_scaling_groups,
- autoscaling_lcm_client,
- autoscaling_mon_client,
- alarming_lcm_client,
- alarming_mon_client):
+ @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, "configure_scaling_groups")
+ @mock.patch.object(AlarmingService, "configure_vnf_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_vnf_alarms,
+ configure_scaling_groups,
+ autoscaling_lcm_client,
+ autoscaling_mon_client,
+ alarming_lcm_client,
+ alarming_mon_client,
+ ):
async def mock_configure_scaling_groups(nsr_id):
pass
assert alarming_lcm_client.called
assert alarming_mon_client.called
content = {
- 'nslcmop_id': 'test_id',
+ "nslcmop_id": "test_id",
}
nslcmop_completed = {
- 'operationState': 'COMPLETED',
- 'nsInstanceId': 'test_nsr_id'
- }
- nslcmop_failed = {
- 'operationState': 'FAILED',
- '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
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.assert_called_with("test_nsr_id")
configure_scaling_groups.reset_mock()
get_nslcmop.return_value = nslcmop_failed
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.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,
- alarming_lcm_client,
- alarming_mon_client):
+ @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.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,
+ alarming_lcm_client,
+ alarming_mon_client,
+ ):
async def mock_handle_alarm(alarm_uuid, status, payload=None):
pass
assert alarming_lcm_client.called
assert alarming_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',
- 'status': 'alarm'
+ "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",
}
}
autoscaling_handle_alarm.side_effect = mock_handle_alarm
alarming_handle_alarm.side_effect = mock_handle_alarm
self.loop.run_until_complete(agent._handle_alarm_notification(content))
- autoscaling_handle_alarm.assert_called_with('test_alarm_uuid', 'alarm')
- alarming_handle_alarm.assert_called_with('test_alarm_uuid', 'alarm', content)
+ autoscaling_handle_alarm.assert_called_with("test_alarm_uuid", "alarm")
+ alarming_handle_alarm.assert_called_with("test_alarm_uuid", "alarm", content)
-if __name__ == '__main__':
+if __name__ == "__main__":
unittest.main()