De-couple backend logic from module logic

mariolenz · mariolenz · commit 0aadaac8f19f · 2025-03-12T19:00:31.000+01:00
diff --git a/plugins/module_utils/clients/_vmware.py b/plugins/module_utils/clients/_vmware.py
@@ -0,0 +1,121 @@
+# -*- coding: utf-8 -*-
+
+# Copyright: (c) 2015, Joseph Callen <jcallen () csc.com>
+# Copyright: (c) 2018, Ansible Project
+# Copyright: (c) 2018, James E. King III (@jeking3) <jking@apache.org>
+# Copyright: (c) 2025, Mario Lenz (@mariolenz) <m@riolenz.de>
+# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+__metaclass__ = type
+
+import atexit
+import ssl
+import traceback
+
+from ansible.module_utils.basic import missing_required_lib
+
+REQUESTS_IMP_ERR = None
+try:
+    # requests is required for exception handling of the ConnectionError
+    import requests
+except ImportError:
+    REQUESTS_IMP_ERR = traceback.format_exc()
+
+PYVMOMI_IMP_ERR = None
+try:
+    from pyVim import connect
+    from pyVmomi import vim, vmodl
+except ImportError:
+    PYVMOMI_IMP_ERR = traceback.format_exc()
+
+
+class MissingLibError(Exception):
+    def __init__(self, library, exception, url=None):
+        self.exception = exception
+        self.library = library
+        self.url = url
+        super().__init__(missing_required_lib(self.library, url=self.url))
+
+
+class ApiAccessError(Exception):
+    def __init__(self, *args, **kwargs):
+        super(ApiAccessError, self).__init__(*args, **kwargs)
+
+
+class PyvmomiClient(object):
+    def __init__(self, hostname, username, password, port=443, validate_certs=True, http_proxy_host=None, http_proxy_port=None):
+        if REQUESTS_IMP_ERR:
+            raise MissingLibError('requests', REQUESTS_IMP_ERR)
+
+        if PYVMOMI_IMP_ERR:
+            raise MissingLibError('pyvmomi', PYVMOMI_IMP_ERR)
+
+        self.si, self.content = self._connect_to_api(hostname, username, password, port, validate_certs, http_proxy_host, http_proxy_port)
+
+    def _connect_to_api(self, hostname, username, password, port, validate_certs, http_proxy_host, http_proxy_port):
+        if not hostname:
+            raise ApiAccessError("Hostname parameter is missing.")
+
+        if not username:
+            raise ApiAccessError("Username parameter is missing.")
+
+        if not password:
+            raise ApiAccessError("Password parameter is missing.")
+
+        if validate_certs and not hasattr(ssl, 'SSLContext'):
+            raise ApiAccessError('pyVim does not support changing verification mode with python < 2.7.9. Either update '
+                                 'python or use validate_certs=false.')
+        elif validate_certs:
+            ssl_context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
+            ssl_context.verify_mode = ssl.CERT_REQUIRED
+            ssl_context.check_hostname = True
+            ssl_context.load_default_certs()
+        elif hasattr(ssl, 'SSLContext'):
+            ssl_context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
+            ssl_context.verify_mode = ssl.CERT_NONE
+            ssl_context.check_hostname = False
+
+        service_instance = None
+
+        connect_args = dict(
+            host=hostname,
+            port=port,
+        )
+        if ssl_context:
+            connect_args.update(sslContext=ssl_context)
+
+        msg_suffix = ''
+        try:
+            if http_proxy_host:
+                msg_suffix = " [proxy: %s:%d]" % (http_proxy_host, http_proxy_port)
+                connect_args.update(httpProxyHost=http_proxy_host, httpProxyPort=http_proxy_port)
+                smart_stub = connect.SmartStubAdapter(**connect_args)
+                session_stub = connect.VimSessionOrientedStub(smart_stub, connect.VimSessionOrientedStub.makeUserLoginMethod(username, password))
+                service_instance = vim.ServiceInstance('ServiceInstance', session_stub)
+            else:
+                connect_args.update(user=username, pwd=password)
+                service_instance = connect.SmartConnect(**connect_args)
+        except vim.fault.InvalidLogin as invalid_login:
+            msg = "Unable to log on to vCenter or ESXi API at %s:%s " % (hostname, port)
+            raise ApiAccessError("%s as %s: %s" % (msg, username, invalid_login.msg) + msg_suffix)
+        except vim.fault.NoPermission as no_permission:
+            raise ApiAccessError("User %s does not have required permission"
+                                 " to log on to vCenter or ESXi API at %s:%s : %s" % (username, hostname, port, no_permission.msg))
+        except (requests.ConnectionError, ssl.SSLError) as generic_req_exc:
+            raise ApiAccessError("Unable to connect to vCenter or ESXi API at %s on TCP/%s: %s" % (hostname, port, generic_req_exc))
+        except vmodl.fault.InvalidRequest as invalid_request:
+            # Request is malformed
+            msg = "Failed to get a response from server %s:%s " % (hostname, port)
+            raise ApiAccessError("%s as request is malformed: %s" % (msg, invalid_request.msg) + msg_suffix)
+        except Exception as generic_exc:
+            msg = "Unknown error while connecting to vCenter or ESXi API at %s:%s" % (hostname, port) + msg_suffix
+            raise ApiAccessError("%s : %s" % (msg, generic_exc))
+
+        if service_instance is None:
+            msg = "Unknown error while connecting to vCenter or ESXi API at %s:%s" % (hostname, port)
+            raise ApiAccessError(msg + msg_suffix)
+
+        atexit.register(connect.Disconnect, service_instance)
+
+        return service_instance, service_instance.RetrieveContent()
diff --git a/plugins/module_utils/vmware.py b/plugins/module_utils/vmware.py
@@ -6,7 +6,6 @@
 # GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
 # SPDX-License-Identifier: GPL-3.0-or-later
 
