migrate proxy unit tests to pytest

tests/pytests/unit/proxy/nxos/test_nxos_nxapi.py

@@ -0,0 +1,253 @@
+"""
+    :codeauthor: Mike Wiebe <@mikewiebe>
+"""
+
+# Copyright (c) 2019 Cisco and/or its affiliates.
+#
+# 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.
+
+import pytest
+
+import salt.proxy.nxos as nxos_proxy
+import salt.utils.nxos as nxos_utils
+from tests.support.mock import create_autospec, patch
+from tests.unit.modules.nxos.nxos_grains import n9k_grains
+from tests.unit.modules.nxos.nxos_show_cmd_output import n9k_show_ver_list
+
+
+@pytest.fixture
+def configure_loader_modules():
+    with patch("salt.proxy.nxos.CONNECTION", "nxapi"):
+        yield {nxos_proxy: {}}
+
+
+def test_check_virtual():
+
+    """UT: nxos module:check_virtual method - return value"""
+
+    result = nxos_proxy.__virtual__()
+    assert "nxos" in result
+
+
+def test_init():
+
+    """UT: nxos module:init method - nxapi proxy"""
+
+    with patch.object(nxos_proxy, "__opts__", {"proxy": {"connection": "nxapi"}}):
+        with patch("salt.proxy.nxos._init_nxapi", autospec=True) as init_nxapi:
+            result = nxos_proxy.init()
+            assert result == init_nxapi.return_value
+
+
+def test_init_opts_none():
+
+    """UT: nxos module:init method - __opts__ connection is None"""
+
+    with patch.object(nxos_proxy, "__opts__", {"proxy": {"connection": None}}):
+        with patch("salt.proxy.nxos._init_nxapi", autospec=True) as init_nxapi:
+            result = nxos_proxy.init()
+            assert result == init_nxapi.return_value
+
+
+def test_init_bad_connection_type():
+
+    """UT: nxos module:init method - bad CONNECTION type"""
+    with patch.object(nxos_proxy, "__opts__", {"proxy": {"connection": "unknown"}}):
+        assert not nxos_proxy.init()
+
+
+def test_initialized():
+
+    """UT: nxos module:initialized method - nxapi proxy"""
+
+    with patch(
+        "salt.proxy.nxos._initialized_nxapi", autospec=True
+    ) as initialized_nxapi:
+        result = nxos_proxy.initialized()
+        assert result == initialized_nxapi.return_value
+
+
+def test_ping():
+
+    """UT: nxos module:ping method - nxapi proxy"""
+
+    with patch("salt.proxy.nxos._ping_nxapi", autospec=True) as ping_nxapi:
+        result = nxos_proxy.ping()
+        assert result == ping_nxapi.return_value
+
+
+def test_grains():
+
+    """UT: nxos module:grains method - nxapi grains"""
+
+    with patch(
+        "salt.proxy.nxos.sendline", autospec=True, return_value=n9k_show_ver_list
+    ):
+        result = nxos_proxy.grains()
+        assert result == n9k_grains
+
+
+def test_grains_cache_set():
+
+    """UT: nxos module:grains method - nxapi grains cache set"""
+
+    with patch("salt.proxy.nxos.DEVICE_DETAILS", {"grains_cache": n9k_grains["nxos"]}):
+        with patch(
+            "salt.proxy.nxos.sendline",
+            autospec=True,
+            return_value=n9k_show_ver_list,
+        ):
+            result = nxos_proxy.grains()
+            assert result == n9k_grains
+
+
+def test_grains_refresh():
+
+    """UT: nxos module:grains_refresh method - nxapi grains"""
+
+    device_details = {"grains_cache": None}
+
+    with patch("salt.proxy.nxos.DEVICE_DETAILS", device_details):
+        with patch("salt.proxy.nxos.grains", autospec=True) as grains:
+            result = nxos_proxy.grains_refresh()
+            assert nxos_proxy.DEVICE_DETAILS["grains_cache"] == {}
+            assert result == grains.return_value
+
+
+def test_sendline():
+
+    """UT: nxos module:sendline method - nxapi"""
+
+    command = "show version"
+
+    with patch("salt.proxy.nxos._nxapi_request", autospec=True) as nxapi_request:
+        result = nxos_proxy.sendline(command)
+        assert result == nxapi_request.return_value
+
+
+def test_proxy_config():
+
+    """UT: nxos module:proxy_config method - nxapi success path"""
+
+    commands = ["feature bgp", "router bgp 65535"]
+
+    with patch("salt.proxy.nxos.DEVICE_DETAILS", {"save_config": False}):
+        with patch("salt.proxy.nxos._nxapi_request", autospec=True) as nxapi_request:
+            result = nxos_proxy.proxy_config(commands)
+            assert result == [commands, nxapi_request.return_value]
+
+
+def test_proxy_config_save_config():
+
+    """UT: nxos module:proxy_config method - nxapi success path"""
+
+    commands = ["feature bgp", "router bgp 65535"]
+
+    with patch("salt.proxy.nxos.DEVICE_DETAILS", {"save_config": None}):
+        with patch("salt.proxy.nxos._nxapi_request", autospec=True) as nxapi_request:
+            result = nxos_proxy.proxy_config(commands, save_config=True)
+            assert result == [commands, nxapi_request.return_value]
+
+
+def test__init_nxapi():
+
+    """UT: nxos module:_init_nxapi method - successful connectinon"""
+
+    opts = {"proxy": {"arg1": None}}
+    nxapi_request = create_autospec(nxos_utils.nxapi_request, return_value="data")
+
+    with patch("salt.proxy.nxos.DEVICE_DETAILS", {}) as device_details:
+        with patch("salt.proxy.nxos.__utils__", {"nxos.nxapi_request": nxapi_request}):
+            result = nxos_proxy._init_nxapi(opts)
+
+            assert device_details["initialized"]
+            assert device_details["up"]
+            assert device_details["save_config"]
+            assert result
+
+            nxapi_request.assert_called_with("show clock", **opts["proxy"])
+
+
+def test_bad__init_nxapi():
+    class NXAPIException(Exception):
+        pass
+
+    nxapi_request = create_autospec(
+        nxos_utils.nxapi_request, side_effect=NXAPIException
+    )
+
+    with patch("salt.proxy.nxos.__utils__", {"nxos.nxapi_request": nxapi_request}):
+        with patch("salt.proxy.nxos.log", autospec=True) as log:
+            with pytest.raises(NXAPIException):
+                nxos_proxy._init_nxapi({"proxy": {"host": "HOST"}})
+            log.error.assert_called()
+
+
+def test__initialized_nxapi():
+
+    """UT: nxos module:_initialized_nxapi method"""
+
+    result = nxos_proxy._initialized_nxapi()
+    assert not result
+
+    with patch("salt.proxy.nxos.DEVICE_DETAILS", {"initialized": True}):
+        result = nxos_proxy._initialized_nxapi()
+        assert result
+
+
+def test__ping_nxapi():
+
+    """UT: nxos module:_ping_nxapi method"""
+
+    result = nxos_proxy._ping_nxapi()
+    assert not result
+
+    with patch("salt.proxy.nxos.DEVICE_DETAILS", {"up": True}):
+        result = nxos_proxy._ping_nxapi()
+        assert result
+
+
+def test__shutdown_nxapi():
+
+    """UT: nxos module:_shutdown_nxapi method"""
+
+    opts = {"id": "value"}
+
+    with patch("salt.proxy.nxos.log", autospec=True):
+        nxos_proxy._shutdown_nxapi()
+        # nothing to test
+
+
+def test__nxapi_request_ssh_return():
+
+    """UT: nxos module:_nxapi_request method - CONNECTION == 'ssh'"""
+
+    commands = "show version"
+
+    with patch("salt.proxy.nxos.CONNECTION", "ssh"):
+        result = nxos_proxy._nxapi_request(commands)
+        assert "_nxapi_request is not available for ssh proxy" == result
+
+
+def test__nxapi_request_connect():
+
+    """UT: nxos module:_nxapi_request method"""
+
+    commands = "show version"
+    nxapi_request = create_autospec(nxos_utils.nxapi_request, return_value="data")
+
+    with patch("salt.proxy.nxos.DEVICE_DETAILS", {"conn_args": {"arg1": None}}):
+        with patch("salt.proxy.nxos.__utils__", {"nxos.nxapi_request": nxapi_request}):
+            result = nxos_proxy._nxapi_request(commands)
+            assert "data" == result
+            nxapi_request.assert_called_with(commands, method="cli_conf", arg1=None)

tests/pytests/unit/proxy/nxos/test_nxos_ssh.py

@@ -0,0 +1,260 @@
+"""
+    :codeauthor: Mike Wiebe <@mikewiebe>
+"""
+
+# Copyright (c) 2019 Cisco and/or its affiliates.
+#
+# 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.
+
+import pytest
+
+import salt.proxy.nxos as nxos_proxy
+from salt.exceptions import CommandExecutionError
+from tests.support.mock import MagicMock, patch
+from tests.unit.modules.nxos.nxos_grains import n9k_grains
+from tests.unit.modules.nxos.nxos_show_cmd_output import n9k_show_ver_list
+
+
+@pytest.fixture
+def configure_loader_modules():
+    with patch("salt.proxy.nxos.CONNECTION", "ssh"):
+        return {
+            nxos_proxy: {
+                "__opts__": {
+                    "proxy": {
+                        "host": "dt-n9k5-1.cisco.com",
+                        "username": "admin",
+                        "password": "password",
+                    }
+                },
+            }
+        }
+
+
+def test_init():
+
+    """UT: nxos module:init method - ssh proxy"""
+
+    with patch("salt.proxy.nxos._init_ssh", autospec=True) as init_ssh:
+        result = nxos_proxy.init()
+        assert result == init_ssh.return_value
+
+
+def test_init_opts_none():
+
+    """UT: nxos module:init method - __opts__ connection is None"""
+
+    with patch("salt.proxy.nxos.__opts__", {"proxy": {"connection": None}}):
+        with patch("salt.proxy.nxos._init_ssh", autospec=True) as init_ssh:
+            result = nxos_proxy.init()
+            assert result == init_ssh.return_value
+
+
+def test_initialized():
+
+    """UT: nxos module:initialized method - ssh proxy"""
+
+    with patch("salt.proxy.nxos._initialized_ssh", autospec=True) as initialized_ssh:
+        result = nxos_proxy.initialized()
+        assert result == initialized_ssh.return_value
+
+
+def test_ping():
+
+    """UT: nxos module:ping method - ssh proxy"""
+
+    with patch("salt.proxy.nxos._ping_ssh", autospec=True) as ping_ssh:
+        result = nxos_proxy.ping()
+        assert result == ping_ssh.return_value
+
+
+def test_grains():
+
+    """UT: nxos module:grains method - ssh grains"""
+
+    with patch(
+        "salt.proxy.nxos.sendline", autospec=True, return_value=n9k_show_ver_list[0]
+    ):
+        result = nxos_proxy.grains()
+        assert result == n9k_grains
+
+
+def test_sendline():
+
+    """UT: nxos module:sendline method - nxapi"""
+
+    command = "show version"
+
+    with patch("salt.proxy.nxos._sendline_ssh", autospec=True) as sendline_ssh:
+        result = nxos_proxy.sendline(command)
+        assert result == sendline_ssh.return_value
+
+
+def test_proxy_config():
+
+    """UT: nxos module:proxy_config method - ssh success path"""
+
+    commands = ["feature bgp", "router bgp 65535"]
+
+    with patch("salt.proxy.nxos.DEVICE_DETAILS", {"save_config": False}):
+        with patch("salt.proxy.nxos._sendline_ssh", autospec=True) as sendline_ssh:
+            result = nxos_proxy.proxy_config(commands)
+            assert result == [commands, sendline_ssh.return_value]
+
+
+def test_proxy_config_save_config():
+
+    """UT: nxos module:proxy_config method - ssh success path"""
+
+    commands = ["feature bgp", "router bgp 65535"]
+
+    with patch("salt.proxy.nxos.DEVICE_DETAILS", {"save_config": None}):
+        with patch("salt.proxy.nxos._sendline_ssh", autospec=True) as sendline_ssh:
+            result = nxos_proxy.proxy_config(commands, save_config=True)
+            assert result == [commands, sendline_ssh.return_value]
+
+
+def test_proxy_config_error():
+
+    """UT: nxos module:proxy_config method - CommandExecutionError"""
+
+    with patch(
+        "salt.proxy.nxos._sendline_ssh",
+        autospec=True,
+        side_effect=CommandExecutionError,
+    ):
+        with pytest.raises(CommandExecutionError):
+            nxos_proxy.proxy_config("show version", save_config=True)
+
+
+def test__init_ssh_device_details():
+    with patch("salt.proxy.nxos.SSHConnection", autospec=True) as SSHConnection:
+        SSHConnection().sendline.return_value = ["", ""]
+
+        with patch("salt.proxy.nxos.DEVICE_DETAILS", {}) as device_details:
+            nxos_proxy._init_ssh(None)
+            assert nxos_proxy._worker_name() in device_details
+            assert device_details["initialized"]
+            assert device_details["save_config"]
+
+        with patch.dict(nxos_proxy.__opts__["proxy"], {"save_config": False}):
+            with patch("salt.proxy.nxos.DEVICE_DETAILS", {}) as device_details:
+                nxos_proxy._init_ssh(None)
+                assert nxos_proxy._worker_name() in device_details
+                assert device_details["initialized"]
+                assert not device_details["save_config"]
+
+
+def test__init_ssh_opts():
+
+    """UT: nxos module:_init_ssh method - successful connectinon"""
+
+    with patch("salt.proxy.nxos.DEVICE_DETAILS", {}):
+        with patch("salt.proxy.nxos.SSHConnection", autospec=True) as SSHConnection:
+            SSHConnection().sendline.return_value = ["", ""]
+            nxos_proxy._init_ssh(None)
+            assert (
+                nxos_proxy.__opts__["proxy"]["host"]
+                == SSHConnection.call_args[1]["host"]
+            )
+
+            opts = MagicMock()
+            nxos_proxy._init_ssh(opts)
+            assert opts["proxy"]["host"] == SSHConnection.call_args[1]["host"]
+
+
+def test__init_ssh_prompt():
+
+    """UT: nxos module:_init_ssh method - prompt regex"""
+
+    with patch("salt.proxy.nxos.DEVICE_DETAILS", {}):
+        with patch("salt.proxy.nxos.SSHConnection", autospec=True) as SSHConnection:
+            SSHConnection().sendline.return_value = ["", ""]
+
+            with patch.dict(
+                nxos_proxy.__opts__["proxy"], {"prompt_regex": "n9k.*device"}
+            ):
+                nxos_proxy._init_ssh(None)
+                assert "n9k.*device" == SSHConnection.call_args[1]["prompt"]
+
+            with patch.dict(
+                nxos_proxy.__opts__["proxy"], {"prompt_name": "n9k-device"}
+            ):
+                nxos_proxy._init_ssh(None)
+                assert "n9k-device.*#" == SSHConnection.call_args[1]["prompt"]
+
+            nxos_proxy._init_ssh(None)
+            assert ".+#$" == SSHConnection.call_args[1]["prompt"]
+
+
+def test__initialized_ssh():
+
+    """UT: nxos module:_initialized_ssh method"""
+
+    with patch("salt.proxy.nxos.DEVICE_DETAILS", {"initialized": True}):
+        result = nxos_proxy._initialized_ssh()
+        assert result
+
+    with patch("salt.proxy.nxos.DEVICE_DETAILS", {}):
+        result = nxos_proxy._initialized_ssh()
+        assert not result
+
+
+def test__parse_output_for_errors():
+
+    """UT: nxos module:_parse_output_for_errors method"""
+
+    data = "% Incomplete command at '^' marker."
+    command = "show"
+
+    with pytest.raises(CommandExecutionError):
+        nxos_proxy._parse_output_for_errors(data, command, error_pattern="Incomplete")
+
+    data = "% Incomplete command at '^' marker."
+    command = "show"
+
+    with pytest.raises(CommandExecutionError):
+        nxos_proxy._parse_output_for_errors(
+            data, command, error_pattern=["Incomplete", "marker"]
+        )
+
+    data = "% Invalid command at '^' marker."
+    command = "show bep"
+
+    with pytest.raises(CommandExecutionError):
+        nxos_proxy._parse_output_for_errors(data, command)
+
+    data = "% Incomplete command at '^' marker."
+    command = "show"
+
+    nxos_proxy._parse_output_for_errors(data, command)
+
+    data = "% Incomplete command at '^' marker."
+    command = "show"
+
+    nxos_proxy._parse_output_for_errors(data, command, error_pattern="foo")
+
+
+def test__init_ssh_raise_exception():
+
+    """UT: nxos module:_init_ssh method - raise exception"""
+
+    class SSHException(Exception):
+        pass
+
+    with patch("salt.proxy.nxos.SSHConnection", autospec=True) as SSHConnection:
+        with patch("salt.proxy.nxos.log", autospec=True) as log:
+            with pytest.raises(SSHException):
+                SSHConnection.side_effect = SSHException
+                nxos_proxy._init_ssh(None)
+            log.error.assert_called()

