1# (c) 2017 Red Hat Inc. 2# 3# This file is part of Ansible 4# 5# Ansible is free software: you can redistribute it and/or modify 6# it under the terms of the GNU General Public License as published by 7# the Free Software Foundation, either version 3 of the License, or 8# (at your option) any later version. 9# 10# Ansible is distributed in the hope that it will be useful, 11# but WITHOUT ANY WARRANTY; without even the implied warranty of 12# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 13# GNU General Public License for more details. 14# 15# You should have received a copy of the GNU General Public License 16# along with Ansible. If not, see <http://www.gnu.org/licenses/>. 17 18# Make coding more python3-ish 19from __future__ import absolute_import, division, print_function 20 21__metaclass__ = type 22 23try: 24 from lxml.etree import fromstring 25except ImportError: 26 from xml.etree.ElementTree import fromstring 27 28from ansible_collections.junipernetworks.junos.tests.unit.compat.mock import ( 29 patch, 30) 31from ansible_collections.junipernetworks.junos.plugins.modules import ( 32 junos_command, 33) 34from ansible_collections.junipernetworks.junos.tests.unit.modules.utils import ( 35 set_module_args, 36) 37from .junos_module import TestJunosModule, load_fixture 38 39RPC_CLI_MAP = {"get-software-information": "show version"} 40 41 42class TestJunosCommandModule(TestJunosModule): 43 44 module = junos_command 45 46 def setUp(self): 47 super(TestJunosCommandModule, self).setUp() 48 49 self.mock_conn = patch( 50 "ansible_collections.junipernetworks.junos.plugins.module_utils.network.junos.junos.Connection" 51 ) 52 self.conn = self.mock_conn.start() 53 54 self.mock_netconf = patch( 55 "ansible_collections.junipernetworks.junos.plugins.module_utils.network.junos.junos.NetconfConnection" 56 ) 57 self.netconf_conn = self.mock_netconf.start() 58 59 self.mock_exec_rpc = patch( 60 "ansible_collections.junipernetworks.junos.plugins.modules.junos_command.exec_rpc" 61 ) 62 self.exec_rpc = self.mock_exec_rpc.start() 63 64 self.mock_netconf_rpc = patch( 65 "ansible_collections.ansible.netcommon.plugins.module_utils.network.common.netconf.NetconfConnection" 66 ) 67 self.netconf_rpc = self.mock_netconf_rpc.start() 68 69 self.mock_get_connection = patch( 70 "ansible_collections.junipernetworks.junos.plugins.modules.junos_command.get_connection" 71 ) 72 self.get_connection = self.mock_get_connection.start() 73 74 self.mock_get_capabilities = patch( 75 "ansible_collections.junipernetworks.junos.plugins.modules.junos_command.get_capabilities" 76 ) 77 self.get_capabilities = self.mock_get_capabilities.start() 78 self.get_capabilities.return_value = {"network_api": "netconf"} 79 80 def tearDown(self): 81 super(TestJunosCommandModule, self).tearDown() 82 self.mock_conn.stop() 83 self.mock_netconf.stop() 84 self.mock_get_capabilities.stop() 85 self.mock_netconf_rpc.stop() 86 self.mock_exec_rpc.stop() 87 self.mock_get_connection.stop() 88 89 def load_fixtures(self, commands=None, format="text", changed=False): 90 def load_from_file(*args, **kwargs): 91 element = fromstring(args[1]) 92 if element.text: 93 path = str(element.text) 94 else: 95 path = RPC_CLI_MAP[str(element.tag)] 96 97 filename = path.replace(" ", "_") 98 filename = "%s_%s.txt" % (filename, format) 99 return load_fixture(filename) 100 101 self.exec_rpc.side_effect = load_from_file 102 103 def test_junos_command_simple(self): 104 set_module_args(dict(commands=["show version"])) 105 result = self.execute_module() 106 self.assertEqual(len(result["stdout"]), 1) 107 self.assertTrue(result["stdout"][0].startswith("Hostname:")) 108 109 def test_junos_command_multiple(self): 110 set_module_args(dict(commands=["show version", "show version"])) 111 result = self.execute_module() 112 self.assertEqual(len(result["stdout"]), 2) 113 self.assertTrue(result["stdout"][0].startswith("Hostname:")) 114 115 def test_junos_command_wait_for(self): 116 wait_for = 'result[0] contains "Junos:"' 117 set_module_args(dict(commands=["show version"], wait_for=wait_for)) 118 self.execute_module() 119 120 def test_junos_command_wait_for_fails(self): 121 wait_for = 'result[0] contains "test string"' 122 set_module_args(dict(commands=["show version"], wait_for=wait_for)) 123 self.execute_module(failed=True) 124 self.assertEqual(self.exec_rpc.call_count, 10) 125 126 def test_junos_command_retries(self): 127 wait_for = 'result[0] contains "test string"' 128 set_module_args( 129 dict(commands=["show version"], wait_for=wait_for, retries=2) 130 ) 131 self.execute_module(failed=True) 132 self.assertEqual(self.exec_rpc.call_count, 2) 133 134 def test_junos_command_match_any(self): 135 wait_for = [ 136 'result[0] contains "Junos:"', 137 'result[0] contains "test string"', 138 ] 139 set_module_args( 140 dict(commands=["show version"], wait_for=wait_for, match="any") 141 ) 142 self.execute_module() 143 144 def test_junos_command_match_all(self): 145 wait_for = [ 146 'result[0] contains "Junos:"', 147 'result[0] contains "JUNOS Software Release"', 148 ] 149 set_module_args( 150 dict(commands=["show version"], wait_for=wait_for, match="all") 151 ) 152 self.execute_module() 153 154 def test_junos_command_match_all_failure(self): 155 wait_for = [ 156 'result[0] contains "Junos:"', 157 'result[0] contains "test string"', 158 ] 159 commands = ["show version", "show version"] 160 set_module_args( 161 dict(commands=commands, wait_for=wait_for, match="all") 162 ) 163 self.execute_module(failed=True) 164 165 def test_junos_command_simple_json(self): 166 set_module_args(dict(commands=["show version"], display="json")) 167 result = self.execute_module(format="json") 168 self.assertEqual(len(result["stdout"]), 1) 169 self.assertTrue("software-information" in result["stdout"][0]) 170 171 def test_junos_command_simple_rpc_text(self): 172 set_module_args( 173 dict(rpcs=["get-software-information"], display="text") 174 ) 175 result = self.execute_module(format="text") 176 self.assertEqual(len(result["stdout"]), 1) 177 self.assertTrue(result["stdout"][0].startswith("Hostname:")) 178 179 def test_junos_command_simple_rpc_json(self): 180 set_module_args( 181 dict(rpcs=["get-software-information"], display="json") 182 ) 183 result = self.execute_module(format="json") 184 self.assertEqual(len(result["stdout"]), 1) 185 self.assertTrue("software-information" in result["stdout"][0]) 186