1 # Copyright (c) 2016-2017 Intel Corporation
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
16 from __future__ import absolute_import
21 from collections import defaultdict
26 from six.moves.configparser import ConfigParser
27 from six.moves import StringIO
29 from yardstick.common import ansible_common
31 PREFIX = 'yardstick.common.ansible_common'
34 class OverwriteDictTestCase(unittest.TestCase):
35 def test_overwrite_dict_cfg(self):
36 c = ConfigParser(allow_no_value=True)
38 "section_a": "empty_value",
39 "section_b": {"key_c": "Val_d", "key_d": "VAL_D"},
40 "section_c": ["key_c", "key_d"],
42 ansible_common.overwrite_dict_to_cfg(c, d)
43 # Python3 and Python2 convert empty values into None or ''
44 # we don't really care but we need to compare correctly for unittest
45 self.assertTrue(c.has_option("section_a", "empty_value"))
46 self.assertEqual(sorted(c.items("section_b")), [('key_c', 'Val_d'), ('key_d', 'VAL_D')])
47 self.assertTrue(c.has_option("section_c", "key_c"))
48 self.assertTrue(c.has_option("section_c", "key_d"))
51 class FilenameGeneratorTestCase(unittest.TestCase):
52 @mock.patch('{}.NamedTemporaryFile'.format(PREFIX))
53 def test__handle_existing_file(self, _):
54 ansible_common.FileNameGenerator._handle_existing_file("/dev/null")
56 def test_get_generator_from_file(self):
57 ansible_common.FileNameGenerator.get_generator_from_filename("/dev/null", "", "", "")
59 def test_get_generator_from_file_middle(self):
60 ansible_common.FileNameGenerator.get_generator_from_filename("/dev/null", "", "",
63 def test_get_generator_from_file_prefix(self):
64 ansible_common.FileNameGenerator.get_generator_from_filename("/dev/null", "", "null",
68 class AnsibleNodeTestCase(unittest.TestCase):
69 def test_ansible_node(self):
70 ansible_common.AnsibleNode()
72 def test_ansible_node_len(self):
73 a = ansible_common.AnsibleNode()
76 def test_ansible_node_repr(self):
77 a = ansible_common.AnsibleNode()
80 def test_ansible_node_iter(self):
81 a = ansible_common.AnsibleNode()
85 def test_is_role(self):
86 a = ansible_common.AnsibleNode()
87 self.assertFalse(a.is_role("", default="foo"))
89 def test_ansible_node_get_tuple(self):
90 a = ansible_common.AnsibleNode({"name": "name"})
91 self.assertEqual(a.get_tuple(), ('name', a))
93 def test_gen_inventory_line(self):
94 a = ansible_common.AnsibleNode(defaultdict(str))
95 self.assertEqual(a.gen_inventory_line(), "")
97 def test_ansible_node_delitem(self):
98 a = ansible_common.AnsibleNode({"name": "name"})
101 def test_ansible_node_getattr(self):
102 a = ansible_common.AnsibleNode({"name": "name"})
103 self.assertIsNone(getattr(a, "nosuch", None))
106 class AnsibleNodeDictTestCase(unittest.TestCase):
107 def test_ansible_node_dict(self):
108 n = ansible_common.AnsibleNode
109 ansible_common.AnsibleNodeDict(n, {})
111 def test_ansible_node_dict_len(self):
112 n = ansible_common.AnsibleNode
113 a = ansible_common.AnsibleNodeDict(n, {})
116 def test_ansible_node_dict_repr(self):
117 n = ansible_common.AnsibleNode
118 a = ansible_common.AnsibleNodeDict(n, {})
121 def test_ansible_node_dict_iter(self):
122 n = ansible_common.AnsibleNode
123 a = ansible_common.AnsibleNodeDict(n, {})
127 def test_ansible_node_dict_get(self):
128 n = ansible_common.AnsibleNode
129 a = ansible_common.AnsibleNodeDict(n, {})
130 self.assertIsNone(a.get(""))
132 def test_gen_inventory_lines_for_all_of_type(self):
133 n = ansible_common.AnsibleNode
134 a = ansible_common.AnsibleNodeDict(n, {})
135 self.assertEqual(a.gen_inventory_lines_for_all_of_type(""), [])
137 def test_gen_inventory_lines(self):
138 n = ansible_common.AnsibleNode
139 a = ansible_common.AnsibleNodeDict(n, [{
140 "name": "name", "user": "user", "password": "PASS",
143 self.assertEqual(a.gen_all_inventory_lines(),
144 ["name ansible_ssh_pass=PASS ansible_user=user"])
147 class AnsibleCommonTestCase(unittest.TestCase):
148 def test_get_timeouts(self):
149 self.assertAlmostEqual(ansible_common.AnsibleCommon.get_timeout(-100), 1200.0)
151 def test__init__(self):
152 ansible_common.AnsibleCommon({})
154 def test_reset(self):
155 a = ansible_common.AnsibleCommon({})
158 def test_do_install_no_dir(self):
159 a = ansible_common.AnsibleCommon({})
160 self.assertRaises(OSError, a.do_install, '', '')
162 def test_gen_inventory_dict(self):
164 "name": "name", "user": "user", "password": "PASS",
167 a = ansible_common.AnsibleCommon(nodes)
168 a.gen_inventory_ini_dict()
169 self.assertEqual(a.inventory_dict, {
170 'nodes': ['name ansible_ssh_pass=PASS ansible_user=user'],
174 def test_deploy_dir(self):
175 a = ansible_common.AnsibleCommon({})
176 self.assertRaises(ValueError, getattr, a, "deploy_dir")
178 def test_deploy_dir_set(self):
179 a = ansible_common.AnsibleCommon({})
182 def test_deploy_dir_set_get(self):
183 a = ansible_common.AnsibleCommon({})
185 self.assertEqual(a.deploy_dir, "d")
187 @mock.patch('{}.open'.format(PREFIX))
188 def test__gen_ansible_playbook_file_list(self, _):
189 d = tempfile.mkdtemp()
191 a = ansible_common.AnsibleCommon({})
192 a._gen_ansible_playbook_file(["a"], d)
196 @mock.patch('{}.NamedTemporaryFile'.format(PREFIX))
197 @mock.patch('{}.open'.format(PREFIX))
198 def test__gen_ansible_inventory_file(self, _, __):
200 "name": "name", "user": "user", "password": "PASS",
203 d = tempfile.mkdtemp()
205 a = ansible_common.AnsibleCommon(nodes)
206 a.gen_inventory_ini_dict()
207 inv_context = a._gen_ansible_inventory_file(d)
210 inv_context.write_func(c)
211 self.assertIn("ansible_ssh_pass=PASS", c.getvalue())
215 @mock.patch('{}.NamedTemporaryFile'.format(PREFIX))
216 @mock.patch('{}.open'.format(PREFIX))
217 def test__gen_ansible_playbook_file_list_multiple(self, _, __):
218 d = tempfile.mkdtemp()
220 a = ansible_common.AnsibleCommon({})
221 a._gen_ansible_playbook_file(["a", "b"], d)
225 @mock.patch('{}.NamedTemporaryFile'.format(PREFIX))
226 @mock.patch('{}.Popen'.format(PREFIX))
227 @mock.patch('{}.open'.format(PREFIX))
228 def test_do_install_tmp_dir(self, _, mock_popen, __):
229 mock_popen.return_value.communicate.return_value = "", ""
230 mock_popen.return_value.wait.return_value = 0
231 d = tempfile.mkdtemp()
233 a = ansible_common.AnsibleCommon({})
238 @mock.patch('{}.NamedTemporaryFile'.format(PREFIX))
239 @mock.patch('{}.Popen'.format(PREFIX))
240 @mock.patch('{}.open'.format(PREFIX))
241 def test_execute_ansible_check(self, _, mock_popen, __):
242 mock_popen.return_value.communicate.return_value = "", ""
243 mock_popen.return_value.wait.return_value = 0
244 d = tempfile.mkdtemp()
246 a = ansible_common.AnsibleCommon({})
247 a.execute_ansible('', d, ansible_check=True, verbose=True)
251 def test_get_sut_info(self):
252 d = tempfile.mkdtemp()
253 a = ansible_common.AnsibleCommon({})
259 def test_get_sut_info_not_exist(self):
260 a = ansible_common.AnsibleCommon({})
262 a.get_sut_info('/hello/world')