tests/pytests/unit/proxy/test_cimc.py

@@ -0,0 +1,331 @@
+"""
+tests.unit.proxy.test_cimc
+~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Unit tests for the cimc proxy module
+"""
+
+import logging
+
+import pytest
+
+import salt.exceptions
+import salt.proxy.cimc as cimc
+from tests.support.mock import MagicMock, patch
+
+log = logging.getLogger(__name__)
+
+
+def http_query_response(*args, data=None, **kwargs):
+    log.debug(
+        "http_query_response side_effect; ARGS: %s // KWARGS: %s // DATA: %s",
+        args,
+        kwargs,
+        data,
+    )
+    login_response = """\
+    <aaaLogin
+        response="yes"
+        outCookie="real-cookie"
+        outRefreshPeriod="600"
+        outPriv="admin">
+    </aaaLogin>"""
+    logout_response = """\
+    <aaaLogout
+        cookie="real-cookie"
+        response="yes"
+        outStatus="success">
+    </aaaLogout>
+    """
+    config_resolve_class_response = """\
+    <configResolveClass
+        cookie="real-cookie"
+        response="yes"
+        classId="computeRackUnit">
+        <outConfig>
+            <computeRackUnit
+                dn="sys/rack-unit-1"
+                adminPower="policy"
+                availableMemory="16384"
+                model="R210-2121605W"
+                memorySpeed="1067"
+                name="UCS C210 M2"
+                numOfAdaptors="2"
+                numOfCores="8"
+                numOfCoresEnabled="8"
+                numOfCpus="2"
+                numOfEthHostIfs="5"
+                numOfFcHostIfs="2"
+                numOfThreads="16"
+                operPower="on"
+                originalUuid="00C9DE3C-370D-DF11-1186-6DD1393A608B"
+                presence="equipped"
+                serverID="1"
+                serial="QCI140205Z2"
+                totalMemory="16384"
+                usrLbl="C210 Row-B Rack-10"
+                uuid="00C9DE3C-370D-DF11-1186-6DD1393A608B"
+                vendor="Cisco Systems Inc" >
+            </computeRackUnit>
+        </outConfig>
+    </configResolveClass>
+    """
+    config_con_mo_response = """\
+    <configConfMo
+        dn="sys/rack-unit-1/locator-led"
+        cookie="real-cookie"
+        response="yes">
+        <outConfig>
+            <equipmentLocatorLed
+                dn="sys/rack-unit-1/locator-led"
+                adminState="inactive"
+                color="unknown"
+                id="1"
+                name=""
+                operState="off">
+            </equipmentLocatorLed>
+        </outConfig>
+    </configConfMo>
+    """
+
+    if data.startswith("<aaaLogin"):
+        response = login_response
+    elif data.startswith("<aaaLogout"):
+        response = logout_response
+    elif data.startswith("<configResolveClass"):
+        response = config_resolve_class_response
+    elif data.startswith("<configConfMo"):
+        response = config_con_mo_response
+    else:
+        response = ""
+    return {"text": response, "status": 200}
+
+
+@pytest.fixture
+def configure_loader_modules():
+    with patch.dict(cimc.DETAILS):
+        yield {cimc: {"__pillar__": {}}}
+
+
+@pytest.fixture(params=[None, True, False])
+def verify_ssl(request):
+    return request.param
+
+
+@pytest.fixture
+def opts(verify_ssl):
+    return {
+        "proxy": {
+            "host": "TheHost",
+            "username": "TheUsername",
+            "password": "ThePassword",
+            "verify_ssl": verify_ssl,
+        }
+    }
+
+
+def _get_expected_verify_ssl(verify_ssl):
+    expected = True if verify_ssl is None else verify_ssl
+
+    log.debug(
+        "verify_ssl: %s // expected verify_ssl: %s",
+        verify_ssl,
+        expected,
+    )
+
+    return expected
+
+
+def test_init():
+    # No host, returns False
+    opts = {"proxy": {"username": "xxxx", "password": "xxx"}}
+    ret = cimc.init(opts)
+    assert not ret
+
+    # No username , returns False
+    opts = {"proxy": {"password": "xxx", "host": "cimc"}}
+    ret = cimc.init(opts)
+    assert not ret
+
+    # No password, returns False
+    opts = {"proxy": {"username": "xxxx", "host": "cimc"}}
+    ret = cimc.init(opts)
+    assert not ret
+
+    opts = {"proxy": {"username": "xxxx", "password": "xxx", "host": "cimc"}}
+    with patch.object(cimc, "logon", return_value="9zVG5U8DFZNsTR"):
+        with patch.object(cimc, "get_config_resolver_class", return_value="True"):
+            ret = cimc.init(opts)
+            assert cimc.DETAILS["url"] == "https://cimc/nuova"
+            assert cimc.DETAILS["username"] == "xxxx"
+            assert cimc.DETAILS["password"] == "xxx"
+            assert cimc.DETAILS["initialized"]
+
+
+def test__validate_response_code():
+    with pytest.raises(
+        salt.exceptions.CommandExecutionError,
+        match="Did not receive a valid response from host.",
+    ):
+        cimc._validate_response_code("404")
+
+    with patch.object(cimc, "logout", return_value=True) as mock_logout:
+        with pytest.raises(
+            salt.exceptions.CommandExecutionError,
+            match="Did not receive a valid response from host.",
+        ):
+            cimc._validate_response_code("404", "9zVG5U8DFZNsTR")
+            mock_logout.assert_called_once_with("9zVG5U8DFZNsTR")
+
+
+def test_init_with_ssl(verify_ssl, opts):
+    http_query_mock = MagicMock(side_effect=http_query_response)
+    expected_verify_ssl_value = _get_expected_verify_ssl(verify_ssl)
+
+    with patch.dict(cimc.__utils__, {"http.query": http_query_mock}):
+        cimc.init(opts)
+
+    for idx, call in enumerate(http_query_mock.mock_calls, 1):
+        condition = call.kwargs["verify_ssl"] is expected_verify_ssl_value
+        condition_error = "{} != {}; Call(number={}): {}".format(
+            idx, call, call.kwargs["verify_ssl"], expected_verify_ssl_value
+        )
+        assert condition, condition_error
+
+
+def test_logon(opts, verify_ssl):
+    http_query_mock = MagicMock(side_effect=http_query_response)
+    expected_verify_ssl_value = _get_expected_verify_ssl(verify_ssl)
+
+    # Let's init the proxy and ignore it's actions, this test is not about them
+    with patch(
+        "salt.proxy.cimc.get_config_resolver_class",
+        MagicMock(return_value=True),
+    ):
+        cimc.init(opts)
+
+    with patch.dict(cimc.__utils__, {"http.query": http_query_mock}):
+        cimc.logon()
+
+    for idx, call in enumerate(http_query_mock.mock_calls, 1):
+        condition = call.kwargs["verify_ssl"] is expected_verify_ssl_value
+        condition_error = "{} != {}; Call(number={}): {}".format(
+            idx, call, call.kwargs["verify_ssl"], expected_verify_ssl_value
+        )
+        assert condition, condition_error
+
+
+def test_logout(opts, verify_ssl):
+    http_query_mock = MagicMock(side_effect=http_query_response)
+    expected_verify_ssl_value = _get_expected_verify_ssl(verify_ssl)
+
+    # Let's init the proxy and ignore it's actions, this test is not about them
+    with patch(
+        "salt.proxy.cimc.get_config_resolver_class",
+        MagicMock(return_value=True),
+    ):
+        cimc.init(opts)
+
+    with patch.dict(cimc.__utils__, {"http.query": http_query_mock}):
+        cimc.logout()
+
+    for idx, call in enumerate(http_query_mock.mock_calls, 1):
+        condition = call.kwargs["verify_ssl"] is expected_verify_ssl_value
+        condition_error = "{} != {}; Call(number={}): {}".format(
+            idx, call, call.kwargs["verify_ssl"], expected_verify_ssl_value
+        )
+        assert condition, condition_error
+
+
+def test_grains(opts, verify_ssl):
+    http_query_mock = MagicMock(side_effect=http_query_response)
+    expected_verify_ssl_value = _get_expected_verify_ssl(verify_ssl)
+
+    # Let's init the proxy and ignore it's actions, this test is not about them
+    with patch(
+        "salt.proxy.cimc.get_config_resolver_class",
+        MagicMock(return_value=True),
+    ):
+        cimc.init(opts)
+
+    with patch.dict(cimc.__utils__, {"http.query": http_query_mock}):
+        cimc.grains()
+
+    for idx, call in enumerate(http_query_mock.mock_calls, 1):
+        condition = call.kwargs["verify_ssl"] is expected_verify_ssl_value
+        condition_error = "{} != {}; Call(number={}): {}".format(
+            idx, call, call.kwargs["verify_ssl"], expected_verify_ssl_value
+        )
+        assert condition, condition_error
+
+
+def test_grains_refresh(opts, verify_ssl):
+    http_query_mock = MagicMock(side_effect=http_query_response)
+    expected_verify_ssl_value = _get_expected_verify_ssl(verify_ssl)
+
+    # Let's init the proxy and ignore it's actions, this test is not about them
+    with patch(
+        "salt.proxy.cimc.get_config_resolver_class",
+        MagicMock(return_value=True),
+    ):
+        cimc.init(opts)
+
+    with patch.dict(cimc.__utils__, {"http.query": http_query_mock}):
+        cimc.grains_refresh()
+
+    for idx, call in enumerate(http_query_mock.mock_calls, 1):
+        condition = call.kwargs["verify_ssl"] is expected_verify_ssl_value
+        condition_error = "{} != {}; Call(number={}): {}".format(
+            idx, call, call.kwargs["verify_ssl"], expected_verify_ssl_value
+        )
+        assert condition, condition_error
+
+
+def test_ping(opts, verify_ssl):
+    http_query_mock = MagicMock(side_effect=http_query_response)
+    expected_verify_ssl_value = _get_expected_verify_ssl(verify_ssl)
+
+    # Let's init the proxy and ignore it's actions, this test is not about them
+    with patch(
+        "salt.proxy.cimc.get_config_resolver_class",
+        MagicMock(return_value=True),
+    ):
+        cimc.init(opts)
+
+    with patch.dict(cimc.__utils__, {"http.query": http_query_mock}):
+        cimc.ping()
+
+    for idx, call in enumerate(http_query_mock.mock_calls, 1):
+        condition = call.kwargs["verify_ssl"] is expected_verify_ssl_value
+        condition_error = "{} != {}; Call(number={}): {}".format(
+            idx, call, call.kwargs["verify_ssl"], expected_verify_ssl_value
+        )
+        assert condition, condition_error
+
+
+def test_set_config_modify(opts, verify_ssl):
+    http_query_mock = MagicMock(side_effect=http_query_response)
+    expected_verify_ssl_value = _get_expected_verify_ssl(verify_ssl)
+
+    # Let's init the proxy and ignore it's actions, this test is not about them
+    with patch(
+        "salt.proxy.cimc.get_config_resolver_class",
+        MagicMock(return_value=True),
+    ):
+        cimc.init(opts)
+
+    with patch.dict(cimc.__utils__, {"http.query": http_query_mock}):
+        cimc.set_config_modify(
+            dn="sys/rack-unit-1/locator-led",
+            inconfig=(
+                "<inConfig><equipmentLocatorLed adminState='on'"
+                " dn='sys/rack-unit-1/locator-led'></equipmentLocatorLed></inConfig>"
+            ),
+        )
+
+    for idx, call in enumerate(http_query_mock.mock_calls, 1):
+        condition = call.kwargs["verify_ssl"] is expected_verify_ssl_value
+        condition_error = "{} != {}; Call(number={}): {}".format(
+            idx, call, call.kwargs["verify_ssl"], expected_verify_ssl_value
+        )
+        assert condition, condition_error

