Initial commit for xenserver module_util unit tests

test/units/module_utils/xenserver/FakeAnsibleModule.py

@@ -0,0 +1,30 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright: (c) 2019, Bojan Vitnik <bvitnik@mainstream.rs>
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+
+class AnsibleModuleException(Exception):
+    def __init__(self, *args, **kwargs):
+        self.args = args
+        self.kwargs = kwargs
+
+
+class ExitJsonException(AnsibleModuleException):
+    pass
+
+
+class FailJsonException(AnsibleModuleException):
+    pass
+
+
+class FakeAnsibleModule:
+    def __init__(self, params=None, check_mode=False):
+        self.params = params
+        self.check_mode = check_mode
+
+    def exit_json(self, *args, **kwargs):
+        raise ExitJsonException(*args, **kwargs)
+
+    def fail_json(self, *args, **kwargs):
+        raise FailJsonException(*args, **kwargs)

test/units/module_utils/xenserver/FakeXenAPI.py

@@ -0,0 +1,66 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright: (c) 2019, Bojan Vitnik <bvitnik@mainstream.rs>
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+FAKE_API_VERSION = "1.1"
+
+
+class Failure(Exception):
+    def __init__(self, details):
+        self.details = details
+
+    def __str__(self):
+        return str(self.details)
+
+
+class Session(object):
+    def __init__(self, uri, transport=None, encoding=None, verbose=0,
+                 allow_none=1, ignore_ssl=False):
+
+        self.transport = transport
+        self._session = None
+        self.last_login_method = None
+        self.last_login_params = None
+        self.API_version = FAKE_API_VERSION
+
+    def _get_api_version(self):
+        return FAKE_API_VERSION
+
+    def _login(self, method, params):
+        self._session = "OpaqueRef:fake-xenapi-session-ref"
+        self.last_login_method = method
+        self.last_login_params = params
+        self.API_version = self._get_api_version()
+
+    def _logout(self):
+        self._session = None
+        self.last_login_method = None
+        self.last_login_params = None
+        self.API_version = FAKE_API_VERSION
+
+    def xenapi_request(self, methodname, params):
+        if methodname.startswith('login'):
+            self._login(methodname, params)
+            return None
+        elif methodname == 'logout' or methodname == 'session.logout':
+            self._logout()
+            return None
+        else:
+            # Should be patched with mocker.patch().
+            return None
+
+    def __getattr__(self, name):
+        if name == 'handle':
+            return self._session
+        elif name == 'xenapi':
+            # Should be patched with mocker.patch().
+            return None
+        elif name.startswith('login') or name.startswith('slave_local'):
+            return lambda *params: self._login(name, params)
+        elif name == 'logout':
+            return self._logout
+
+
+def xapi_local():
+    return Session("http://_var_lib_xcp_xapi/")

test/units/module_utils/xenserver/common.py

@@ -0,0 +1,25 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright: (c) 2019, Bojan Vitnik <bvitnik@mainstream.rs>
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+
+def fake_xenapi_ref(xenapi_class):
+    return "OpaqueRef:fake-xenapi-%s-ref" % xenapi_class
+
+
+testcase_bad_xenapi_refs = {
+    "params": [
+        None,
+        '',
+        'OpaqueRef:NULL',
+    ],
+    "ids": [
+        'none',
+        'empty',
+        'ref-null',
+    ],
+}

test/units/module_utils/xenserver/conftest.py

