adding action to get attribute for list of vms

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
+        }

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: ~

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)