tests/pytests/unit/proxy/test_esxcluster.py

@@ -0,0 +1,190 @@
+"""
+    :codeauthor: :email:`Alexandru Bleotu <alexandru.bleotu@morganstanley.com>`
+
+    Tests for esxcluster proxy
+"""
+
+import pytest
+
+import salt.exceptions
+import salt.proxy.esxcluster as esxcluster
+from salt.config.schemas.esxcluster import EsxclusterProxySchema
+from tests.support.mock import MagicMock, patch
+
+try:
+    import jsonschema
+
+    HAS_JSONSCHEMA = True
+except ImportError:
+    HAS_JSONSCHEMA = False
+
+
+pytestmark = [pytest.mark.skipif(not HAS_JSONSCHEMA, reason="jsonschema is required")]
+
+
+@pytest.fixture
+def opts_userpass():
+    return {
+        "proxy": {
+            "proxytype": "esxcluster",
+            "vcenter": "fake_vcenter",
+            "datacenter": "fake_dc",
+            "cluster": "fake_cluster",
+            "mechanism": "userpass",
+            "username": "fake_username",
+            "passwords": ["fake_password"],
+            "protocol": "fake_protocol",
+            "port": 100,
+        }
+    }
+
+
+@pytest.fixture
+def opts_sspi():
+    return {
+        "proxy": {
+            "proxytype": "esxcluster",
+            "vcenter": "fake_vcenter",
+            "datacenter": "fake_dc",
+            "cluster": "fake_cluster",
+            "mechanism": "sspi",
+            "domain": "fake_domain",
+            "principal": "fake_principal",
+            "protocol": "fake_protocol",
+            "port": 100,
+        }
+    }
+
+
+@pytest.fixture
+def configure_loader_modules(opts_sspi):
+    with patch.dict(esxcluster.DETAILS):
+        with patch(
+            "salt.proxy.esxcluster.merge",
+            MagicMock(return_value=opts_sspi["proxy"]),
+        ):
+            yield {esxcluster: {"__pillar__": {}}}
+
+
+def test_merge(opts_sspi):
+    mock_pillar_proxy = MagicMock()
+    mock_opts_proxy = MagicMock()
+    mock_merge = MagicMock(return_value=opts_sspi["proxy"])
+    with patch.dict(esxcluster.__pillar__, {"proxy": mock_pillar_proxy}):
+        with patch("salt.proxy.esxcluster.merge", mock_merge):
+            esxcluster.init(opts={"proxy": mock_opts_proxy})
+    mock_merge.assert_called_once_with(mock_opts_proxy, mock_pillar_proxy)
+
+
+def test_esxcluster_schema(opts_sspi):
+    mock_json_validate = MagicMock()
+    serialized_schema = EsxclusterProxySchema().serialize()
+    with patch("salt.proxy.esxcluster.jsonschema.validate", mock_json_validate):
+        esxcluster.init(opts_sspi)
+    mock_json_validate.assert_called_once_with(opts_sspi["proxy"], serialized_schema)
+
+
+def test_invalid_proxy_input_error(opts_userpass):
+    with patch(
+        "salt.proxy.esxcluster.jsonschema.validate",
+        MagicMock(
+            side_effect=jsonschema.exceptions.ValidationError("Validation Error")
+        ),
+    ):
+        with pytest.raises(salt.exceptions.InvalidConfigError) as excinfo:
+            esxcluster.init(opts_userpass)
+    assert excinfo.value.message == "Validation Error"
+
+
+def test_no_username(opts_userpass):
+    opts = opts_userpass.copy()
+    del opts["proxy"]["username"]
+    with patch("salt.proxy.esxcluster.merge", MagicMock(return_value=opts["proxy"])):
+        with pytest.raises(salt.exceptions.InvalidConfigError) as excinfo:
+            esxcluster.init(opts)
+    assert (
+        excinfo.value.message == "Mechanism is set to 'userpass', but no "
+        "'username' key found in proxy config."
+    )
+
+
+def test_no_passwords(opts_userpass):
+    opts = opts_userpass.copy()
+    del opts["proxy"]["passwords"]
+    with patch("salt.proxy.esxcluster.merge", MagicMock(return_value=opts["proxy"])):
+        with pytest.raises(salt.exceptions.InvalidConfigError) as excinfo:
+            esxcluster.init(opts)
+    assert (
+        excinfo.value.message == "Mechanism is set to 'userpass', but no "
+        "'passwords' key found in proxy config."
+    )
+
+
+def test_no_domain(opts_sspi):
+    opts = opts_sspi.copy()
+    del opts["proxy"]["domain"]
+    with patch("salt.proxy.esxcluster.merge", MagicMock(return_value=opts["proxy"])):
+        with pytest.raises(salt.exceptions.InvalidConfigError) as excinfo:
+            esxcluster.init(opts)
+    assert (
+        excinfo.value.message
+        == "Mechanism is set to 'sspi', but no 'domain' key found in proxy config."
+    )
+
+
+def test_no_principal(opts_sspi):
+    opts = opts_sspi.copy()
+    del opts["proxy"]["principal"]
+    with patch("salt.proxy.esxcluster.merge", MagicMock(return_value=opts["proxy"])):
+        with pytest.raises(salt.exceptions.InvalidConfigError) as excinfo:
+            esxcluster.init(opts)
+    assert (
+        excinfo.value.message
+        == "Mechanism is set to 'sspi', but no 'principal' key found in proxy config."
+    )
+
+
+def test_find_credentials(opts_userpass):
+    mock_find_credentials = MagicMock(return_value=("fake_username", "fake_password"))
+    with patch(
+        "salt.proxy.esxcluster.merge",
+        MagicMock(return_value=opts_userpass["proxy"]),
+    ):
+        with patch("salt.proxy.esxcluster.find_credentials", mock_find_credentials):
+            esxcluster.init(opts_userpass)
+    mock_find_credentials.assert_called_once_with()
+
+
+def test_details_userpass(opts_userpass):
+    mock_find_credentials = MagicMock(return_value=("fake_username", "fake_password"))
+    with patch(
+        "salt.proxy.esxcluster.merge",
+        MagicMock(return_value=opts_userpass["proxy"]),
+    ):
+        with patch("salt.proxy.esxcluster.find_credentials", mock_find_credentials):
+            esxcluster.init(opts_userpass)
+    assert esxcluster.DETAILS == {
+        "vcenter": "fake_vcenter",
+        "datacenter": "fake_dc",
+        "cluster": "fake_cluster",
+        "mechanism": "userpass",
+        "username": "fake_username",
+        "password": "fake_password",
+        "passwords": ["fake_password"],
+        "protocol": "fake_protocol",
+        "port": 100,
+    }
+
+
+def test_details_sspi(opts_sspi):
+    esxcluster.init(opts_sspi)
+    assert esxcluster.DETAILS == {
+        "vcenter": "fake_vcenter",
+        "datacenter": "fake_dc",
+        "cluster": "fake_cluster",
+        "mechanism": "sspi",
+        "domain": "fake_domain",
+        "principal": "fake_principal",
+        "protocol": "fake_protocol",
+        "port": 100,
+    }

tests/pytests/unit/proxy/test_esxdatacenter.py