@@ -0,0 +1,118 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright: (c) 2019, Bojan Vitnik <bvitnik@mainstream.rs>
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+
+import sys
+import importlib
+import os
+import json
+import pytest
+
+from .FakeAnsibleModule import FakeAnsibleModule, ExitJsonException, FailJsonException
+from ansible.module_utils import six
+from mock import MagicMock
+
+
+@pytest.fixture
+def fake_ansible_module(request):
+    """Returns fake AnsibleModule with fake module params."""
+    if hasattr(request, 'param'):
+        return FakeAnsibleModule(request.param)
+    else:
+        params = {
+            "hostname": "somehost",
+            "username": "someuser",
+            "password": "somepwd",
+            "validate_certs": True,
+        }
+
+        return FakeAnsibleModule(params)
+
+
+@pytest.fixture(autouse=True)
+def XenAPI():
+    """Imports and returns fake XenAPI module."""
+
+    # Import of fake XenAPI module is wrapped by fixture so that it does not
+    # affect other unit tests which could potentialy also use XenAPI module.
+
+    # First we use importlib.import_module() to import the module and assign
+    # it to a local symbol.
+    fake_xenapi = importlib.import_module('units.module_utils.xenserver.FakeXenAPI')
+
+    # Now we populate Python module cache with imported fake module using the
+    # original module name (XenAPI). That way, any 'import XenAPI' statement
+    # will just load already imported fake module from the cache.
+    sys.modules['XenAPI'] = fake_xenapi
+
+    return fake_xenapi
+
+
+@pytest.fixture(autouse=True)
+def xenserver(XenAPI):
+    """Imports and returns xenserver module util."""
+
+    # Since we are wrapping fake XenAPI module inside a fixture, all modules
+    # that depend on it have to be imported inside a test function. To make
+    # this easier to handle and remove some code repetition, we wrap the import
+    # of xenserver module util with a fixture.
+    from ansible.module_utils import xenserver
+
+    return xenserver
+
+
+@pytest.fixture
+def mock_xenapi_failure(XenAPI, mocker):
+    """
+    Returns mock object that raises XenAPI.Failure on any XenAPI
+    method call.
+    """
+    fake_error_msg = "Fake XAPI method call error!"
+
+    # We need to use our MagicMock based class that passes side_effect to its
+    # children because calls to xenapi methods can generate an arbitrary
+    # hierarchy of mock objects. Any such object when called should use the
+    # same side_effect as its parent mock object.
+    class MagicMockSideEffect(MagicMock):
+        def _get_child_mock(self, **kw):
+            child_mock = super(MagicMockSideEffect, self)._get_child_mock(**kw)
+            child_mock.side_effect = self.side_effect
+            return child_mock
+
+    mocked_xenapi = mocker.patch.object(XenAPI.Session, 'xenapi', new=MagicMockSideEffect(), create=True)
+    mocked_xenapi.side_effect = XenAPI.Failure(fake_error_msg)
+
+    return mocked_xenapi, fake_error_msg
+
+
+@pytest.fixture
+def fixture_data_from_file(request):
+    """Loads fixture data from files."""
+    if not hasattr(request, 'param'):
+        return {}
+
+    fixture_path = os.path.join(os.path.dirname(__file__), 'fixtures')
+    fixture_data = {}
+
+    if isinstance(request.param, six.string_types):
+        request.param = [request.param]
+
+    for fixture_name in request.param:
+        path = os.path.join(fixture_path, fixture_name)
+
+        with open(path) as f:
+            data = f.read()
+
+        try:
+            data = json.loads(data)
+        except Exception:
+            pass
+
+        fixture_data[fixture_name] = data
+
+    return fixture_data

test/units/module_utils/xenserver/fixtures/ansible-test-vm-1-facts.json

@@ -0,0 +1,73 @@
+{
+    "cdrom": {
+        "type": "none"
+    },
+    "customization_agent": "native",
+    "disks": [
+        {
+            "name": "ansible-test-vm-1-C",
+            "name_desc": "C:\\",
+            "os_device": "xvda",
+            "size": 42949672960,
+            "sr": "Ansible Test Storage 1",
+            "sr_uuid": "767b30e4-f8db-a83d-8ba7-f5e6e732e06f",
+            "vbd_userdevice": "0"
+        }
+    ],
+    "domid": "143",
+    "folder": "/Ansible/Test",
+    "hardware": {
+        "memory_mb": 2048,
+        "num_cpu_cores_per_socket": 2,
+        "num_cpus": 2
+    },
+    "home_server": "",
+    "is_template": false,
+    "name": "ansible-test-vm-1",
+    "name_desc": "Created by Ansible",
+    "networks": [
+        {
+            "gateway": "10.0.0.1",
+            "gateway6": "",
+            "ip": "10.0.0.2",
+            "ip6": [
+                "fe80:0000:0000:0000:11e1:12c9:ef3b:75a0"
+            ],
+            "mac": "7a:a6:48:1e:31:46",
+            "mtu": "1500",
+            "name": "Host internal management network",
+            "netmask": "255.255.255.0",
+            "prefix": "24",
+            "prefix6": "",
+            "vif_device": "0"
+        }
+    ],
+    "other_config": {
+        "base_template_name": "Windows Server 2016 (64-bit)",
+        "folder": "/Ansible/Test",
+        "import_task": "OpaqueRef:e43eb71c-45d6-5351-09ff-96e4fb7d0fa5",
+        "install-methods": "cdrom",
+        "instant": "true",
+        "mac_seed": "366fe8e0-878b-4320-8731-90d1ed3c0b93"
+    },
+    "platform": {
+        "acpi": "1",
+        "apic": "true",
+        "cores-per-socket": "2",
+        "device_id": "0002",
+        "hpet": "true",
+        "nx": "true",
+        "pae": "true",
+        "timeoffset": "-28800",
+        "vga": "std",
+        "videoram": "8",
+        "viridian": "true",
+        "viridian_reference_tsc": "true",
+        "viridian_time_ref_count": "true"
+    },
+    "state": "poweredon",
+    "uuid": "81c373d7-a407-322f-911b-31386eb5215d",
+    "xenstore_data": {
+        "vm-data": ""
+    }
+}