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
20 from collections import defaultdict
25 from six.moves.configparser import ConfigParser
26 from six.moves import StringIO
28 from yardstick.common import ansible_common
30 PREFIX = 'yardstick.common.ansible_common'
33 class OverwriteDictTestCase(unittest.TestCase):
34 def test_overwrite_dict_cfg(self):
35 c = ConfigParser(allow_no_value=True)
37 "section_a": "empty_value",
38 "section_b": {"key_c": "Val_d", "key_d": "VAL_D"},
39 "section_c": ["key_c", "key_d"],
41 ansible_common.overwrite_dict_to_cfg(c, d)
42 # Python3 and Python2 convert empty values into None or ''
43 # we don't really care but we need to compare correctly for unittest
44 self.assertTrue(c.has_option("section_a", "empty_value"))
45 self.assertEqual(sorted(c.items("section_b")), [('key_c', 'Val_d'), ('key_d', 'VAL_D')])
46 self.assertTrue(c.has_option("section_c", "key_c"))
47 self.assertTrue(c.has_option("section_c", "key_d"))
50 class FilenameGeneratorTestCase(unittest.TestCase):
51 @mock.patch('{}.NamedTemporaryFile'.format(PREFIX))
52 def test__handle_existing_file(self, _):
53 ansible_common.FileNameGenerator._handle_existing_file("/dev/null")
55 def test_get_generator_from_file(self):
56 ansible_common.FileNameGenerator.get_generator_from_filename("/dev/null", "", "", "")
58 def test_get_generator_from_file_middle(self):
59 ansible_common.FileNameGenerator.get_generator_from_filename("/dev/null", "", "",
62 def test_get_generator_from_file_prefix(self):
63 ansible_common.FileNameGenerator.get_generator_from_filename("/dev/null", "", "null",
67 class AnsibleNodeTestCase(unittest.TestCase):
68 def test_ansible_node(self):
69 ansible_common.AnsibleNode()
71 def test_ansible_node_len(self):
72 a = ansible_common.AnsibleNode()
75 def test_ansible_node_repr(self):
76 a = ansible_common.AnsibleNode()
79 def test_ansible_node_iter(self):
80 a = ansible_common.AnsibleNode()
84 def test_is_role(self):
85 a = ansible_common.AnsibleNode()
86 self.assertFalse(a.is_role("", default="foo"))
88 def test_ansible_node_get_tuple(self):
89 a = ansible_common.AnsibleNode({"name": "name"})
90 self.assertEqual(a.get_tuple(), ('name', a))
92 def test_gen_inventory_line(self):
93 a = ansible_common.AnsibleNode(defaultdict(str))
94 self.assertEqual(a.gen_inventory_line(), "")
96 def test_ansible_node_delitem(self):
97 a = ansible_common.AnsibleNode({"name": "name"})
100 def test_ansible_node_getattr(self):
101 a = ansible_common.AnsibleNode({"name": "name"})
102 self.assertEqual(getattr(a, "nosuch", None), None)
105 class AnsibleNodeDictTestCase(unittest.TestCase):
106 def test_ansible_node_dict(self):
107 n = ansible_common.AnsibleNode
108 ansible_common.AnsibleNodeDict(n, {})
110 def test_ansible_node_dict_len(self):
111 n = ansible_common.AnsibleNode
112 a = ansible_common.AnsibleNodeDict(n, {})
115 def test_ansible_node_dict_repr(self):
116 n = ansible_common.AnsibleNode
117 a = ansible_common.AnsibleNodeDict(n, {})
120 def test_ansible_node_dict_iter(self):
121 n = ansible_common.AnsibleNode
122 a = ansible_common.AnsibleNodeDict(n, {})
126 def test_ansible_node_dict_get(self):
127 n = ansible_common.AnsibleNode
128 a = ansible_common.AnsibleNodeDict(n, {})
129 self.assertIsNone(a.get(""))
131 def test_gen_inventory_lines_for_all_of_type(self):
132 n = ansible_common.AnsibleNode
133 a = ansible_common.AnsibleNodeDict(n, {})
134 self.assertEqual(a.gen_inventory_lines_for_all_of_type(""), [])
136 def test_gen_inventory_lines(self):
137 n = ansible_common.AnsibleNode
138 a = ansible_common.AnsibleNodeDict(n, [{
139 "name": "name", "user": "user", "password": "PASS",
142 self.assertEqual(a.gen_all_inventory_lines(),
143 ["name ansible_ssh_pass=PASS ansible_user=user"])
146 class AnsibleCommonTestCase(unittest.TestCase):
147 def test_get_timeouts(self):
148 self.assertAlmostEquals(ansible_common.AnsibleCommon.get_timeout(-100), 1200.0)
150 def test__init__(self):
151 ansible_common.AnsibleCommon({})
153 def test_reset(self):
154 a = ansible_common.AnsibleCommon({})
157 def test_do_install_no_dir(self):
158 a = ansible_common.AnsibleCommon({})
159 self.assertRaises(OSError, a.do_install, '', '')
161 def test_gen_inventory_dict(self):
163 "name": "name", "user": "user", "password": "PASS",
166 a = ansible_common.AnsibleCommon(nodes)
167 a.gen_inventory_ini_dict()
168 self.assertEqual(a.inventory_dict, {
169 'nodes': ['name ansible_ssh_pass=PASS ansible_user=user'],
173 def test_deploy_dir(self):
174 a = ansible_common.AnsibleCommon({})
175 self.assertRaises(ValueError, getattr, a, "deploy_dir")
177 def test_deploy_dir_set(self):
178 a = ansible_common.AnsibleCommon({})
181 def test_deploy_dir_set_get(self):
182 a = ansible_common.AnsibleCommon({})
184 self.assertEqual(a.deploy_dir, "d")
186 @mock.patch('{}.open'.format(PREFIX))
187 def test__gen_ansible_playbook_file_list(self, _):
188 d = tempfile.mkdtemp()
190 a = ansible_common.AnsibleCommon({})
191 a._gen_ansible_playbook_file(["a"], d)
195 @mock.patch('{}.NamedTemporaryFile'.format(PREFIX))
196 @mock.patch('{}.open'.format(PREFIX))
197 def test__gen_ansible_inventory_file(self, _, __):
199 "name": "name", "user": "user", "password": "PASS",
202 d = tempfile.mkdtemp()
204 a = ansible_common.AnsibleCommon(nodes)
205 a.gen_inventory_ini_dict()
206 inv_context = a._gen_ansible_inventory_file(d)
209 inv_context.write_func(c)
210 self.assertIn("ansible_ssh_pass=PASS", c.getvalue())
214 @mock.patch('{}.NamedTemporaryFile'.format(PREFIX))
215 @mock.patch('{}.open'.format(PREFIX))
216 def test__gen_ansible_playbook_file_list_multiple(self, _, __):
217 d = tempfile.mkdtemp()
219 a = ansible_common.AnsibleCommon({})
220 a._gen_ansible_playbook_file(["a", "b"], d)
224 @mock.patch('{}.NamedTemporaryFile'.format(PREFIX))
225 @mock.patch('{}.Popen'.format(PREFIX))
226 @mock.patch('{}.open'.format(PREFIX))
227 def test_do_install_tmp_dir(self, _, mock_popen, __):
228 mock_popen.return_value.communicate.return_value = "", ""
229 mock_popen.return_value.wait.return_value = 0
230 d = tempfile.mkdtemp()
232 a = ansible_common.AnsibleCommon({})
237 @mock.patch('{}.NamedTemporaryFile'.format(PREFIX))
238 @mock.patch('{}.Popen'.format(PREFIX))
239 @mock.patch('{}.open'.format(PREFIX))
240 def test_execute_ansible_check(self, _, mock_popen, __):
241 mock_popen.return_value.communicate.return_value = "", ""
242 mock_popen.return_value.wait.return_value = 0
243 d = tempfile.mkdtemp()
245 a = ansible_common.AnsibleCommon({})
246 a.execute_ansible('', d, ansible_check=True, verbose=True)