@@ -0,0 +1,185 @@
+"""
+    :codeauthor: :email:`Alexandru Bleotu <alexandru.bleotu@morganstanley.com>`
+
+    Tests for esxdatacenter proxy
+"""
+import pytest
+
+import salt.exceptions
+import salt.proxy.esxdatacenter as esxdatacenter
+from salt.config.schemas.esxdatacenter import EsxdatacenterProxySchema
+from tests.support.mock import MagicMock, patch
+
+try:
+    import jsonschema
+
+    HAS_JSONSCHEMA = True
+except ImportError:
+    HAS_JSONSCHEMA = False
+
+
+pytestmark = [pytest.mark.skipif(not HAS_JSONSCHEMA, reason="jsonschema is required")]
+
+
+@pytest.fixture
+def opts_userpass():
+    return {
+        "proxy": {
+            "proxytype": "esxdatacenter",
+            "vcenter": "fake_vcenter",
+            "datacenter": "fake_dc",
+            "mechanism": "userpass",
+            "username": "fake_username",
+            "passwords": ["fake_password"],
+            "protocol": "fake_protocol",
+            "port": 100,
+        }
+    }
+
+
+@pytest.fixture
+def opts_sspi():
+    return {
+        "proxy": {
+            "proxytype": "esxdatacenter",
+            "vcenter": "fake_vcenter",
+            "datacenter": "fake_dc",
+            "mechanism": "sspi",
+            "domain": "fake_domain",
+            "principal": "fake_principal",
+            "protocol": "fake_protocol",
+            "port": 100,
+        }
+    }
+
+
+@pytest.fixture
+def configure_loader_modules(opts_sspi):
+    with patch.dict(esxdatacenter.DETAILS):
+        with patch(
+            "salt.proxy.esxdatacenter.merge",
+            MagicMock(return_value=opts_sspi["proxy"]),
+        ):
+            yield {esxdatacenter: {"__pillar__": {}}}
+
+
+def test_merge(opts_sspi):
+    mock_pillar_proxy = MagicMock()
+    mock_opts_proxy = MagicMock()
+    mock_merge = MagicMock(return_value=opts_sspi["proxy"])
+    with patch.dict(esxdatacenter.__pillar__, {"proxy": mock_pillar_proxy}):
+        with patch("salt.proxy.esxdatacenter.merge", mock_merge):
+            esxdatacenter.init(opts={"proxy": mock_opts_proxy})
+    mock_merge.assert_called_once_with(mock_opts_proxy, mock_pillar_proxy)
+
+
+def test_esxdatacenter_schema(opts_sspi):
+    mock_json_validate = MagicMock()
+    serialized_schema = EsxdatacenterProxySchema().serialize()
+    with patch("salt.proxy.esxdatacenter.jsonschema.validate", mock_json_validate):
+        esxdatacenter.init(opts_sspi)
+    mock_json_validate.assert_called_once_with(opts_sspi["proxy"], serialized_schema)
+
+
+def test_invalid_proxy_input_error(opts_userpass):
+    with patch(
+        "salt.proxy.esxdatacenter.jsonschema.validate",
+        MagicMock(
+            side_effect=jsonschema.exceptions.ValidationError("Validation Error")
+        ),
+    ):
+        with pytest.raises(salt.exceptions.InvalidConfigError) as excinfo:
+            esxdatacenter.init(opts_userpass)
+    assert excinfo.value.message == "Validation Error"
+
+
+def test_no_username(opts_userpass):
+    opts = opts_userpass.copy()
+    del opts["proxy"]["username"]
+    with patch("salt.proxy.esxdatacenter.merge", MagicMock(return_value=opts["proxy"])):
+        with pytest.raises(salt.exceptions.InvalidConfigError) as excinfo:
+            esxdatacenter.init(opts)
+    assert (
+        excinfo.value.message == "Mechanism is set to 'userpass', but no "
+        "'username' key found in proxy config."
+    )
+
+
+def test_no_passwords(opts_userpass):
+    opts = opts_userpass.copy()
+    del opts["proxy"]["passwords"]
+    with patch("salt.proxy.esxdatacenter.merge", MagicMock(return_value=opts["proxy"])):
+        with pytest.raises(salt.exceptions.InvalidConfigError) as excinfo:
+            esxdatacenter.init(opts)
+    assert (
+        excinfo.value.message == "Mechanism is set to 'userpass', but no "
+        "'passwords' key found in proxy config."
+    )
+
+
+def test_no_domain(opts_sspi):
+    opts = opts_sspi.copy()
+    del opts["proxy"]["domain"]
+    with patch("salt.proxy.esxdatacenter.merge", MagicMock(return_value=opts["proxy"])):
+        with pytest.raises(salt.exceptions.InvalidConfigError) as excinfo:
+            esxdatacenter.init(opts)
+    assert (
+        excinfo.value.message
+        == "Mechanism is set to 'sspi', but no 'domain' key found in proxy config."
+    )
+
+
+def test_no_principal(opts_sspi):
+    opts = opts_sspi.copy()
+    del opts["proxy"]["principal"]
+    with patch("salt.proxy.esxdatacenter.merge", MagicMock(return_value=opts["proxy"])):
+        with pytest.raises(salt.exceptions.InvalidConfigError) as excinfo:
+            esxdatacenter.init(opts)
+    assert (
+        excinfo.value.message
+        == "Mechanism is set to 'sspi', but no 'principal' key found in proxy config."
+    )
+
+
+def test_find_credentials(opts_userpass):
+    mock_find_credentials = MagicMock(return_value=("fake_username", "fake_password"))
+    with patch(
+        "salt.proxy.esxdatacenter.merge",
+        MagicMock(return_value=opts_userpass["proxy"]),
+    ):
+        with patch("salt.proxy.esxdatacenter.find_credentials", mock_find_credentials):
+            esxdatacenter.init(opts_userpass)
+    mock_find_credentials.assert_called_once_with()
+
+
+def test_details_userpass(opts_userpass):
+    mock_find_credentials = MagicMock(return_value=("fake_username", "fake_password"))
+    with patch(
+        "salt.proxy.esxdatacenter.merge",
+        MagicMock(return_value=opts_userpass["proxy"]),
+    ):
+        with patch("salt.proxy.esxdatacenter.find_credentials", mock_find_credentials):
+            esxdatacenter.init(opts_userpass)
+    assert esxdatacenter.DETAILS == {
+        "vcenter": "fake_vcenter",
+        "datacenter": "fake_dc",
+        "mechanism": "userpass",
+        "username": "fake_username",
+        "password": "fake_password",
+        "passwords": ["fake_password"],
+        "protocol": "fake_protocol",
+        "port": 100,
+    }
+
+
+def test_details_sspi(opts_sspi):
+    esxdatacenter.init(opts_sspi)
+    assert esxdatacenter.DETAILS == {
+        "vcenter": "fake_vcenter",
+        "datacenter": "fake_dc",
+        "mechanism": "sspi",
+        "domain": "fake_domain",
+        "principal": "fake_principal",
+        "protocol": "fake_protocol",
+        "port": 100,
+    }

tests/pytests/unit/proxy/test_junos.py

@@ -0,0 +1,72 @@
+import io
+
+import pytest
+
+import salt.proxy.junos as junos
+from tests.support.mock import ANY, patch
+
+try:
+    import jxmlease  # pylint: disable=unused-import
+    from jnpr.junos.device import Device  # pylint: disable=unused-import
+    from jnpr.junos.exception import ConnectError
+
+    HAS_JUNOS = True
+except ImportError:
+    HAS_JUNOS = False
+
+
+pytestmark = [
+    pytest.mark.skipif(
+        not HAS_JUNOS, reason="The junos-eznc and jxmlease modules are required"
+    ),
+]
+
+
+@pytest.fixture
+def opts():
+    return {
+        "proxy": {
+            "username": "xxxx",
+            "password]": "xxx",
+            "host": "junos",
+            "port": "960",
+        }
+    }
+
+
+@pytest.fixture
+def configure_loader_modules():
+    return {junos: {"__pillar__": {}}}
+
+
+def test_init(opts):
+    with patch("ncclient.manager.connect") as mock_connect:
+        junos.init(opts)
+        assert junos.thisproxy.get("initialized")
+        mock_connect.assert_called_with(
+            allow_agent=True,
+            device_params={"name": "junos", "local": False, "use_filter": False},
+            host="junos",
+            hostkey_verify=False,
+            key_filename=None,
+            password=None,
+            port="960",
+            sock_fd=None,
+            ssh_config=ANY,
+            username="xxxx",
+        )
+
+
+def test_init_err(opts):
+    with patch("ncclient.manager.connect") as mock_connect:
+        mock_connect.side_effect = ConnectError
+        junos.init(opts)
+        assert not junos.thisproxy.get("initialized")
+
+
+def test_alive(opts):
+    with patch("ncclient.manager.connect") as mock_connect:
+        junos.init(opts)
+        junos.thisproxy["conn"]._conn._session._buffer = io.BytesIO()
+        assert junos.alive(opts)
+        assert junos.thisproxy.get("initialized")

tests/pytests/unit/proxy/test_napalm.py

@@ -0,0 +1,94 @@
+"""
+    :codeauthor: :email:`Anthony Shaw <anthonyshaw@apache.org>`
+"""
+import pytest
+
+import salt.proxy.napalm as napalm_proxy
+import tests.support.napalm as napalm_test_support
+from tests.support.mock import MagicMock, patch
+
+
+@pytest.fixture
+def test_opts():
+    return {
+        "proxytype": "napalm",
+        "driver": "junos",
+        "host": "core05.nrt02",
+        "id": "core05.nrt02",
+    }
+
+
+@pytest.fixture
+def configure_loader_modules(test_opts):
+    def mock_get_device(opts, *args, **kwargs):
+        assert opts == test_opts
+        return {"DRIVER": napalm_test_support.MockNapalmDevice(), "UP": True}
+
+    module_globals = {
+        "__salt__": {
+            "config.option": MagicMock(
+                return_value={"test": {"driver": "test", "key": "2orgk34kgk34g"}}
+            )
+        }
+    }
+    module_globals["napalm_base"] = MagicMock()
+    with patch("salt.utils.napalm.get_device", mock_get_device):
+        yield {napalm_proxy: module_globals}
+
+
+def test_init(test_opts):
+    ret = napalm_proxy.init(test_opts)
+    assert ret is True
+
+
+def test_alive(test_opts):
+    napalm_proxy.init(test_opts)
+    ret = napalm_proxy.alive(test_opts)
+    assert ret is True
+
+
+def test_ping(test_opts):
+    napalm_proxy.init(test_opts)
+    ret = napalm_proxy.ping()
+    assert ret is True
+
+
+def test_initialized(test_opts):
+    napalm_proxy.init(test_opts)
+    ret = napalm_proxy.initialized()
+    assert ret is True
+
+
+def test_get_device(test_opts):
+    napalm_proxy.init(test_opts)
+    ret = napalm_proxy.get_device()
+    assert ret["UP"] is True
+
+
+def test_get_grains(test_opts):
+    napalm_proxy.init(test_opts)
+    ret = napalm_proxy.get_grains()
+    assert ret["out"] == napalm_test_support.TEST_FACTS.copy()
+
+
+def test_grains_refresh(test_opts):
+    napalm_proxy.init(test_opts)
+    ret = napalm_proxy.grains_refresh()
+    assert ret["out"] == napalm_test_support.TEST_FACTS.copy()
+
+
+def test_fns():
+    ret = napalm_proxy.fns()
+    assert "details" in ret.keys()
+
+
+def test_shutdown(test_opts):
+    napalm_proxy.init(test_opts)
+    ret = napalm_proxy.shutdown(test_opts)
+    assert ret is True
+
+
+def test_call(test_opts):
+    napalm_proxy.init(test_opts)
+    ret = napalm_proxy.call("get_arp_table")
+    assert ret["out"] == napalm_test_support.TEST_ARP_TABLE.copy()

tests/pytests/unit/proxy/test_panos.py

@@ -0,0 +1,45 @@
+import pytest
+
+import salt.proxy.panos as panos
+from tests.support.mock import MagicMock, call, patch
+
+
+@pytest.fixture
+def configure_loader_modules():
+    with patch.dict(panos.DETAILS):
+        yield {panos: {"__pillar__": {}}}
+
+
+@pytest.fixture
+def opts():
+    return {"proxy": {"proxytype": "panos", "host": "hosturl.com", "apikey": "api_key"}}
+
+
+def test_init(opts):
+    for verify in [True, False, None]:
+        opts["proxy"]["verify_ssl"] = verify
+        if verify is None:
+            opts["proxy"].pop("verify_ssl")
+            verify = True
+        mock_http = MagicMock(
+            return_value={"status": 200, "text": "<data>some_test_data</data>"}
+        )
+        patch_http = patch.dict(panos.__utils__, {"http.query": mock_http})
+        with patch_http:
+            panos.init(opts)
+        assert mock_http.call_args_list == [
+            call(
+                "https://hosturl.com/api/",
+                data={
+                    "type": "op",
+                    "cmd": "<show><system><info></info></system></show>",
+                    "key": "api_key",
+                },
+                decode=True,
+                decode_type="plain",
+                method="POST",
+                raise_error=True,
+                status=True,
+                verify_ssl=verify,
+            )
+        ]

tests/unit/proxy/test_cimc.py