-from __future__ import absolute_import, division, print_function
 __metaclass__ = type
 
 import atexit
@@ -22,6 +21,7 @@
 import datetime
 from collections import OrderedDict
 from ansible.module_utils.compat.version import StrictVersion
+from ansible_collections.community.vmware.plugins.module_utils.clients._vmware import PyvmomiClient, ApiAccessError
 from random import randint
 
 
@@ -54,11 +54,6 @@ def __init__(self, *args, **kwargs):
         super(TaskError, self).__init__(*args, **kwargs)
 
 
-class ApiAccessError(Exception):
-    def __init__(self, *args, **kwargs):
-        super(ApiAccessError, self).__init__(*args, **kwargs)
-
-
 def check_answer_question_status(vm):
     """Check whether locked a virtual machine.
 
@@ -1067,11 +1062,9 @@ def quote_obj_name(object_name=None):
     return object_name
 
 
-class PyVmomi(object):
+class PyVmomi(PyvmomiClient):
     def __init__(self, module):
-        """
-        Constructor
-        """
+        self.module = module
         if not HAS_REQUESTS:
             module.fail_json(msg=missing_required_lib('requests'),
                              exception=REQUESTS_IMP_ERR)
@@ -1080,10 +1073,16 @@ def __init__(self, module):
             module.fail_json(msg=missing_required_lib('PyVmomi'),
                              exception=PYVMOMI_IMP_ERR)
 
-        self.module = module
+        try:
+            super().__init__(module.params['hostname'], module.params['username'],
+                             module.params['password'], module.params['port'],
+                             module.params['validate_certs'],
+                             module.params['proxy_host'],
+                             module.params['proxy_port'])
+        except ApiAccessError as aae:
+            module.fail_json(msg=str(aae))
         self.params = module.params
         self.current_vm_obj = None
-        self.si, self.content = connect_to_api(self.module, return_si=True)
         self.custom_field_mgr = []
         if self.content.customFieldsManager:  # not an ESXi
             self.custom_field_mgr = self.content.customFieldsManager.field
diff --git a/tests/unit/module_utils/test_vmware.py b/tests/unit/module_utils/test_vmware.py
@@ -25,7 +25,10 @@
             username='Administrator@vsphere.local',
             password='Esxi@123$%',
             hostname=False,
+            port=443,
             validate_certs=False,
+            proxy_host=None,
+            proxy_port=None,
         ),
         "Hostname parameter is missing. Please specify this parameter in task or"
         " export environment variable like 'export VMWARE_HOST=ESXI_HOSTNAME'"
@@ -35,6 +38,7 @@
             username=False,
             password='Esxi@123$%',
             hostname='esxi1',
+            port=443,
             validate_certs=False,
         ),
         "Username parameter is missing. Please specify this parameter in task or"
@@ -45,6 +49,7 @@
             username='Administrator@vsphere.local',
             password=False,
             hostname='esxi1',
+            port=443,
             validate_certs=False,
         ),
         "Password parameter is missing. Please specify this parameter in task or"
@@ -64,6 +69,7 @@
             username='Administrator@vsphere.local',
             password='Esxi@123$%',
             hostname='esxi1',
+            port=443,
             proxy_host='myproxyserver.com',
             proxy_port=80,
             validate_certs=False,
@@ -126,20 +132,6 @@ def test_required_params(request, params, msg, fake_ansible_module):
     # TODO: assert msg in fake_ansible_module.fail_json.call_args[1]['msg']
 
 
-def test_validate_certs(monkeypatch, fake_ansible_module):
-    """ Test if SSL is required or not"""
-    fake_ansible_module.params = test_data[3][0]
-
-    monkeypatch.setattr(vmware_module_utils, 'ssl', mock.Mock())
-    del vmware_module_utils.ssl.SSLContext
-    with pytest.raises(FailJsonException):
-        vmware_module_utils.PyVmomi(fake_ansible_module)
-    msg = 'pyVim does not support changing verification mode with python < 2.7.9.' \
-          ' Either update python or use validate_certs=false.'
-    fake_ansible_module.fail_json.assert_called_once()
-    assert msg == fake_ansible_module.fail_json.call_args[1]['msg']
-
-
 def test_vmdk_disk_path_split(monkeypatch, fake_ansible_module):
     """ Test vmdk_disk_path_split function"""
     fake_ansible_module.params = test_data[0][0]
@@ -162,75 +154,6 @@ def test_vmdk_disk_path_split_negative(monkeypatch, fake_ansible_module):
     assert 'Bad path' in fake_ansible_module.fail_json.call_args[1]['msg']
 
 
-@pytest.mark.skipif(sys.version_info < (2, 7), reason="requires python2.7 and greater")
-def test_connect_to_api_validate_certs(monkeypatch, fake_ansible_module):
-    monkeypatch.setattr(vmware_module_utils, 'connect', mock.Mock())
-
-    def MockSSLContext(proto):
-        ssl_context.proto = proto
-        return ssl_context
-
-    # New Python with SSLContext + validate_certs=True
-    vmware_module_utils.connect.reset_mock()
-    ssl_context = mock.Mock()
-    monkeypatch.setattr(vmware_module_utils.ssl, 'SSLContext', MockSSLContext)
-    fake_ansible_module.params['validate_certs'] = True
-    vmware_module_utils.connect_to_api(fake_ansible_module)
-    assert ssl_context.proto == ssl.PROTOCOL_SSLv23
-    assert ssl_context.verify_mode == ssl.CERT_REQUIRED
-    assert ssl_context.check_hostname is True
-    vmware_module_utils.connect.SmartConnect.assert_called_once_with(
-        host='esxi1',
-        port=443,
-        pwd='Esxi@123$%',
-        user='Administrator@vsphere.local',
-        sslContext=ssl_context)
-
-    # New Python with SSLContext + validate_certs=False
-    vmware_module_utils.connect.reset_mock()
-    ssl_context = mock.Mock()
-    monkeypatch.setattr(vmware_module_utils.ssl, 'SSLContext', MockSSLContext)
-    fake_ansible_module.params['validate_certs'] = False
-    vmware_module_utils.connect_to_api(fake_ansible_module)
-    assert ssl_context.proto == ssl.PROTOCOL_SSLv23
-    assert ssl_context.verify_mode == ssl.CERT_NONE
-    assert ssl_context.check_hostname is False
-    vmware_module_utils.connect.SmartConnect.assert_called_once_with(
-        host='esxi1',
-        port=443,
-        pwd='Esxi@123$%',
-        user='Administrator@vsphere.local',
-        sslContext=ssl_context)
-
-    # Old Python with no SSLContext + validate_certs=True
-    vmware_module_utils.connect.reset_mock()
-    ssl_context = mock.Mock()
-    ssl_context.proto = None
-    monkeypatch.delattr(vmware_module_utils.ssl, 'SSLContext')
-    fake_ansible_module.params['validate_certs'] = True
-    with pytest.raises(FailJsonException):
-        vmware_module_utils.connect_to_api(fake_ansible_module)
-    assert ssl_context.proto is None
-    fake_ansible_module.fail_json.assert_called_once_with(msg=(
-        'pyVim does not support changing verification mode with python '
-        '< 2.7.9. Either update python or use validate_certs=false.'))
-    assert not vmware_module_utils.connect.SmartConnect.called
-
-    # Old Python with no SSLContext + validate_certs=False
-    vmware_module_utils.connect.reset_mock()
-    ssl_context = mock.Mock()
-    ssl_context.proto = None
-    monkeypatch.delattr(vmware_module_utils.ssl, 'SSLContext', raising=False)
-    fake_ansible_module.params['validate_certs'] = False
-    vmware_module_utils.connect_to_api(fake_ansible_module)
-    assert ssl_context.proto is None
-    vmware_module_utils.connect.SmartConnect.assert_called_once_with(
-        host='esxi1',
-        port=443,
-        pwd='Esxi@123$%',
-        user='Administrator@vsphere.local')
-
-
 @pytest.mark.parametrize("test_options, test_current_options, test_truthy_strings_as_bool", [
     ({"data": True}, [], True),
     ({"data": 1}, [], True),
