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