@@ -1,446 +0,0 @@
-"""
-tests.unit.proxy.test_cimc
-~~~~~~~~~~~~~~~~~~~~~~~~~~
-
-Unit tests for the cimc proxy module
-"""
-
-import logging
-
-import salt.exceptions
-import salt.proxy.cimc as cimc
-from tests.support.mixins import LoaderModuleMockMixin
-from tests.support.mock import MagicMock, patch
-from tests.support.unit import TestCase
-
-log = logging.getLogger(__name__)
-
-
-LOGIN_RESPONSE = """\
-<aaaLogin
-    response="yes"
-    outCookie="real-cookie"
-       outRefreshPeriod="600"
-       outPriv="admin">
-</aaaLogin>"""
-LOGOUT_RESPONSE = """\
-<aaaLogout
-    cookie="real-cookie"
-    response="yes"
-    outStatus="success">
-</aaaLogout>
-"""
-CONFIG_RESOLVE_CLASS_RESPONSE = """\
-<configResolveClass
-    cookie="real-cookie"
-    response="yes"
-    classId="computeRackUnit">
-    <outConfig>
-        <computeRackUnit
-            dn="sys/rack-unit-1"
-            adminPower="policy"
-            availableMemory="16384"
-            model="R210-2121605W"
-            memorySpeed="1067"
-            name="UCS C210 M2"
-            numOfAdaptors="2"
-            numOfCores="8"
-            numOfCoresEnabled="8"
-            numOfCpus="2"
-            numOfEthHostIfs="5"
-            numOfFcHostIfs="2"
-            numOfThreads="16"
-            operPower="on"
-            originalUuid="00C9DE3C-370D-DF11-1186-6DD1393A608B"
-            presence="equipped"
-            serverID="1"
-            serial="QCI140205Z2"
-            totalMemory="16384"
-            usrLbl="C210 Row-B Rack-10"
-            uuid="00C9DE3C-370D-DF11-1186-6DD1393A608B"
-            vendor="Cisco Systems Inc" >
-        </computeRackUnit>
-    </outConfig>
-</configResolveClass>
-"""
-CONFIG_CON_MO_RESPONSE = """\
-<configConfMo
-    dn="sys/rack-unit-1/locator-led"
-    cookie="real-cookie"
-    response="yes">
-    <outConfig>
-        <equipmentLocatorLed
-            dn="sys/rack-unit-1/locator-led"
-            adminState="inactive"
-            color="unknown"
-            id="1"
-            name=""
-            operState="off">
-        </equipmentLocatorLed>
-    </outConfig>
-</configConfMo>
-"""
-
-
-def http_query_response(*args, data=None, **kwargs):
-    log.debug(
-        "http_query_response side_effect; ARGS: %s // KWARGS: %s // DATA: %s",
-        args,
-        kwargs,
-        data,
-    )
-    if data.startswith("<aaaLogin"):
-        response = LOGIN_RESPONSE
-    elif data.startswith("<aaaLogout"):
-        response = LOGOUT_RESPONSE
-    elif data.startswith("<configResolveClass"):
-        response = CONFIG_RESOLVE_CLASS_RESPONSE
-    elif data.startswith("<configConfMo"):
-        response = CONFIG_CON_MO_RESPONSE
-    else:
-        response = ""
-    return {"text": response, "status": 200}
-
-
-class CIMCProxyTestCase(TestCase, LoaderModuleMockMixin):
-    def setup_loader_modules(self):
-        return {cimc: {"DETAILS": {}, "__pillar__": {}}}
-
-    def setUp(self):
-        self.opts = {"proxy": {"username": "xxxx", "password": "xxx", "host": "cimc"}}
-        self.addCleanup(delattr, self, "opts")
-
-    def test_init(self):
-        # No host, returns False
-        opts = {"proxy": {"username": "xxxx", "password": "xxx"}}
-        ret = cimc.init(opts)
-        self.assertFalse(ret)
-
-        # No username , returns False
-        opts = {"proxy": {"password": "xxx", "host": "cimc"}}
-        ret = cimc.init(opts)
-        self.assertFalse(ret)
-
-        # No password, returns False
-        opts = {"proxy": {"username": "xxxx", "host": "cimc"}}
-        ret = cimc.init(opts)
-        self.assertFalse(ret)
-
-        with patch.object(cimc, "logon", return_value="9zVG5U8DFZNsTR") as mock_logon:
-            with patch.object(
-                cimc, "get_config_resolver_class", return_value="True"
-            ) as mock_logon:
-                ret = cimc.init(self.opts)
-                self.assertEqual(cimc.DETAILS["url"], "https://cimc/nuova")
-                self.assertEqual(cimc.DETAILS["username"], "xxxx")
-                self.assertEqual(cimc.DETAILS["password"], "xxx")
-                self.assertTrue(cimc.DETAILS["initialized"])
-
-    def test__validate_response_code(self):
-        with self.assertRaisesRegex(
-            salt.exceptions.CommandExecutionError,
-            "Did not receive a valid response from host.",
-        ):
-            cimc._validate_response_code("404")
-
-        with patch.object(cimc, "logout", return_value=True) as mock_logout:
-            with self.assertRaisesRegex(
-                salt.exceptions.CommandExecutionError,
-                "Did not receive a valid response from host.",
-            ):
-                cimc._validate_response_code("404", "9zVG5U8DFZNsTR")
-                mock_logout.assert_called_once_with("9zVG5U8DFZNsTR")
-
-
-class ValidateSSLTestCase(TestCase, LoaderModuleMockMixin):
-    def setup_loader_modules(self):
-        return {cimc: {}}
-
-    def setUp(self):
-        cimc.DETAILS.clear()
-
-    def test_init(self):
-        verify_ssl_values = (None, True, False)
-        for verify_ssl_value in verify_ssl_values:
-            cimc.DETAILS.clear()
-            opts = {
-                "proxy": {
-                    "host": "TheHost",
-                    "username": "TheUsername",
-                    "password": "ThePassword",
-                    "verify_ssl": verify_ssl_value,
-                }
-            }
-            http_query_mock = MagicMock(side_effect=http_query_response)
-            if verify_ssl_value is None:
-                expected_verify_ssl_value = True
-            else:
-                expected_verify_ssl_value = verify_ssl_value
-
-            log.debug(
-                "verify_ssl: %s // expected verify_ssl: %s",
-                verify_ssl_value,
-                expected_verify_ssl_value,
-            )
-
-            with patch.dict(cimc.__utils__, {"http.query": http_query_mock}):
-                cimc.init(opts)
-
-            for idx, call in enumerate(http_query_mock.mock_calls, 1):
-                condition = call.kwargs["verify_ssl"] is expected_verify_ssl_value
-                condition_error = "{} != {}; Call(number={}): {}".format(
-                    idx, call, call.kwargs["verify_ssl"], expected_verify_ssl_value
-                )
-                self.assertTrue(condition, msg=condition_error)
-
-    def test_logon(self):
-        verify_ssl_values = (None, True, False)
-        for verify_ssl_value in verify_ssl_values:
-            cimc.DETAILS.clear()
-            opts = {
-                "proxy": {
-                    "host": "TheHost",
-                    "username": "TheUsername",
-                    "password": "ThePassword",
-                    "verify_ssl": verify_ssl_value,
-                }
-            }
-            http_query_mock = MagicMock(side_effect=http_query_response)
-            if verify_ssl_value is None:
-                expected_verify_ssl_value = True
-            else:
-                expected_verify_ssl_value = verify_ssl_value
-
-            # Let's init the proxy and ignore it's actions, this test is not about them
-            with patch(
-                "salt.proxy.cimc.get_config_resolver_class",
-                MagicMock(return_value=True),
-            ):
-                cimc.init(opts)
-
-            log.debug(
-                "verify_ssl: %s // expected verify_ssl: %s",
-                verify_ssl_value,
-                expected_verify_ssl_value,
-            )
-
-            with patch.dict(cimc.__utils__, {"http.query": http_query_mock}):
-                cimc.logon()
-
-            for idx, call in enumerate(http_query_mock.mock_calls, 1):
-                condition = call.kwargs["verify_ssl"] is expected_verify_ssl_value
-                condition_error = "{} != {}; Call(number={}): {}".format(
-                    idx, call, call.kwargs["verify_ssl"], expected_verify_ssl_value
-                )
-                self.assertTrue(condition, msg=condition_error)
-
-    def test_logout(self):
-        verify_ssl_values = (None, True, False)
-        for verify_ssl_value in verify_ssl_values:
-            cimc.DETAILS.clear()
-            opts = {
-                "proxy": {
-                    "host": "TheHost",
-                    "username": "TheUsername",
-                    "password": "ThePassword",
-                    "verify_ssl": verify_ssl_value,
-                }
-            }
-            http_query_mock = MagicMock(side_effect=http_query_response)
-            if verify_ssl_value is None:
-                expected_verify_ssl_value = True
-            else:
-                expected_verify_ssl_value = verify_ssl_value
-
-            # Let's init the proxy and ignore it's actions, this test is not about them
-            with patch(
-                "salt.proxy.cimc.get_config_resolver_class",
-                MagicMock(return_value=True),
-            ):
-                cimc.init(opts)
-
-            log.debug(
-                "verify_ssl: %s // expected verify_ssl: %s",
-                verify_ssl_value,
-                expected_verify_ssl_value,
-            )
-
-            with patch.dict(cimc.__utils__, {"http.query": http_query_mock}):
-                cimc.logout()
-
-            for idx, call in enumerate(http_query_mock.mock_calls, 1):
-                condition = call.kwargs["verify_ssl"] is expected_verify_ssl_value
-                condition_error = "{} != {}; Call(number={}): {}".format(
-                    idx, call, call.kwargs["verify_ssl"], expected_verify_ssl_value
-                )
-                self.assertTrue(condition, msg=condition_error)
-
-    def test_grains(self):
-        verify_ssl_values = (None, True, False)
-        for verify_ssl_value in verify_ssl_values:
-            cimc.DETAILS.clear()
-            opts = {
-                "proxy": {
-                    "host": "TheHost",
-                    "username": "TheUsername",
-                    "password": "ThePassword",
-                    "verify_ssl": verify_ssl_value,
-                }
-            }
-            http_query_mock = MagicMock(side_effect=http_query_response)
-            if verify_ssl_value is None:
-                expected_verify_ssl_value = True
-            else:
-                expected_verify_ssl_value = verify_ssl_value
-
-            # Let's init the proxy and ignore it's actions, this test is not about them
-            with patch(
-                "salt.proxy.cimc.get_config_resolver_class",
-                MagicMock(return_value=True),
-            ):
-                cimc.init(opts)
-
-            log.debug(
-                "verify_ssl: %s // expected verify_ssl: %s",
-                verify_ssl_value,
-                expected_verify_ssl_value,
-            )
-
-            with patch.dict(cimc.__utils__, {"http.query": http_query_mock}):
-                cimc.grains()
-
-            for idx, call in enumerate(http_query_mock.mock_calls, 1):
-                condition = call.kwargs["verify_ssl"] is expected_verify_ssl_value
-                condition_error = "{} != {}; Call(number={}): {}".format(
-                    idx, call, call.kwargs["verify_ssl"], expected_verify_ssl_value
-                )
-                self.assertTrue(condition, msg=condition_error)
-
-    def test_grains_refresh(self):
-        verify_ssl_values = (None, True, False)
-        for verify_ssl_value in verify_ssl_values:
-            cimc.DETAILS.clear()
-            opts = {
-                "proxy": {
-                    "host": "TheHost",
-                    "username": "TheUsername",
-                    "password": "ThePassword",
-                    "verify_ssl": verify_ssl_value,
-                }
-            }
-            http_query_mock = MagicMock(side_effect=http_query_response)
-            if verify_ssl_value is None:
-                expected_verify_ssl_value = True
-            else:
-                expected_verify_ssl_value = verify_ssl_value
-
-            # Let's init the proxy and ignore it's actions, this test is not about them
-            with patch(
-                "salt.proxy.cimc.get_config_resolver_class",
-                MagicMock(return_value=True),
-            ):
-                cimc.init(opts)
-
-            log.debug(
-                "verify_ssl: %s // expected verify_ssl: %s",
-                verify_ssl_value,
-                expected_verify_ssl_value,
-            )
-
-            with patch.dict(cimc.__utils__, {"http.query": http_query_mock}):
-                cimc.grains_refresh()
-
-            for idx, call in enumerate(http_query_mock.mock_calls, 1):
-                condition = call.kwargs["verify_ssl"] is expected_verify_ssl_value
-                condition_error = "{} != {}; Call(number={}): {}".format(
-                    idx, call, call.kwargs["verify_ssl"], expected_verify_ssl_value
-                )
-                self.assertTrue(condition, msg=condition_error)
-
-    def test_ping(self):
-        verify_ssl_values = (None, True, False)
-        for verify_ssl_value in verify_ssl_values:
-            cimc.DETAILS.clear()
-            opts = {
-                "proxy": {
-                    "host": "TheHost",
-                    "username": "TheUsername",
-                    "password": "ThePassword",
-                    "verify_ssl": verify_ssl_value,
-                }
-            }
-            http_query_mock = MagicMock(side_effect=http_query_response)
-            if verify_ssl_value is None:
-                expected_verify_ssl_value = True
-            else:
-                expected_verify_ssl_value = verify_ssl_value
-
-            # Let's init the proxy and ignore it's actions, this test is not about them
-            with patch(
-                "salt.proxy.cimc.get_config_resolver_class",
-                MagicMock(return_value=True),
-            ):
-                cimc.init(opts)
-
-            log.debug(
-                "verify_ssl: %s // expected verify_ssl: %s",
-                verify_ssl_value,
-                expected_verify_ssl_value,
-            )
-
-            with patch.dict(cimc.__utils__, {"http.query": http_query_mock}):
-                cimc.ping()
-
-            for idx, call in enumerate(http_query_mock.mock_calls, 1):
-                condition = call.kwargs["verify_ssl"] is expected_verify_ssl_value
-                condition_error = "{} != {}; Call(number={}): {}".format(
-                    idx, call, call.kwargs["verify_ssl"], expected_verify_ssl_value
-                )
-                self.assertTrue(condition, msg=condition_error)
-
-    def test_set_config_modify(self):
-        verify_ssl_values = (None, True, False)
-        for verify_ssl_value in verify_ssl_values:
-            cimc.DETAILS.clear()
-            opts = {
-                "proxy": {
-                    "host": "TheHost",
-                    "username": "TheUsername",
-                    "password": "ThePassword",
-                    "verify_ssl": verify_ssl_value,
-                }
-            }
-            http_query_mock = MagicMock(side_effect=http_query_response)
-            if verify_ssl_value is None:
-                expected_verify_ssl_value = True
-            else:
-                expected_verify_ssl_value = verify_ssl_value
-
-            # Let's init the proxy and ignore it's actions, this test is not about them
-            with patch(
-                "salt.proxy.cimc.get_config_resolver_class",
-                MagicMock(return_value=True),
-            ):
-                cimc.init(opts)
-
-            log.debug(
-                "verify_ssl: %s // expected verify_ssl: %s",
-                verify_ssl_value,
-                expected_verify_ssl_value,
-            )
-
-            with patch.dict(cimc.__utils__, {"http.query": http_query_mock}):
-                cimc.set_config_modify(
-                    dn="sys/rack-unit-1/locator-led",
-                    inconfig=(
-                        "<inConfig><equipmentLocatorLed adminState='on'"
-                        " dn='sys/rack-unit-1/locator-led'></equipmentLocatorLed></inConfig>"
-                    ),
-                )
-
-            for idx, call in enumerate(http_query_mock.mock_calls, 1):
-                condition = call.kwargs["verify_ssl"] is expected_verify_ssl_value
-                condition_error = "{} != {}; Call(number={}): {}".format(
-                    idx, call, call.kwargs["verify_ssl"], expected_verify_ssl_value
-                )
-                self.assertTrue(condition, msg=condition_error)

tests/unit/proxy/test_esxcluster.py

