feature/vms-get-with-label-action

CHANGES.md

@@ -1,3 +1,6 @@
+## 1.1.7
+- Added action to get list of VMs with given label
+
 ## 1.1.6
 - Adds hypervisor type to data get action
 

actions/vms_get_with_label.py

@@ -0,0 +1,45 @@
+#!/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
+import xmltojson
+import json
+
+
+class VmsGetWithLabel(BaseAction):
+    def run(self, label_name, open_nebula=None):
+        """ Retrieves a list of VMs on an Open Nebula system that include the given label
+        :returns: VM information with name and ID appended
+        """
+        one_session = self.xmlrpc_session_create(open_nebula)
+        response = one_session.one.vmpool.infoextended(self.auth_string, *tuple([-2, -1, -1, -1]))
+
+        # Check the result for an error (first element will be FALSE on error)
+        if response[0]:
+            vm_pool = json.loads(xmltojson.parse(response[1]))
+            vms = vm_pool['VM_POOL']['VM']
+        else:
+            raise Exception(response[1])
+
+        label_vms = []
+        for vm in vms:
+            try:
+                labels = vm['USER_TEMPLATE']['LABELS']
+                if labels and label_name in labels.split(','):
+                    label_vms.append(vm)
+            except KeyError:
+                continue
+
+        return label_vms

actions/vms_get_with_label.yaml

@@ -0,0 +1,17 @@
+---
+name: vms_get_with_label
+runner_type: python-script
+description: Retrieves a list of VMs that have the given label
+enabled: true
+entry_point: vms_get_with_label.py
+parameters:
+    label_name:
+        type: string
+        description: Name of the label to return a list of VMs with
+        required: true
+    open_nebula:
+        type: string
+        description: >
+            Pre-Configured Open Nebula connection details
+        required: false
+        default: ~

pack.yaml

@@ -3,7 +3,7 @@ ref: open_nebula
 name: open_nebula
 description: Open Nebula
 stackstorm_version: ">=2.9.0"
-version: 1.1.6
+version: 1.1.7
 author: John Schoewe
 email: john.schoewe@encore.tech
 contributors:

tests/test_action_vms_get_with_label.py

@@ -0,0 +1,101 @@
+#!/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_get_with_label import VmsGetWithLabel
+import unittest.mock as mock
+import xmltodict
+
+
+__all__ = [
+    'VmsGetWithLabelTestCase'
+]
+
+
+class VmsGetWithLabelTestCase(OneBaseActionTestCase):
+    __test__ = True
+    action_cls = VmsGetWithLabel
+
+    @mock.patch("lib.action_base.BaseAction.xmlrpc_session_create")
+    def test_run(self, mock_session):
+        action = self.get_action_instance(self._config_good)
+
+        # Define test parameters
+        label_name = 'LABEL'
+        open_nebula = 'default'
+        auth_string = 'user:pass'
+        action.auth_string = auth_string
+        expected_result = 'test-server.com'
+
+        # Mock one object and run action
+        vm1 = {
+            'NAME': 'vm1',
+            'USER_TEMPLATE': {
+                'LABELS': 'LABEL'
+            }
+        }
+        vm2 = {
+            'NAME': 'vm2',
+            'USER_TEMPLATE': {
+                'LABELS': 'NOTFOUND'
+            }
+        }
+        vm3 = {
+            'NAME': 'vm3',
+            'USER_TEMPLATE': {
+                'LABELS': 'TEST,LABEL'
+            }
+        }
+        vm4 = {
+            'NAME': 'vm4',
+            'USER_TEMPLATE': {'DONT': 'INCLUDE'}
+        }
+        test_vms = {
+            'VM_POOL': {
+                'VM': [vm1, vm2, vm3, vm4]
+            }
+        }
+        expected_result = [vm1, vm3]
+
+        mock_one = mock.Mock()
+        # Convert the templates dict to xml for the xmltojson.parse function in the action
+        mock_one.one.vmpool.infoextended.return_value = [
+            True,
+            xmltodict.unparse(test_vms, pretty=True)
+        ]
+        mock_session.return_value = mock_one
+        result = action.run(label_name, open_nebula)
+
+        # Verify result and calls
+        self.assertEqual(expected_result, result)
+        mock_session.assert_called_with(open_nebula)
+        mock_one.one.vmpool.infoextended.assert_called_with(auth_string, *tuple([-2, -1, -1, -1]))
+
+    @mock.patch("lib.action_base.BaseAction.xmlrpc_session_create")
+    def test_run_error(self, mock_session):
+        action = self.get_action_instance(self._config_good)
+
+        # Define test parameters
+        label_name = 'LABEL'
+        open_nebula = 'default'
+        auth_string = 'user:pass'
+        action.auth_string = auth_string
+
+        # Mock one object and run action
+        mock_one = mock.Mock()
+        mock_one.one.vmpool.infoextended.return_value = [False, 'error']
+        mock_session.return_value = mock_one
+
+        with self.assertRaises(Exception):
+            action.run(label_name, open_nebula)