X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;ds=sidebyside;f=osm_policy_module%2Ftests%2Funit%2Fautoscaling%2Ftest_autoscaling_service.py;h=053b278a4cb5d0a81e67e3b844aca6f35f3d2346;hb=4584f8e86a492d67d120bfea1195eff1475c0a65;hp=4491fbbdd28dfa2b3c1216274b7169863e7d7fcf;hpb=f7451f8879d188c9fef9c976d82aa39141601938;p=osm%2FPOL.git diff --git a/osm_policy_module/tests/unit/autoscaling/test_autoscaling_service.py b/osm_policy_module/tests/unit/autoscaling/test_autoscaling_service.py index 4491fbb..053b278 100644 --- a/osm_policy_module/tests/unit/autoscaling/test_autoscaling_service.py +++ b/osm_policy_module/tests/unit/autoscaling/test_autoscaling_service.py @@ -37,246 +37,297 @@ from osm_policy_module.core.database import ScalingAlarmRepository @mock.patch.object(MonClient, "__init__", lambda *args, **kwargs: None) @mock.patch.object(CommonDbClient, "__init__", lambda *args, **kwargs: None) 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') + @mock.patch.object(ScalingAlarmRepository, "get") + @mock.patch("osm_policy_module.core.database.db") def test_update_alarm_status(self, database, get_alarm): mock_alarm = mock.Mock() - mock_alarm.last_status = 'insufficient_data' + mock_alarm.last_status = "insufficient_data" get_alarm.return_value = mock_alarm service = AutoscalingService(self.config) - self.loop.run_until_complete(service.update_alarm_status('test_uuid', 'alarm')) - self.assertEqual(mock_alarm.last_status, 'alarm') + self.loop.run_until_complete(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')) - self.assertEqual(mock_alarm.last_status, 'ok') + self.loop.run_until_complete(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')) - self.assertEqual(mock_alarm.last_status, 'insufficient_data') + self.loop.run_until_complete( + service.update_alarm_status("test_uuid", "insufficient_data") + ) + self.assertEqual(mock_alarm.last_status, "insufficient_data") mock_alarm.save.assert_called_with() - @mock.patch.object(ScalingAlarmRepository, 'list') - @mock.patch.object(ScalingAlarmRepository, 'get') - @mock.patch('osm_policy_module.core.database.db') + @mock.patch.object(ScalingAlarmRepository, "list") + @mock.patch.object(ScalingAlarmRepository, "get") + @mock.patch("osm_policy_module.core.database.db") def test_evaluate_policy_not_enabled(self, database, get_alarm, list_alarms): mock_alarm = mock.Mock() mock_alarm.scaling_criteria.scaling_policy.enabled = False get_alarm.return_value = mock_alarm service = AutoscalingService(self.config) - self.loop.run_until_complete(service.evaluate_policy('test_uuid')) + self.loop.run_until_complete(service.evaluate_policy("test_uuid")) list_alarms.assert_not_called() - @mock.patch.object(ScalingAlarmRepository, 'list') - @mock.patch.object(ScalingAlarmRepository, 'get') - @mock.patch.object(LcmClient, 'scale') - @mock.patch('osm_policy_module.core.database.db') - def test_evaluate_policy_scale_in_and_equal(self, database, scale, get_alarm, list_alarms): + @mock.patch.object(ScalingAlarmRepository, "list") + @mock.patch.object(ScalingAlarmRepository, "get") + @mock.patch.object(LcmClient, "scale") + @mock.patch("osm_policy_module.core.database.db") + def test_evaluate_policy_scale_in_and_equal( + self, database, scale, get_alarm, list_alarms + ): """ Tests scale in with AND operation, both alarms triggered """ future = asyncio.Future(loop=self.loop) - future.set_result('mock') + future.set_result("mock") scale.return_value = future - mock_alarm = self._build_mock_alarm(action='scale_in', last_status='alarm', enabled=True, scale_in_op='AND') + mock_alarm = self._build_mock_alarm( + action="scale_in", last_status="alarm", enabled=True, scale_in_op="AND" + ) get_alarm.return_value = mock_alarm - mock_alarm_2 = self._build_mock_alarm(action='scale_in', last_status='alarm', enabled=True, scale_in_op='AND') + mock_alarm_2 = self._build_mock_alarm( + action="scale_in", last_status="alarm", enabled=True, scale_in_op="AND" + ) list_alarms.return_value = [mock_alarm, mock_alarm_2] service = AutoscalingService(self.config) - self.loop.run_until_complete(service.evaluate_policy('test_uuid')) - scale.assert_called_with('test_nsr_id', 'test_group', '1', 'scale_in') - - @mock.patch.object(ScalingAlarmRepository, 'list') - @mock.patch.object(ScalingAlarmRepository, 'get') - @mock.patch.object(LcmClient, 'scale') - @mock.patch('osm_policy_module.core.database.db') - def test_evaluate_policy_scale_in_and_diff(self, database, scale, get_alarm, list_alarms): + self.loop.run_until_complete(service.evaluate_policy("test_uuid")) + scale.assert_called_with("test_nsr_id", "test_group", "1", "scale_in") + + @mock.patch.object(ScalingAlarmRepository, "list") + @mock.patch.object(ScalingAlarmRepository, "get") + @mock.patch.object(LcmClient, "scale") + @mock.patch("osm_policy_module.core.database.db") + def test_evaluate_policy_scale_in_and_diff( + self, database, scale, get_alarm, list_alarms + ): """ Tests scale in with AND operation, only one alarm triggered. """ future = asyncio.Future(loop=self.loop) - future.set_result('mock') + future.set_result("mock") scale.return_value = future - mock_alarm = self._build_mock_alarm(action='scale_in', last_status='alarm', enabled=True, scale_in_op='AND') + mock_alarm = self._build_mock_alarm( + action="scale_in", last_status="alarm", enabled=True, scale_in_op="AND" + ) get_alarm.return_value = mock_alarm - mock_alarm_2 = self._build_mock_alarm(action='scale_in', last_status='ok', enabled=True, scale_in_op='OR') + mock_alarm_2 = self._build_mock_alarm( + action="scale_in", last_status="ok", enabled=True, scale_in_op="OR" + ) list_alarms.return_value = [mock_alarm, mock_alarm_2] service = AutoscalingService(self.config) - self.loop.run_until_complete(service.evaluate_policy('test_uuid')) + self.loop.run_until_complete(service.evaluate_policy("test_uuid")) scale.assert_not_called() - @mock.patch.object(ScalingAlarmRepository, 'list') - @mock.patch.object(ScalingAlarmRepository, 'get') - @mock.patch.object(LcmClient, 'scale') - @mock.patch('osm_policy_module.core.database.db') - def test_evaluate_policy_scale_in_or_equal(self, database, scale, get_alarm, list_alarms): + @mock.patch.object(ScalingAlarmRepository, "list") + @mock.patch.object(ScalingAlarmRepository, "get") + @mock.patch.object(LcmClient, "scale") + @mock.patch("osm_policy_module.core.database.db") + def test_evaluate_policy_scale_in_or_equal( + self, database, scale, get_alarm, list_alarms + ): """ Tests scale in with OR operation, both alarms triggered """ future = asyncio.Future(loop=self.loop) - future.set_result('mock') + future.set_result("mock") scale.return_value = future - mock_alarm = self._build_mock_alarm(action='scale_in', last_status='alarm', enabled=True, scale_in_op='OR') + mock_alarm = self._build_mock_alarm( + action="scale_in", last_status="alarm", enabled=True, scale_in_op="OR" + ) get_alarm.return_value = mock_alarm - mock_alarm_2 = self._build_mock_alarm(action='scale_in', last_status='alarm', enabled=True, scale_in_op='OR') + mock_alarm_2 = self._build_mock_alarm( + action="scale_in", last_status="alarm", enabled=True, scale_in_op="OR" + ) list_alarms.return_value = [mock_alarm, mock_alarm_2] service = AutoscalingService(self.config) - self.loop.run_until_complete(service.evaluate_policy('test_uuid')) - scale.assert_called_with('test_nsr_id', 'test_group', '1', 'scale_in') - - @mock.patch.object(ScalingAlarmRepository, 'list') - @mock.patch.object(ScalingAlarmRepository, 'get') - @mock.patch.object(LcmClient, 'scale') - @mock.patch('osm_policy_module.core.database.db') - def test_evaluate_policy_scale_in_or_diff(self, database, scale, get_alarm, list_alarms): + self.loop.run_until_complete(service.evaluate_policy("test_uuid")) + scale.assert_called_with("test_nsr_id", "test_group", "1", "scale_in") + + @mock.patch.object(ScalingAlarmRepository, "list") + @mock.patch.object(ScalingAlarmRepository, "get") + @mock.patch.object(LcmClient, "scale") + @mock.patch("osm_policy_module.core.database.db") + def test_evaluate_policy_scale_in_or_diff( + self, database, scale, get_alarm, list_alarms + ): """ Tests scale in with OR operation, only one alarm triggered """ future = asyncio.Future(loop=self.loop) - future.set_result('mock') + future.set_result("mock") scale.return_value = future - mock_alarm = self._build_mock_alarm(action='scale_in', last_status='alarm', enabled=True, scale_in_op='OR') + mock_alarm = self._build_mock_alarm( + action="scale_in", last_status="alarm", enabled=True, scale_in_op="OR" + ) get_alarm.return_value = mock_alarm - mock_alarm_2 = self._build_mock_alarm(action='scale_in', last_status='ok', enabled=True, scale_in_op='OR') + mock_alarm_2 = self._build_mock_alarm( + action="scale_in", last_status="ok", enabled=True, scale_in_op="OR" + ) list_alarms.return_value = [mock_alarm, mock_alarm_2] service = AutoscalingService(self.config) - self.loop.run_until_complete(service.evaluate_policy('test_uuid')) - scale.assert_called_with('test_nsr_id', 'test_group', '1', 'scale_in') - - @mock.patch.object(ScalingAlarmRepository, 'list') - @mock.patch.object(ScalingAlarmRepository, 'get') - @mock.patch.object(LcmClient, 'scale') - @mock.patch('osm_policy_module.core.database.db') - def test_evaluate_policy_scale_out_and_equal(self, database, scale, get_alarm, list_alarms): + self.loop.run_until_complete(service.evaluate_policy("test_uuid")) + scale.assert_called_with("test_nsr_id", "test_group", "1", "scale_in") + + @mock.patch.object(ScalingAlarmRepository, "list") + @mock.patch.object(ScalingAlarmRepository, "get") + @mock.patch.object(LcmClient, "scale") + @mock.patch("osm_policy_module.core.database.db") + def test_evaluate_policy_scale_out_and_equal( + self, database, scale, get_alarm, list_alarms + ): """ Tests scale out with AND operation, both alarms triggered """ future = asyncio.Future(loop=self.loop) - future.set_result('mock') + future.set_result("mock") scale.return_value = future - mock_alarm = self._build_mock_alarm(action='scale_out', last_status='alarm', enabled=True, scale_out_op='AND') + mock_alarm = self._build_mock_alarm( + action="scale_out", last_status="alarm", enabled=True, scale_out_op="AND" + ) get_alarm.return_value = mock_alarm - mock_alarm_2 = self._build_mock_alarm(action='scale_out', last_status='alarm', enabled=True, scale_out_op='AND') + mock_alarm_2 = self._build_mock_alarm( + action="scale_out", last_status="alarm", enabled=True, scale_out_op="AND" + ) list_alarms.return_value = [mock_alarm, mock_alarm_2] service = AutoscalingService(self.config) - self.loop.run_until_complete(service.evaluate_policy('test_uuid')) - scale.assert_called_with('test_nsr_id', 'test_group', '1', 'scale_out') - - @mock.patch.object(ScalingAlarmRepository, 'list') - @mock.patch.object(ScalingAlarmRepository, 'get') - @mock.patch.object(LcmClient, 'scale') - @mock.patch('osm_policy_module.core.database.db') - def test_evaluate_policy_scale_out_and_diff(self, database, scale, get_alarm, list_alarms): + self.loop.run_until_complete(service.evaluate_policy("test_uuid")) + scale.assert_called_with("test_nsr_id", "test_group", "1", "scale_out") + + @mock.patch.object(ScalingAlarmRepository, "list") + @mock.patch.object(ScalingAlarmRepository, "get") + @mock.patch.object(LcmClient, "scale") + @mock.patch("osm_policy_module.core.database.db") + def test_evaluate_policy_scale_out_and_diff( + self, database, scale, get_alarm, list_alarms + ): """ Tests scale out with AND operation, only one alarm triggered. """ future = asyncio.Future(loop=self.loop) - future.set_result('mock') + future.set_result("mock") scale.return_value = future - mock_alarm = self._build_mock_alarm(action='scale_out', last_status='alarm', enabled=True, scale_out_op='AND') + mock_alarm = self._build_mock_alarm( + action="scale_out", last_status="alarm", enabled=True, scale_out_op="AND" + ) get_alarm.return_value = mock_alarm - mock_alarm_2 = self._build_mock_alarm(action='scale_out', last_status='ok', enabled=True, scale_out_op='OR') + mock_alarm_2 = self._build_mock_alarm( + action="scale_out", last_status="ok", enabled=True, scale_out_op="OR" + ) list_alarms.return_value = [mock_alarm, mock_alarm_2] service = AutoscalingService(self.config) - self.loop.run_until_complete(service.evaluate_policy('test_uuid')) + self.loop.run_until_complete(service.evaluate_policy("test_uuid")) scale.assert_not_called() - @mock.patch.object(ScalingAlarmRepository, 'list') - @mock.patch.object(ScalingAlarmRepository, 'get') - @mock.patch.object(LcmClient, 'scale') - @mock.patch('osm_policy_module.core.database.db') - def test_evaluate_policy_scale_out_or_equal(self, database, scale, get_alarm, list_alarms): + @mock.patch.object(ScalingAlarmRepository, "list") + @mock.patch.object(ScalingAlarmRepository, "get") + @mock.patch.object(LcmClient, "scale") + @mock.patch("osm_policy_module.core.database.db") + def test_evaluate_policy_scale_out_or_equal( + self, database, scale, get_alarm, list_alarms + ): """ Tests scale out with OR operation, both alarms triggered """ future = asyncio.Future(loop=self.loop) - future.set_result('mock') + future.set_result("mock") scale.return_value = future - mock_alarm = self._build_mock_alarm(action='scale_out', last_status='alarm', enabled=True, scale_out_op='OR') + mock_alarm = self._build_mock_alarm( + action="scale_out", last_status="alarm", enabled=True, scale_out_op="OR" + ) get_alarm.return_value = mock_alarm - mock_alarm_2 = self._build_mock_alarm(action='scale_out', last_status='alarm', enabled=True, scale_out_op='OR') + mock_alarm_2 = self._build_mock_alarm( + action="scale_out", last_status="alarm", enabled=True, scale_out_op="OR" + ) list_alarms.return_value = [mock_alarm, mock_alarm_2] service = AutoscalingService(self.config) - self.loop.run_until_complete(service.evaluate_policy('test_uuid')) - scale.assert_called_with('test_nsr_id', 'test_group', '1', 'scale_out') - - @mock.patch.object(ScalingAlarmRepository, 'list') - @mock.patch.object(ScalingAlarmRepository, 'get') - @mock.patch.object(LcmClient, 'scale') - @mock.patch('osm_policy_module.core.database.db') - def test_evaluate_policy_scale_out_or_diff(self, database, scale, get_alarm, list_alarms): + self.loop.run_until_complete(service.evaluate_policy("test_uuid")) + scale.assert_called_with("test_nsr_id", "test_group", "1", "scale_out") + + @mock.patch.object(ScalingAlarmRepository, "list") + @mock.patch.object(ScalingAlarmRepository, "get") + @mock.patch.object(LcmClient, "scale") + @mock.patch("osm_policy_module.core.database.db") + def test_evaluate_policy_scale_out_or_diff( + self, database, scale, get_alarm, list_alarms + ): """ Tests scale out with OR operation, only one alarm triggered """ future = asyncio.Future(loop=self.loop) - future.set_result('mock') + future.set_result("mock") scale.return_value = future - mock_alarm = self._build_mock_alarm(action='scale_out', last_status='alarm', enabled=True, scale_out_op='OR') + mock_alarm = self._build_mock_alarm( + action="scale_out", last_status="alarm", enabled=True, scale_out_op="OR" + ) get_alarm.return_value = mock_alarm - mock_alarm_2 = self._build_mock_alarm(action='scale_out', last_status='ok', enabled=True, scale_out_op='OR') + mock_alarm_2 = self._build_mock_alarm( + action="scale_out", last_status="ok", enabled=True, scale_out_op="OR" + ) list_alarms.return_value = [mock_alarm, mock_alarm_2] service = AutoscalingService(self.config) - self.loop.run_until_complete(service.evaluate_policy('test_uuid')) - scale.assert_called_with('test_nsr_id', 'test_group', '1', 'scale_out') - - def _build_mock_alarm(self, - action='scale_in', - last_status='alarm', - last_scale=datetime.datetime.min, - cooldown_time=10, - enabled=True, - scale_in_op='AND', - scale_out_op='AND'): + self.loop.run_until_complete(service.evaluate_policy("test_uuid")) + scale.assert_called_with("test_nsr_id", "test_group", "1", "scale_out") + + def _build_mock_alarm( + self, + action="scale_in", + last_status="alarm", + last_scale=datetime.datetime.min, + cooldown_time=10, + enabled=True, + scale_in_op="AND", + scale_out_op="AND", + ): mock_alarm = mock.Mock() mock_alarm.action = action mock_alarm.last_status = last_status - mock_alarm.vnf_member_index = '1' + mock_alarm.vnf_member_index = "1" mock_alarm.scaling_criteria.scaling_policy.last_scale = last_scale mock_alarm.scaling_criteria.scaling_policy.cooldown_time = cooldown_time mock_alarm.scaling_criteria.scaling_policy.enabled = enabled mock_alarm.scaling_criteria.scaling_policy.scale_in_operation = scale_in_op mock_alarm.scaling_criteria.scaling_policy.scale_out_operation = scale_out_op - mock_alarm.scaling_criteria.scaling_policy.scaling_group.nsr_id = 'test_nsr_id' - mock_alarm.scaling_criteria.scaling_policy.scaling_group.name = 'test_group' + mock_alarm.scaling_criteria.scaling_policy.scaling_group.nsr_id = "test_nsr_id" + mock_alarm.scaling_criteria.scaling_policy.scaling_group.name = "test_group" return mock_alarm