@@ -1,201 +0,0 @@
-"""
-    :codeauthor: :email:`Alexandru Bleotu <alexandru.bleotu@morganstanley.com>`
-
-    Tests for esxcluster proxy
-"""
-
-import salt.exceptions
-import salt.proxy.esxcluster as esxcluster
-from salt.config.schemas.esxcluster import EsxclusterProxySchema
-from tests.support.mixins import LoaderModuleMockMixin
-from tests.support.mock import MagicMock, patch
-from tests.support.unit import TestCase, skipIf
-
-try:
-    import jsonschema
-
-    HAS_JSONSCHEMA = True
-except ImportError:
-    HAS_JSONSCHEMA = False
-
-
-@skipIf(not HAS_JSONSCHEMA, "jsonschema is required")
-class InitTestCase(TestCase, LoaderModuleMockMixin):
-    """Tests for salt.proxy.esxcluster.init"""
-
-    def setup_loader_modules(self):
-        return {esxcluster: {"DETAILS": {}, "__pillar__": {}}}
-
-    def setUp(self):
-        self.opts_userpass = {
-            "proxy": {
-                "proxytype": "esxcluster",
-                "vcenter": "fake_vcenter",
-                "datacenter": "fake_dc",
-                "cluster": "fake_cluster",
-                "mechanism": "userpass",
-                "username": "fake_username",
-                "passwords": ["fake_password"],
-                "protocol": "fake_protocol",
-                "port": 100,
-            }
-        }
-        self.opts_sspi = {
-            "proxy": {
-                "proxytype": "esxcluster",
-                "vcenter": "fake_vcenter",
-                "datacenter": "fake_dc",
-                "cluster": "fake_cluster",
-                "mechanism": "sspi",
-                "domain": "fake_domain",
-                "principal": "fake_principal",
-                "protocol": "fake_protocol",
-                "port": 100,
-            }
-        }
-        patches = (
-            (
-                "salt.proxy.esxcluster.merge",
-                MagicMock(return_value=self.opts_sspi["proxy"]),
-            ),
-        )
-        for mod, mock in patches:
-            patcher = patch(mod, mock)
-            patcher.start()
-            self.addCleanup(patcher.stop)
-
-    def test_merge(self):
-        mock_pillar_proxy = MagicMock()
-        mock_opts_proxy = MagicMock()
-        mock_merge = MagicMock(return_value=self.opts_sspi["proxy"])
-        with patch.dict(esxcluster.__pillar__, {"proxy": mock_pillar_proxy}):
-            with patch("salt.proxy.esxcluster.merge", mock_merge):
-                esxcluster.init(opts={"proxy": mock_opts_proxy})
-        mock_merge.assert_called_once_with(mock_opts_proxy, mock_pillar_proxy)
-
-    def test_esxcluster_schema(self):
-        mock_json_validate = MagicMock()
-        serialized_schema = EsxclusterProxySchema().serialize()
-        with patch("salt.proxy.esxcluster.jsonschema.validate", mock_json_validate):
-            esxcluster.init(self.opts_sspi)
-        mock_json_validate.assert_called_once_with(
-            self.opts_sspi["proxy"], serialized_schema
-        )
-
-    def test_invalid_proxy_input_error(self):
-        with patch(
-            "salt.proxy.esxcluster.jsonschema.validate",
-            MagicMock(
-                side_effect=jsonschema.exceptions.ValidationError("Validation Error")
-            ),
-        ):
-            with self.assertRaises(salt.exceptions.InvalidConfigError) as excinfo:
-                esxcluster.init(self.opts_userpass)
-        self.assertEqual(excinfo.exception.strerror, "Validation Error")
-
-    def test_no_username(self):
-        opts = self.opts_userpass.copy()
-        del opts["proxy"]["username"]
-        with patch(
-            "salt.proxy.esxcluster.merge", MagicMock(return_value=opts["proxy"])
-        ):
-            with self.assertRaises(salt.exceptions.InvalidConfigError) as excinfo:
-                esxcluster.init(opts)
-        self.assertEqual(
-            excinfo.exception.strerror,
-            "Mechanism is set to 'userpass', but no "
-            "'username' key found in proxy config.",
-        )
-
-    def test_no_passwords(self):
-        opts = self.opts_userpass.copy()
-        del opts["proxy"]["passwords"]
-        with patch(
-            "salt.proxy.esxcluster.merge", MagicMock(return_value=opts["proxy"])
-        ):
-            with self.assertRaises(salt.exceptions.InvalidConfigError) as excinfo:
-                esxcluster.init(opts)
-        self.assertEqual(
-            excinfo.exception.strerror,
-            "Mechanism is set to 'userpass', but no "
-            "'passwords' key found in proxy config.",
-        )
-
-    def test_no_domain(self):
-        opts = self.opts_sspi.copy()
-        del opts["proxy"]["domain"]
-        with patch(
-            "salt.proxy.esxcluster.merge", MagicMock(return_value=opts["proxy"])
-        ):
-            with self.assertRaises(salt.exceptions.InvalidConfigError) as excinfo:
-                esxcluster.init(opts)
-        self.assertEqual(
-            excinfo.exception.strerror,
-            "Mechanism is set to 'sspi', but no 'domain' key found in proxy config.",
-        )
-
-    def test_no_principal(self):
-        opts = self.opts_sspi.copy()
-        del opts["proxy"]["principal"]
-        with patch(
-            "salt.proxy.esxcluster.merge", MagicMock(return_value=opts["proxy"])
-        ):
-            with self.assertRaises(salt.exceptions.InvalidConfigError) as excinfo:
-                esxcluster.init(opts)
-        self.assertEqual(
-            excinfo.exception.strerror,
-            "Mechanism is set to 'sspi', but no 'principal' key found in proxy config.",
-        )
-
-    def test_find_credentials(self):
-        mock_find_credentials = MagicMock(
-            return_value=("fake_username", "fake_password")
-        )
-        with patch(
-            "salt.proxy.esxcluster.merge",
-            MagicMock(return_value=self.opts_userpass["proxy"]),
-        ):
-            with patch("salt.proxy.esxcluster.find_credentials", mock_find_credentials):
-                esxcluster.init(self.opts_userpass)
-        mock_find_credentials.assert_called_once_with()
-
-    def test_details_userpass(self):
-        mock_find_credentials = MagicMock(
-            return_value=("fake_username", "fake_password")
-        )
-        with patch(
-            "salt.proxy.esxcluster.merge",
-            MagicMock(return_value=self.opts_userpass["proxy"]),
-        ):
-            with patch("salt.proxy.esxcluster.find_credentials", mock_find_credentials):
-                esxcluster.init(self.opts_userpass)
-        self.assertDictEqual(
-            esxcluster.DETAILS,
-            {
-                "vcenter": "fake_vcenter",
-                "datacenter": "fake_dc",
-                "cluster": "fake_cluster",
-                "mechanism": "userpass",
-                "username": "fake_username",
-                "password": "fake_password",
-                "passwords": ["fake_password"],
-                "protocol": "fake_protocol",
-                "port": 100,
-            },
-        )
-
-    def test_details_sspi(self):
-        esxcluster.init(self.opts_sspi)
-        self.assertDictEqual(
-            esxcluster.DETAILS,
-            {
-                "vcenter": "fake_vcenter",
-                "datacenter": "fake_dc",
-                "cluster": "fake_cluster",
-                "mechanism": "sspi",
-                "domain": "fake_domain",
-                "principal": "fake_principal",
-                "protocol": "fake_protocol",
-                "port": 100,
-            },
-        )

tests/unit/proxy/test_esxdatacenter.py

@@ -1,202 +0,0 @@
-"""
-    :codeauthor: :email:`Alexandru Bleotu <alexandru.bleotu@morganstanley.com>`
-
-    Tests for esxdatacenter proxy
-"""
-
-
-import salt.exceptions
-import salt.proxy.esxdatacenter as esxdatacenter
-from salt.config.schemas.esxdatacenter import EsxdatacenterProxySchema
-from tests.support.mixins import LoaderModuleMockMixin
-from tests.support.mock import MagicMock, patch
-from tests.support.unit import TestCase, skipIf
-
-try:
-    import jsonschema
-
-    HAS_JSONSCHEMA = True
-except ImportError:
-    HAS_JSONSCHEMA = False
-
-
-@skipIf(not HAS_JSONSCHEMA, "jsonschema is required")
-class InitTestCase(TestCase, LoaderModuleMockMixin):
-    """Tests for salt.proxy.esxdatacenter.init"""
-
-    def setup_loader_modules(self):
-        return {esxdatacenter: {"DETAILS": {}, "__pillar__": {}}}
-
-    def setUp(self):
-        self.opts_userpass = {
-            "proxy": {
-                "proxytype": "esxdatacenter",
-                "vcenter": "fake_vcenter",
-                "datacenter": "fake_dc",
-                "mechanism": "userpass",
-                "username": "fake_username",
-                "passwords": ["fake_password"],
-                "protocol": "fake_protocol",
-                "port": 100,
-            }
-        }
-        self.opts_sspi = {
-            "proxy": {
-                "proxytype": "esxdatacenter",
-                "vcenter": "fake_vcenter",
-                "datacenter": "fake_dc",
-                "mechanism": "sspi",
-                "domain": "fake_domain",
-                "principal": "fake_principal",
-                "protocol": "fake_protocol",
-                "port": 100,
-            }
-        }
-        patches = (
-            (
-                "salt.proxy.esxdatacenter.merge",
-                MagicMock(return_value=self.opts_sspi["proxy"]),
-            ),
-        )
-        for mod, mock in patches:
-            patcher = patch(mod, mock)
-            patcher.start()
-            self.addCleanup(patcher.stop)
-
-    def test_merge(self):
-        mock_pillar_proxy = MagicMock()
-        mock_opts_proxy = MagicMock()
-        mock_merge = MagicMock(return_value=self.opts_sspi["proxy"])
-        with patch.dict(esxdatacenter.__pillar__, {"proxy": mock_pillar_proxy}):
-            with patch("salt.proxy.esxdatacenter.merge", mock_merge):
-                esxdatacenter.init(opts={"proxy": mock_opts_proxy})
-        mock_merge.assert_called_once_with(mock_opts_proxy, mock_pillar_proxy)
-
-    def test_esxdatacenter_schema(self):
-        mock_json_validate = MagicMock()
-        serialized_schema = EsxdatacenterProxySchema().serialize()
-        with patch("salt.proxy.esxdatacenter.jsonschema.validate", mock_json_validate):
-            esxdatacenter.init(self.opts_sspi)
-        mock_json_validate.assert_called_once_with(
-            self.opts_sspi["proxy"], serialized_schema
-        )
-
-    def test_invalid_proxy_input_error(self):
-        with patch(
-            "salt.proxy.esxdatacenter.jsonschema.validate",
-            MagicMock(
-                side_effect=jsonschema.exceptions.ValidationError("Validation Error")
-            ),
-        ):
-            with self.assertRaises(salt.exceptions.InvalidConfigError) as excinfo:
-                esxdatacenter.init(self.opts_userpass)
-        self.assertEqual(excinfo.exception.strerror, "Validation Error")
-
-    def test_no_username(self):
-        opts = self.opts_userpass.copy()
-        del opts["proxy"]["username"]
-        with patch(
-            "salt.proxy.esxdatacenter.merge", MagicMock(return_value=opts["proxy"])
-        ):
-            with self.assertRaises(salt.exceptions.InvalidConfigError) as excinfo:
-                esxdatacenter.init(opts)
-        self.assertEqual(
-            excinfo.exception.strerror,
-            "Mechanism is set to 'userpass', but no "
-            "'username' key found in proxy config.",
-        )
-
-    def test_no_passwords(self):
-        opts = self.opts_userpass.copy()
-        del opts["proxy"]["passwords"]
-        with patch(
-            "salt.proxy.esxdatacenter.merge", MagicMock(return_value=opts["proxy"])
-        ):
-            with self.assertRaises(salt.exceptions.InvalidConfigError) as excinfo:
-                esxdatacenter.init(opts)
-        self.assertEqual(
-            excinfo.exception.strerror,
-            "Mechanism is set to 'userpass', but no "
-            "'passwords' key found in proxy config.",
-        )
-
-    def test_no_domain(self):
-        opts = self.opts_sspi.copy()
-        del opts["proxy"]["domain"]
-        with patch(
-            "salt.proxy.esxdatacenter.merge", MagicMock(return_value=opts["proxy"])
-        ):
-            with self.assertRaises(salt.exceptions.InvalidConfigError) as excinfo:
-                esxdatacenter.init(opts)
-        self.assertEqual(
-            excinfo.exception.strerror,
-            "Mechanism is set to 'sspi', but no 'domain' key found in proxy config.",
-        )
-
-    def test_no_principal(self):
-        opts = self.opts_sspi.copy()
-        del opts["proxy"]["principal"]
-        with patch(
-            "salt.proxy.esxdatacenter.merge", MagicMock(return_value=opts["proxy"])
-        ):
-            with self.assertRaises(salt.exceptions.InvalidConfigError) as excinfo:
-                esxdatacenter.init(opts)
-        self.assertEqual(
-            excinfo.exception.strerror,
-            "Mechanism is set to 'sspi', but no 'principal' key found in proxy config.",
-        )
-
-    def test_find_credentials(self):
-        mock_find_credentials = MagicMock(
-            return_value=("fake_username", "fake_password")
-        )
-        with patch(
-            "salt.proxy.esxdatacenter.merge",
-            MagicMock(return_value=self.opts_userpass["proxy"]),
-        ):
-            with patch(
-                "salt.proxy.esxdatacenter.find_credentials", mock_find_credentials
-            ):
-                esxdatacenter.init(self.opts_userpass)
-        mock_find_credentials.assert_called_once_with()
-
-    def test_details_userpass(self):
-        mock_find_credentials = MagicMock(
-            return_value=("fake_username", "fake_password")
-        )
-        with patch(
-            "salt.proxy.esxdatacenter.merge",
-            MagicMock(return_value=self.opts_userpass["proxy"]),
-        ):
-            with patch(
-                "salt.proxy.esxdatacenter.find_credentials", mock_find_credentials
-            ):
-                esxdatacenter.init(self.opts_userpass)
-        self.assertDictEqual(
-            esxdatacenter.DETAILS,
-            {
-                "vcenter": "fake_vcenter",
-                "datacenter": "fake_dc",
-                "mechanism": "userpass",
-                "username": "fake_username",
-                "password": "fake_password",
-                "passwords": ["fake_password"],
-                "protocol": "fake_protocol",
-                "port": 100,
-            },
-        )
-
-    def test_details_sspi(self):
-        esxdatacenter.init(self.opts_sspi)
-        self.assertDictEqual(
-            esxdatacenter.DETAILS,
-            {
-                "vcenter": "fake_vcenter",
-                "datacenter": "fake_dc",
-                "mechanism": "sspi",
-                "domain": "fake_domain",
-                "principal": "fake_principal",
-                "protocol": "fake_protocol",
-                "port": 100,
-            },
-        )

