class TestAutoscalingService(TestCase):
def setUp(self):
self.config = Config()
- self.loop = asyncio.new_event_loop()
@mock.patch.object(ScalingAlarmRepository, "get")
@mock.patch("osm_policy_module.core.database.db")
get_alarm.return_value = mock_alarm
service = AutoscalingService(self.config)
- self.loop.run_until_complete(service.update_alarm_status("test_uuid", "alarm"))
+ asyncio.run(service.update_alarm_status("test_uuid", "alarm"))
self.assertEqual(mock_alarm.last_status, "alarm")
mock_alarm.save.assert_called_with()
service = AutoscalingService(self.config)
- self.loop.run_until_complete(service.update_alarm_status("test_uuid", "ok"))
+ asyncio.run(service.update_alarm_status("test_uuid", "ok"))
self.assertEqual(mock_alarm.last_status, "ok")
mock_alarm.save.assert_called_with()
service = AutoscalingService(self.config)
- self.loop.run_until_complete(
- service.update_alarm_status("test_uuid", "insufficient_data")
- )
+ asyncio.run(service.update_alarm_status("test_uuid", "insufficient_data"))
self.assertEqual(mock_alarm.last_status, "insufficient_data")
mock_alarm.save.assert_called_with()
get_alarm.return_value = mock_alarm
service = AutoscalingService(self.config)
- self.loop.run_until_complete(service.evaluate_policy("test_uuid"))
+ asyncio.run(service.evaluate_policy("test_uuid"))
list_alarms.assert_not_called()
@mock.patch.object(ScalingAlarmRepository, "list")
"""
Tests scale in with AND operation, both alarms triggered
"""
- future = asyncio.Future(loop=self.loop)
+ future = asyncio.Future(loop=asyncio.new_event_loop())
future.set_result("mock")
scale.return_value = future
list_alarms.return_value = [mock_alarm, mock_alarm_2]
service = AutoscalingService(self.config)
- self.loop.run_until_complete(service.evaluate_policy("test_uuid"))
+ asyncio.run(service.evaluate_policy("test_uuid"))
scale.assert_called_with("test_nsr_id", "test_group", "1", "scale_in")
@mock.patch.object(ScalingAlarmRepository, "list")
"""
Tests scale in with AND operation, only one alarm triggered.
"""
- future = asyncio.Future(loop=self.loop)
+ future = asyncio.Future(loop=asyncio.new_event_loop())
future.set_result("mock")
scale.return_value = future
list_alarms.return_value = [mock_alarm, mock_alarm_2]
service = AutoscalingService(self.config)
- self.loop.run_until_complete(service.evaluate_policy("test_uuid"))
+ asyncio.run(service.evaluate_policy("test_uuid"))
scale.assert_not_called()
@mock.patch.object(ScalingAlarmRepository, "list")
"""
Tests scale in with OR operation, both alarms triggered
"""
- future = asyncio.Future(loop=self.loop)
+ future = asyncio.Future(loop=asyncio.new_event_loop())
future.set_result("mock")
scale.return_value = future
list_alarms.return_value = [mock_alarm, mock_alarm_2]
service = AutoscalingService(self.config)
- self.loop.run_until_complete(service.evaluate_policy("test_uuid"))
+ asyncio.run(service.evaluate_policy("test_uuid"))
scale.assert_called_with("test_nsr_id", "test_group", "1", "scale_in")
@mock.patch.object(ScalingAlarmRepository, "list")
"""
Tests scale in with OR operation, only one alarm triggered
"""
- future = asyncio.Future(loop=self.loop)
+ future = asyncio.Future(loop=asyncio.new_event_loop())
future.set_result("mock")
scale.return_value = future
list_alarms.return_value = [mock_alarm, mock_alarm_2]
service = AutoscalingService(self.config)
- self.loop.run_until_complete(service.evaluate_policy("test_uuid"))
+ asyncio.run(service.evaluate_policy("test_uuid"))
scale.assert_called_with("test_nsr_id", "test_group", "1", "scale_in")
@mock.patch.object(ScalingAlarmRepository, "list")
"""
Tests scale out with AND operation, both alarms triggered
"""
- future = asyncio.Future(loop=self.loop)
+ future = asyncio.Future(loop=asyncio.new_event_loop())
future.set_result("mock")
scale.return_value = future
list_alarms.return_value = [mock_alarm, mock_alarm_2]
service = AutoscalingService(self.config)
- self.loop.run_until_complete(service.evaluate_policy("test_uuid"))
+ asyncio.run(service.evaluate_policy("test_uuid"))
scale.assert_called_with("test_nsr_id", "test_group", "1", "scale_out")
@mock.patch.object(ScalingAlarmRepository, "list")
"""
Tests scale out with AND operation, only one alarm triggered.
"""
- future = asyncio.Future(loop=self.loop)
+ future = asyncio.Future(loop=asyncio.new_event_loop())
future.set_result("mock")
scale.return_value = future
list_alarms.return_value = [mock_alarm, mock_alarm_2]
service = AutoscalingService(self.config)
- self.loop.run_until_complete(service.evaluate_policy("test_uuid"))
+ asyncio.run(service.evaluate_policy("test_uuid"))
scale.assert_not_called()
@mock.patch.object(ScalingAlarmRepository, "list")
"""
Tests scale out with OR operation, both alarms triggered
"""
- future = asyncio.Future(loop=self.loop)
+ future = asyncio.Future(loop=asyncio.new_event_loop())
future.set_result("mock")
scale.return_value = future
list_alarms.return_value = [mock_alarm, mock_alarm_2]
service = AutoscalingService(self.config)
- self.loop.run_until_complete(service.evaluate_policy("test_uuid"))
+ asyncio.run(service.evaluate_policy("test_uuid"))
scale.assert_called_with("test_nsr_id", "test_group", "1", "scale_out")
@mock.patch.object(ScalingAlarmRepository, "list")
"""
Tests scale out with OR operation, only one alarm triggered
"""
- future = asyncio.Future(loop=self.loop)
+ future = asyncio.Future(loop=asyncio.new_event_loop())
future.set_result("mock")
scale.return_value = future
list_alarms.return_value = [mock_alarm, mock_alarm_2]
service = AutoscalingService(self.config)
- self.loop.run_until_complete(service.evaluate_policy("test_uuid"))
+ asyncio.run(service.evaluate_policy("test_uuid"))
scale.assert_called_with("test_nsr_id", "test_group", "1", "scale_out")
def _build_mock_alarm(