From 9b109a59fb05234b71c79087346d9e78c3473c25 Mon Sep 17 00:00:00 2001 From: Alex Chrystal Date: Thu, 7 Nov 2024 12:54:39 -0500 Subject: [PATCH] adding action to get attribute for list of vms --- actions/vms_attribute_get.py | 60 +++++++++++ actions/vms_attribute_get.yaml | 24 +++++ tests/test_action_vms_attribute_get.py | 141 +++++++++++++++++++++++++ 3 files changed, 225 insertions(+) create mode 100644 actions/vms_attribute_get.py create mode 100644 actions/vms_attribute_get.yaml create mode 100644 tests/test_action_vms_attribute_get.py diff --git a/actions/vms_attribute_get.py b/actions/vms_attribute_get.py new file mode 100644 index 0000000..5b01ef5 --- /dev/null +++ b/actions/vms_attribute_get.py @@ -0,0 +1,60 @@ +#!/usr/bin/env python +# Copyright 2024 Encore Technologies +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from lib.action_base import BaseAction +from collections.abc import Mapping +import json + + +class VmsAttributeGet(BaseAction): + def run(self, attribute_name, vm_ids, open_nebula=None): + """ Return the value of the given attribute from the VMs + :returns: Dictionary with results and attribute_not_set lists + """ + results = [] + attribute_not_set = [] + + one = self.pyone_session_create(open_nebula) + + for vm_id in vm_ids: + vm = one.vm.info(vm_id) + return_value = vm + attribute_found = True + + for attr in attribute_name: + if hasattr(return_value, attr): + return_value = json.loads(json.dumps(getattr(return_value, attr))) + elif isinstance(return_value, Mapping) and attr in return_value: + return_value = return_value[attr] + else: + attribute_found = False + break + + if attribute_found: + results.append({ + 'vm_id': vm_id, + 'vm_name': vm.NAME, + attribute_name[-1]: return_value + }) + else: + attribute_not_set.append({ + 'vm_id': vm_id, + 'vm_name': vm.NAME + }) + + return { + 'vms_with_attribute': results, + 'attribute_not_set': attribute_not_set + } diff --git a/actions/vms_attribute_get.yaml b/actions/vms_attribute_get.yaml new file mode 100644 index 0000000..0c4dd09 --- /dev/null +++ b/actions/vms_attribute_get.yaml @@ -0,0 +1,24 @@ +--- +name: vms_attribute_get +runner_type: python-script +description: Return the value of the given attribute from the VM +enabled: true +entry_point: vms_attribute_get.py +parameters: + attribute_name: + type: array + description: > + Name/path of the attribute on the VM. This action returns only 1 value, multiple items + in this list will be interpreted as nested keys in the attributes object. + e.g. ['TEMPLATE','DISK'] will return the VM's disk info found at vm['TEMPLATE']['DISK'] + required: true + vm_ids: + type: array + description: Array of IDs of the VMs to get the given attribute for + required: true + open_nebula: + type: string + description: > + Pre-Configured ON connection details + required: false + default: ~ diff --git a/tests/test_action_vms_attribute_get.py b/tests/test_action_vms_attribute_get.py new file mode 100644 index 0000000..f35e027 --- /dev/null +++ b/tests/test_action_vms_attribute_get.py @@ -0,0 +1,141 @@ +#!/usr/bin/env python +# Copyright 2024 Encore Technologies +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +from one_base_action_test_case import OneBaseActionTestCase + +from vms_attribute_get import VmsAttributeGet +import unittest.mock as mock + +__all__ = [ + 'VmsAttributeGetTestCase' +] + + +class VmsAttributeGetTestCase(OneBaseActionTestCase): + __test__ = True + action_cls = VmsAttributeGet + + @mock.patch("lib.action_base.BaseAction.pyone_session_create") + def test_run_attribute_found(self, mock_session): + action = self.get_action_instance(self._config_good) + + # Define test parameters + mock_vm1 = mock.Mock() + mock_vm2 = mock.Mock() + mock_vm1.NAME = "vm1" + mock_vm2.NAME = "vm2" + mock_vm1.USER_TEMPLATE = {"BACKUPS": "true"} + mock_vm2.USER_TEMPLATE = {"BACKUPS": "false"} + + test_vms = [mock_vm1, mock_vm2] + vm_ids = [1, 2] + attribute_name = ["USER_TEMPLATE", "BACKUPS"] + open_nebula = "default" + expected_result = { + 'vms_with_attribute': [ + {'vm_id': 1, 'vm_name': "vm1", 'BACKUPS': "true"}, + {'vm_id': 2, 'vm_name': "vm2", 'BACKUPS': "false"} + ], + 'attribute_not_set': [] + } + + # Mock one object and run action + mock_one = mock.Mock() + mock_one.vm.info.side_effect = test_vms + mock_session.return_value = mock_one + result = action.run(attribute_name, vm_ids, open_nebula) + + # Verify result and calls + self.assertEqual(expected_result, result) + mock_session.assert_called_with(open_nebula) + mock_one.vm.info.assert_any_call(1) + mock_one.vm.info.assert_any_call(2) + + @mock.patch("lib.action_base.BaseAction.pyone_session_create") + def test_run_attribute_not_found(self, mock_session): + action = self.get_action_instance(self._config_good) + + # Define test parameters + mock_vm1 = mock.Mock() + mock_vm2 = mock.Mock() + mock_vm1.NAME = "vm1" + mock_vm2.NAME = "vm2" + mock_vm1.USER_TEMPLATE = {"DESCRIPTION": "desc"} + mock_vm2.USER_TEMPLATE = {"DESCRIPTION": "desc"} + + test_vms = [mock_vm1, mock_vm2] + vm_ids = [1, 2] + attribute_name = ["USER_TEMPLATE", "BACKUPS"] + open_nebula = "default" + expected_result = { + 'vms_with_attribute': [], + 'attribute_not_set': [ + {'vm_id': 1, 'vm_name': "vm1"}, + {'vm_id': 2, 'vm_name': "vm2"} + ] + } + + # Mock one object and run action + mock_one = mock.Mock() + mock_one.vm.info.side_effect = test_vms + mock_session.return_value = mock_one + result = action.run(attribute_name, vm_ids, open_nebula) + + # Verify result and calls + self.assertEqual(expected_result, result) + mock_session.assert_called_with(open_nebula) + mock_one.vm.info.assert_any_call(1) + mock_one.vm.info.assert_any_call(2) + + @mock.patch("lib.action_base.BaseAction.pyone_session_create") + def test_run_mixed_attributes(self, mock_session): + action = self.get_action_instance(self._config_good) + + # Define test parameters + mock_vm1 = mock.Mock() + mock_vm2 = mock.Mock() + mock_vm3 = mock.Mock() + mock_vm1.NAME = "vm1" + mock_vm2.NAME = "vm2" + mock_vm3.NAME = "vm3" + mock_vm1.USER_TEMPLATE = {"BACKUPS": "true"} + mock_vm2.USER_TEMPLATE = {"DESCRIPTION": "desc"} + mock_vm3.USER_TEMPLATE = {"BACKUPS": "false"} + + test_vms = [mock_vm1, mock_vm2, mock_vm3] + vm_ids = [1, 2, 3] + attribute_name = ["USER_TEMPLATE", "BACKUPS"] + open_nebula = "default" + expected_result = { + 'vms_with_attribute': [ + {'vm_id': 1, 'vm_name': "vm1", 'BACKUPS': "true"}, + {'vm_id': 3, 'vm_name': "vm3", 'BACKUPS': "false"} + ], + 'attribute_not_set': [ + {'vm_id': 2, 'vm_name': "vm2"} + ] + } + + # Mock one object and run action + mock_one = mock.Mock() + mock_one.vm.info.side_effect = test_vms + mock_session.return_value = mock_one + result = action.run(attribute_name, vm_ids, open_nebula) + + # Verify result and calls + self.assertEqual(expected_result, result) + mock_session.assert_called_with(open_nebula) + mock_one.vm.info.assert_any_call(1) + mock_one.vm.info.assert_any_call(2) + mock_one.vm.info.assert_any_call(3) \ No newline at end of file