tests/unit/proxy/test_junos.py

@@ -1,61 +0,0 @@
-import io
-
-import salt.proxy.junos as junos
-from tests.support.mixins import LoaderModuleMockMixin
-from tests.support.mock import ANY, patch
-from tests.support.unit import TestCase, skipIf
-
-try:
-    import jxmlease  # pylint: disable=unused-import
-    from jnpr.junos.device import Device  # pylint: disable=unused-import
-    from jnpr.junos.exception import ConnectError
-
-    HAS_JUNOS = True
-except ImportError:
-    HAS_JUNOS = False
-
-
-@skipIf(not HAS_JUNOS, "The junos-eznc and jxmlease modules are required")
-class JunosProxyTestCase(TestCase, LoaderModuleMockMixin):
-    def setup_loader_modules(self):
-        return {junos: {"DETAILS": {}, "__pillar__": {}}}
-
-    def setUp(self):
-        self.opts = {
-            "proxy": {
-                "username": "xxxx",
-                "password]": "xxx",
-                "host": "junos",
-                "port": "960",
-            }
-        }
-
-    @patch("ncclient.manager.connect")
-    def test_init(self, mock_connect):
-        junos.init(self.opts)
-        self.assertTrue(junos.thisproxy.get("initialized"))
-        mock_connect.assert_called_with(
-            allow_agent=True,
-            device_params={"name": "junos", "local": False, "use_filter": False},
-            host="junos",
-            hostkey_verify=False,
-            key_filename=None,
-            password=None,
-            port="960",
-            sock_fd=None,
-            ssh_config=ANY,
-            username="xxxx",
-        )
-
-    @patch("ncclient.manager.connect")
-    def test_init_err(self, mock_connect):
-        mock_connect.side_effect = ConnectError
-        junos.init(self.opts)
-        self.assertFalse(junos.thisproxy.get("initialized"))
-
-    @patch("ncclient.manager.connect")
-    def test_alive(self, mock_connect):
-        junos.init(self.opts)
-        junos.thisproxy["conn"]._conn._session._buffer = io.BytesIO()
-        self.assertTrue(junos.alive(self.opts))
-        self.assertTrue(junos.thisproxy.get("initialized"))

tests/unit/proxy/test_napalm.py

@@ -1,84 +0,0 @@
-"""
-    :codeauthor: :email:`Anthony Shaw <anthonyshaw@apache.org>`
-"""
-
-
-import salt.proxy.napalm as napalm_proxy
-import tests.support.napalm as napalm_test_support
-from tests.support.mixins import LoaderModuleMockMixin
-from tests.support.mock import MagicMock, patch
-from tests.support.unit import TestCase
-
-
-class NapalmProxyTestCase(TestCase, LoaderModuleMockMixin):
-    def setup_loader_modules(self):
-        def mock_get_device(opts, *args, **kwargs):
-            assert opts == self.test_opts
-            return {"DRIVER": napalm_test_support.MockNapalmDevice(), "UP": True}
-
-        patcher = patch("salt.utils.napalm.get_device", mock_get_device)
-        patcher.start()
-        self.addCleanup(patcher.stop)
-        self.test_opts = {
-            "proxytype": "napalm",
-            "driver": "junos",
-            "host": "core05.nrt02",
-            "id": "core05.nrt02",
-        }
-        module_globals = {
-            "__salt__": {
-                "config.option": MagicMock(
-                    return_value={"test": {"driver": "test", "key": "2orgk34kgk34g"}}
-                )
-            }
-        }
-        module_globals["napalm_base"] = MagicMock()
-        return {napalm_proxy: module_globals}
-
-    def test_init(self):
-        ret = napalm_proxy.init(self.test_opts)
-        assert ret is True
-
-    def test_alive(self):
-        napalm_proxy.init(self.test_opts)
-        ret = napalm_proxy.alive(self.test_opts)
-        assert ret is True
-
-    def test_ping(self):
-        napalm_proxy.init(self.test_opts)
-        ret = napalm_proxy.ping()
-        assert ret is True
-
-    def test_initialized(self):
-        napalm_proxy.init(self.test_opts)
-        ret = napalm_proxy.initialized()
-        assert ret is True
-
-    def test_get_device(self):
-        napalm_proxy.init(self.test_opts)
-        ret = napalm_proxy.get_device()
-        assert ret["UP"] is True
-
-    def test_get_grains(self):
-        napalm_proxy.init(self.test_opts)
-        ret = napalm_proxy.get_grains()
-        assert ret["out"] == napalm_test_support.TEST_FACTS.copy()
-
-    def test_grains_refresh(self):
-        napalm_proxy.init(self.test_opts)
-        ret = napalm_proxy.grains_refresh()
-        assert ret["out"] == napalm_test_support.TEST_FACTS.copy()
-
-    def test_fns(self):
-        ret = napalm_proxy.fns()
-        assert "details" in ret.keys()
-
-    def test_shutdown(self):
-        napalm_proxy.init(self.test_opts)
-        ret = napalm_proxy.shutdown(self.test_opts)
-        assert ret is True
-
-    def test_call(self):
-        napalm_proxy.init(self.test_opts)
-        ret = napalm_proxy.call("get_arp_table")
-        assert ret["out"] == napalm_test_support.TEST_ARP_TABLE.copy()

tests/unit/proxy/test_nxos.py

