# -*- coding: utf-8 -*- """ :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 Python Libs from __future__ import absolute_import, print_function, unicode_literals import salt.proxy.nxos as nxos_proxy import salt.utils.nxos as nxos_utils from salt.exceptions import CommandExecutionError # Import Salt Testing Libs 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()