@@ -1,475 +0,0 @@
-"""
-    :codeauthor: Mike Wiebe <@mikewiebe>
-"""
-
-# Copyright (c) 2019 Cisco and/or its affiliates.
-#
-# 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.
-
-
-import salt.proxy.nxos as nxos_proxy
-import salt.utils.nxos as nxos_utils
-from salt.exceptions import CommandExecutionError
-from tests.support.mixins import LoaderModuleMockMixin
-from tests.support.mock import MagicMock, create_autospec, patch
-from tests.support.unit import TestCase
-from tests.unit.modules.nxos.nxos_grains import n9k_grains
-from tests.unit.modules.nxos.nxos_show_cmd_output import n9k_show_ver_list
-
-
-class NxosNxapiProxyTestCase(TestCase, LoaderModuleMockMixin):
-    def setup_loader_modules(self):
-        return {nxos_proxy: {"CONNECTION": "nxapi"}}
-
-    def test_check_virtual(self):
-
-        """UT: nxos module:check_virtual method - return value"""
-
-        result = nxos_proxy.__virtual__()
-        self.assertIn("nxos", result)
-
-    def test_init(self):
-
-        """UT: nxos module:init method - nxapi proxy"""
-
-        with patch.object(nxos_proxy, "__opts__", {"proxy": {"connection": "nxapi"}}):
-            with patch("salt.proxy.nxos._init_nxapi", autospec=True) as init_nxapi:
-                result = nxos_proxy.init()
-                self.assertEqual(result, init_nxapi.return_value)
-
-    def test_init_opts_none(self):
-
-        """UT: nxos module:init method - __opts__ connection is None"""
-
-        with patch.object(nxos_proxy, "__opts__", {"proxy": {"connection": None}}):
-            with patch("salt.proxy.nxos._init_nxapi", autospec=True) as init_nxapi:
-                result = nxos_proxy.init()
-                self.assertEqual(result, init_nxapi.return_value)
-
-    def test_init_bad_connection_type(self):
-
-        """UT: nxos module:init method - bad CONNECTION type"""
-        with patch.object(nxos_proxy, "__opts__", {"proxy": {"connection": "unknown"}}):
-            self.assertFalse(nxos_proxy.init())
-
-    def test_initialized(self):
-
-        """UT: nxos module:initialized method - nxapi proxy"""
-
-        with patch(
-            "salt.proxy.nxos._initialized_nxapi", autospec=True
-        ) as initialized_nxapi:
-            result = nxos_proxy.initialized()
-            self.assertEqual(result, initialized_nxapi.return_value)
-
-    def test_ping(self):
-
-        """UT: nxos module:ping method - nxapi proxy"""
-
-        with patch("salt.proxy.nxos._ping_nxapi", autospec=True) as ping_nxapi:
-            result = nxos_proxy.ping()
-            self.assertEqual(result, ping_nxapi.return_value)
-
-    def test_grains(self):
-
-        """UT: nxos module:grains method - nxapi grains"""
-
-        with patch(
-            "salt.proxy.nxos.sendline", autospec=True, return_value=n9k_show_ver_list
-        ):
-            result = nxos_proxy.grains()
-            self.assertEqual(result, n9k_grains)
-
-    def test_grains_cache_set(self):
-
-        """UT: nxos module:grains method - nxapi grains cache set"""
-
-        with patch(
-            "salt.proxy.nxos.DEVICE_DETAILS", {"grains_cache": n9k_grains["nxos"]}
-        ):
-            with patch(
-                "salt.proxy.nxos.sendline",
-                autospec=True,
-                return_value=n9k_show_ver_list,
-            ):
-                result = nxos_proxy.grains()
-                self.assertEqual(result, n9k_grains)
-
-    def test_grains_refresh(self):
-
-        """UT: nxos module:grains_refresh method - nxapi grains"""
-
-        device_details = {"grains_cache": None}
-
-        with patch("salt.proxy.nxos.DEVICE_DETAILS", device_details):
-            with patch("salt.proxy.nxos.grains", autospec=True) as grains:
-                result = nxos_proxy.grains_refresh()
-                self.assertEqual(nxos_proxy.DEVICE_DETAILS["grains_cache"], {})
-                self.assertEqual(result, grains.return_value)
-
-    def test_sendline(self):
-
-        """UT: nxos module:sendline method - nxapi"""
-
-        command = "show version"
-
-        with patch("salt.proxy.nxos._nxapi_request", autospec=True) as nxapi_request:
-            result = nxos_proxy.sendline(command)
-            self.assertEqual(result, nxapi_request.return_value)
-
-    def test_proxy_config(self):
-
-        """UT: nxos module:proxy_config method - nxapi success path"""
-
-        commands = ["feature bgp", "router bgp 65535"]
-
-        with patch("salt.proxy.nxos.DEVICE_DETAILS", {"save_config": False}):
-            with patch(
-                "salt.proxy.nxos._nxapi_request", autospec=True
-            ) as nxapi_request:
-                result = nxos_proxy.proxy_config(commands)
-                self.assertEqual(result, [commands, nxapi_request.return_value])
-
-    def test_proxy_config_save_config(self):
-
-        """UT: nxos module:proxy_config method - nxapi success path"""
-
-        commands = ["feature bgp", "router bgp 65535"]
-
-        with patch("salt.proxy.nxos.DEVICE_DETAILS", {"save_config": None}):
-            with patch(
-                "salt.proxy.nxos._nxapi_request", autospec=True
-            ) as nxapi_request:
-                result = nxos_proxy.proxy_config(commands, save_config=True)
-                self.assertEqual(result, [commands, nxapi_request.return_value])
-
-    def test__init_nxapi(self):
-
-        """UT: nxos module:_init_nxapi method - successful connectinon"""
-
-        opts = {"proxy": {"arg1": None}}
-        nxapi_request = create_autospec(nxos_utils.nxapi_request, return_value="data")
-
-        with patch("salt.proxy.nxos.DEVICE_DETAILS", {}) as device_details:
-            with patch(
-                "salt.proxy.nxos.__utils__", {"nxos.nxapi_request": nxapi_request}
-            ):
-                result = nxos_proxy._init_nxapi(opts)
-
-                self.assertTrue(device_details["initialized"])
-                self.assertTrue(device_details["up"])
-                self.assertTrue(device_details["save_config"])
-                self.assertTrue(result)
-
-                nxapi_request.assert_called_with("show clock", **opts["proxy"])
-
-    def test_bad__init_nxapi(self):
-        class NXAPIException(Exception):
-            pass
-
-        nxapi_request = create_autospec(
-            nxos_utils.nxapi_request, side_effect=NXAPIException
-        )
-
-        with patch("salt.proxy.nxos.__utils__", {"nxos.nxapi_request": nxapi_request}):
-            with patch("salt.proxy.nxos.log", autospec=True) as log:
-                with self.assertRaises(NXAPIException):
-                    nxos_proxy._init_nxapi({"proxy": {"host": "HOST"}})
-                log.error.assert_called()
-
-    def test__initialized_nxapi(self):
-
-        """UT: nxos module:_initialized_nxapi method"""
-
-        result = nxos_proxy._initialized_nxapi()
-        self.assertFalse(result)
-
-        with patch("salt.proxy.nxos.DEVICE_DETAILS", {"initialized": True}):
-            result = nxos_proxy._initialized_nxapi()
-            self.assertTrue(result)
-
-    def test__ping_nxapi(self):
-
-        """UT: nxos module:_ping_nxapi method"""
-
-        result = nxos_proxy._ping_nxapi()
-        self.assertFalse(result)
-
-        with patch("salt.proxy.nxos.DEVICE_DETAILS", {"up": True}):
-            result = nxos_proxy._ping_nxapi()
-            self.assertTrue(result)
-
-    def test__shutdown_nxapi(self):
-
-        """UT: nxos module:_shutdown_nxapi method"""
-
-        opts = {"id": "value"}
-
-        with patch("salt.proxy.nxos.log", autospec=True):
-            nxos_proxy._shutdown_nxapi()
-            # nothing to test
-
-    def test__nxapi_request_ssh_return(self):
-
-        """UT: nxos module:_nxapi_request method - CONNECTION == 'ssh'"""
-
-        commands = "show version"
-
-        with patch("salt.proxy.nxos.CONNECTION", "ssh"):
-            result = nxos_proxy._nxapi_request(commands)
-            self.assertEqual("_nxapi_request is not available for ssh proxy", result)
-
-    def test__nxapi_request_connect(self):
-
-        """UT: nxos module:_nxapi_request method"""
-
-        commands = "show version"
-        nxapi_request = create_autospec(nxos_utils.nxapi_request, return_value="data")
-
-        with patch("salt.proxy.nxos.DEVICE_DETAILS", {"conn_args": {"arg1": None}}):
-            with patch(
-                "salt.proxy.nxos.__utils__", {"nxos.nxapi_request": nxapi_request}
-            ):
-                result = nxos_proxy._nxapi_request(commands)
-                self.assertEqual("data", result)
-                nxapi_request.assert_called_with(commands, method="cli_conf", arg1=None)
-
-
-class NxosSSHProxyTestCase(TestCase, LoaderModuleMockMixin):
-    def setup_loader_modules(self):
-        return {
-            nxos_proxy: {
-                "__opts__": {
-                    "proxy": {
-                        "host": "dt-n9k5-1.cisco.com",
-                        "username": "admin",
-                        "password": "password",
-                    }
-                },
-                "CONNECTION": "ssh",
-            }
-        }
-
-    def test_init(self):
-
-        """UT: nxos module:init method - ssh proxy"""
-
-        with patch("salt.proxy.nxos._init_ssh", autospec=True) as init_ssh:
-            result = nxos_proxy.init()
-            self.assertEqual(result, init_ssh.return_value)
-
-    def test_init_opts_none(self):
-
-        """UT: nxos module:init method - __opts__ connection is None"""
-
-        with patch("salt.proxy.nxos.__opts__", {"proxy": {"connection": None}}):
-            with patch("salt.proxy.nxos._init_ssh", autospec=True) as init_ssh:
-                result = nxos_proxy.init()
-                self.assertEqual(result, init_ssh.return_value)
-
-    def test_initialized(self):
-
-        """UT: nxos module:initialized method - ssh proxy"""
-
-        with patch(
-            "salt.proxy.nxos._initialized_ssh", autospec=True
-        ) as initialized_ssh:
-            result = nxos_proxy.initialized()
-            self.assertEqual(result, initialized_ssh.return_value)
-
-    def test_ping(self):
-
-        """UT: nxos module:ping method - ssh proxy"""
-
-        with patch("salt.proxy.nxos._ping_ssh", autospec=True) as ping_ssh:
-            result = nxos_proxy.ping()
-            self.assertEqual(result, ping_ssh.return_value)
-
-    def test_grains(self):
-
-        """UT: nxos module:grains method - ssh grains"""
-
-        with patch(
-            "salt.proxy.nxos.sendline", autospec=True, return_value=n9k_show_ver_list[0]
-        ):
-            result = nxos_proxy.grains()
-            self.assertEqual(result, n9k_grains)
-
-    def test_sendline(self):
-
-        """UT: nxos module:sendline method - nxapi"""
-
-        command = "show version"
-
-        with patch("salt.proxy.nxos._sendline_ssh", autospec=True) as sendline_ssh:
-            result = nxos_proxy.sendline(command)
-            self.assertEqual(result, sendline_ssh.return_value)
-
-    def test_proxy_config(self):
-
-        """UT: nxos module:proxy_config method - ssh success path"""
-
-        commands = ["feature bgp", "router bgp 65535"]
-
-        with patch("salt.proxy.nxos.DEVICE_DETAILS", {"save_config": False}):
-            with patch("salt.proxy.nxos._sendline_ssh", autospec=True) as sendline_ssh:
-                result = nxos_proxy.proxy_config(commands)
-                self.assertEqual(result, [commands, sendline_ssh.return_value])
-
-    def test_proxy_config_save_config(self):
-
-        """UT: nxos module:proxy_config method - ssh success path"""
-
-        commands = ["feature bgp", "router bgp 65535"]
-
-        with patch("salt.proxy.nxos.DEVICE_DETAILS", {"save_config": None}):
-            with patch("salt.proxy.nxos._sendline_ssh", autospec=True) as sendline_ssh:
-                result = nxos_proxy.proxy_config(commands, save_config=True)
-                self.assertEqual(result, [commands, sendline_ssh.return_value])
-
-    def test_proxy_config_error(self):
-
-        """UT: nxos module:proxy_config method - CommandExecutionError"""
-
-        with patch(
-            "salt.proxy.nxos._sendline_ssh",
-            autospec=True,
-            side_effect=CommandExecutionError,
-        ):
-            with self.assertRaises(CommandExecutionError):
-                nxos_proxy.proxy_config("show version", save_config=True)
-
-    def test__init_ssh_device_details(self):
-        with patch("salt.proxy.nxos.SSHConnection", autospec=True) as SSHConnection:
-            SSHConnection().sendline.return_value = ["", ""]
-
-            with patch("salt.proxy.nxos.DEVICE_DETAILS", {}) as device_details:
-                nxos_proxy._init_ssh(None)
-                self.assertIn(nxos_proxy._worker_name(), device_details)
-                self.assertTrue(device_details["initialized"])
-                self.assertTrue(device_details["save_config"])
-
-            with patch.dict(nxos_proxy.__opts__["proxy"], {"save_config": False}):
-                with patch("salt.proxy.nxos.DEVICE_DETAILS", {}) as device_details:
-                    nxos_proxy._init_ssh(None)
-                    self.assertIn(nxos_proxy._worker_name(), device_details)
-                    self.assertTrue(device_details["initialized"])
-                    self.assertFalse(device_details["save_config"])
-
-    def test__init_ssh_opts(self):
-
-        """UT: nxos module:_init_ssh method - successful connectinon"""
-
-        with patch("salt.proxy.nxos.DEVICE_DETAILS", {}):
-            with patch("salt.proxy.nxos.SSHConnection", autospec=True) as SSHConnection:
-                SSHConnection().sendline.return_value = ["", ""]
-                nxos_proxy._init_ssh(None)
-                self.assertEqual(
-                    nxos_proxy.__opts__["proxy"]["host"],
-                    SSHConnection.call_args[1]["host"],
-                )
-
-                opts = MagicMock()
-                nxos_proxy._init_ssh(opts)
-                self.assertEqual(
-                    opts["proxy"]["host"], SSHConnection.call_args[1]["host"]
-                )
-
-    def test__init_ssh_prompt(self):
-
-        """UT: nxos module:_init_ssh method - prompt regex"""
-
-        with patch("salt.proxy.nxos.DEVICE_DETAILS", {}):
-            with patch("salt.proxy.nxos.SSHConnection", autospec=True) as SSHConnection:
-                SSHConnection().sendline.return_value = ["", ""]
-
-                with patch.dict(
-                    nxos_proxy.__opts__["proxy"], {"prompt_regex": "n9k.*device"}
-                ):
-                    nxos_proxy._init_ssh(None)
-                    self.assertEqual(
-                        "n9k.*device", SSHConnection.call_args[1]["prompt"]
-                    )
-
-                with patch.dict(
-                    nxos_proxy.__opts__["proxy"], {"prompt_name": "n9k-device"}
-                ):
-                    nxos_proxy._init_ssh(None)
-                    self.assertEqual(
-                        "n9k-device.*#", SSHConnection.call_args[1]["prompt"]
-                    )
-
-                nxos_proxy._init_ssh(None)
-                self.assertEqual(".+#$", SSHConnection.call_args[1]["prompt"])
-
-    def test__initialized_ssh(self):
-
-        """UT: nxos module:_initialized_ssh method"""
-
-        with patch("salt.proxy.nxos.DEVICE_DETAILS", {"initialized": True}):
-            result = nxos_proxy._initialized_ssh()
-            self.assertTrue(result)
-
-        with patch("salt.proxy.nxos.DEVICE_DETAILS", {}):
-            result = nxos_proxy._initialized_ssh()
-            self.assertFalse(result)
-
-    def test__parse_output_for_errors(self):
-
-        """UT: nxos module:_parse_output_for_errors method"""
-
-        data = "% Incomplete command at '^' marker."
-        command = "show"
-
-        with self.assertRaises(CommandExecutionError):
-            nxos_proxy._parse_output_for_errors(
-                data, command, error_pattern="Incomplete"
-            )
-
-        data = "% Incomplete command at '^' marker."
-        command = "show"
-
-        with self.assertRaises(CommandExecutionError):
-            nxos_proxy._parse_output_for_errors(
-                data, command, error_pattern=["Incomplete", "marker"]
-            )
-
-        data = "% Invalid command at '^' marker."
-        command = "show bep"
-
-        with self.assertRaises(CommandExecutionError):
-            nxos_proxy._parse_output_for_errors(data, command)
-
-        data = "% Incomplete command at '^' marker."
-        command = "show"
-
-        nxos_proxy._parse_output_for_errors(data, command)
-
-        data = "% Incomplete command at '^' marker."
-        command = "show"
-
-        nxos_proxy._parse_output_for_errors(data, command, error_pattern="foo")
-
-    def test__init_ssh_raise_exception(self):
-
-        """UT: nxos module:_init_ssh method - raise exception"""
-
-        class SSHException(Exception):
-            pass
-
-        with patch("salt.proxy.nxos.SSHConnection", autospec=True) as SSHConnection:
-            with patch("salt.proxy.nxos.log", autospec=True) as log:
-                with self.assertRaises(SSHException):
-                    SSHConnection.side_effect = SSHException
-                    nxos_proxy._init_ssh(None)
-                log.error.assert_called()

tests/unit/proxy/test_panos.py

@@ -1,46 +0,0 @@
-import salt.proxy.panos as panos
-from tests.support.mixins import LoaderModuleMockMixin
-from tests.support.mock import MagicMock, call, patch
-from tests.support.unit import TestCase
-
-
-class PanosProxyTestCase(TestCase, LoaderModuleMockMixin):
-    def setup_loader_modules(self):
-        return {panos: {"DETAILS": {}, "__pillar__": {}}}
-
-    def setUp(self):
-        self.opts = {
-            "proxy": {"proxytype": "panos", "host": "hosturl.com", "apikey": "api_key"}
-        }
-
-    def test_init(self):
-        for verify in [True, False, None]:
-            self.opts["proxy"]["verify_ssl"] = verify
-            if verify is None:
-                self.opts["proxy"].pop("verify_ssl")
-                verify = True
-            mock_http = MagicMock(
-                return_value={"status": 200, "text": "<data>some_test_data</data>"}
-            )
-            patch_http = patch.dict(panos.__utils__, {"http.query": mock_http})
-            with patch_http:
-                panos.init(self.opts)
-            self.assertEqual(
-                mock_http.call_args_list,
-                [
-                    call(
-                        "https://hosturl.com/api/",
-                        data={
-                            "type": "op",
-                            "cmd": "<show><system><info></info></system></show>",
-                            "key": "api_key",
-                        },
-                        decode=True,
-                        decode_type="plain",
-                        method="POST",
-                        raise_error=True,
-                        status=True,
-                        verify_ssl=verify,
-                    )
-                ],
-            )