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.
22 from six.moves import configparser
24 from yardstick.common import ansible_common
25 from yardstick.tests.unit import base as ut_base
28 class OverwriteDictTestCase(ut_base.BaseUnitTestCase):
30 def test_overwrite_dict_cfg(self):
31 c = configparser.ConfigParser(allow_no_value=True)
33 "section_a": "empty_value",
34 "section_b": {"key_c": "Val_d", "key_d": "VAL_D"},
35 "section_c": ["key_c", "key_d"],
37 ansible_common.overwrite_dict_to_cfg(c, d)
38 # Python3 and Python2 convert empty values into None or ''
39 # we don't really care but we need to compare correctly for unittest
40 self.assertTrue(c.has_option("section_a", "empty_value"))
41 self.assertEqual(sorted(c.items("section_b")),
42 [('key_c', 'Val_d'), ('key_d', 'VAL_D')])
43 self.assertTrue(c.has_option("section_c", "key_c"))
44 self.assertTrue(c.has_option("section_c", "key_d"))
47 class FilenameGeneratorTestCase(ut_base.BaseUnitTestCase):
49 @mock.patch.object(tempfile, 'NamedTemporaryFile')
50 def test__handle_existing_file(self, _):
51 ansible_common.FileNameGenerator._handle_existing_file('/dev/null')
53 def test_get_generator_from_file(self):
54 ansible_common.FileNameGenerator.get_generator_from_filename(
55 '/dev/null', '', '', '')
57 def test_get_generator_from_file_middle(self):
58 ansible_common.FileNameGenerator.get_generator_from_filename(
59 '/dev/null', '', '', 'null')
61 def test_get_generator_from_file_prefix(self):
62 ansible_common.FileNameGenerator.get_generator_from_filename(
63 '/dev/null', '', 'null', 'middle')
66 class AnsibleNodeTestCase(ut_base.BaseUnitTestCase):
68 def test_ansible_node_len(self):
69 self.assertEqual(0, len(ansible_common.AnsibleNode()))
71 def test_ansible_node_repr(self):
72 self.assertEqual('AnsibleNode<{}>', repr(ansible_common.AnsibleNode()))
74 def test_ansible_node_iter(self):
75 node = ansible_common.AnsibleNode(data={'a': 1, 'b': 2, 'c': 3})
77 self.assertIn(key, ('a', 'b', 'c'))
79 def test_is_role(self):
80 node = ansible_common.AnsibleNode()
81 self.assertFalse(node.is_role('', default='foo'))
83 def test_ansible_node_get_tuple(self):
84 node = ansible_common.AnsibleNode({'name': 'name'})
85 self.assertEqual(node.get_tuple(), ('name', node))
87 def test_gen_inventory_line(self):
88 a = ansible_common.AnsibleNode(collections.defaultdict(str))
89 self.assertEqual(a.gen_inventory_line(), "")
91 def test_ansible_node_delitem(self):
92 node = ansible_common.AnsibleNode({'name': 'name'})
93 self.assertEqual(1, len(node))
95 self.assertEqual(0, len(node))
97 def test_ansible_node_getattr(self):
98 node = ansible_common.AnsibleNode({'name': 'name'})
99 self.assertIsNone(getattr(node, 'nosuch', None))
102 class AnsibleNodeDictTestCase(ut_base.BaseUnitTestCase):
104 def test_ansible_node_dict_len(self):
105 n = ansible_common.AnsibleNode
106 a = ansible_common.AnsibleNodeDict(n, {})
107 self.assertEqual(0, len(a))
109 def test_ansible_node_dict_repr(self):
110 n = ansible_common.AnsibleNode
111 a = ansible_common.AnsibleNodeDict(n, {})
112 self.assertEqual('{}', repr(a))
114 def test_ansible_node_dict_get(self):
115 n = ansible_common.AnsibleNode
116 a = ansible_common.AnsibleNodeDict(n, {})
117 self.assertIsNone(a.get(""))
119 def test_gen_inventory_lines_for_all_of_type(self):
120 n = ansible_common.AnsibleNode
121 a = ansible_common.AnsibleNodeDict(n, {})
122 self.assertEqual(a.gen_inventory_lines_for_all_of_type(""), [])
124 def test_gen_inventory_lines(self):
125 n = ansible_common.AnsibleNode
126 a = ansible_common.AnsibleNodeDict(n, [{
127 "name": "name", "user": "user", "password": "PASS",
130 self.assertEqual(a.gen_all_inventory_lines(),
131 ["name ansible_ssh_pass=PASS ansible_user=user"])
134 class AnsibleCommonTestCase(ut_base.BaseUnitTestCase):
137 def _delete_tmpdir(dir):
140 def test_get_timeouts(self):
141 self.assertAlmostEqual(
142 ansible_common.AnsibleCommon.get_timeout(-100), 1200.0)
144 def test_reset(self):
145 a = ansible_common.AnsibleCommon({})
148 def test_do_install_no_dir(self):
149 a = ansible_common.AnsibleCommon({})
150 self.assertRaises(OSError, a.do_install, '', '')
152 def test_gen_inventory_dict(self):
154 "name": "name", "user": "user", "password": "PASS",
157 a = ansible_common.AnsibleCommon(nodes)
158 a.gen_inventory_ini_dict()
159 self.assertEqual(a.inventory_dict, {
160 'nodes': ['name ansible_ssh_pass=PASS ansible_user=user'],
164 def test_deploy_dir(self):
165 a = ansible_common.AnsibleCommon({})
166 self.assertRaises(ValueError, getattr, a, "deploy_dir")
168 def test_deploy_dir_set(self):
169 a = ansible_common.AnsibleCommon({})
172 def test_deploy_dir_set_get(self):
173 a = ansible_common.AnsibleCommon({})
175 self.assertEqual(a.deploy_dir, "d")
177 @mock.patch.object(moves.builtins, 'open')
178 def test__gen_ansible_playbook_file_list(self, *args):
179 d = tempfile.mkdtemp()
180 self.addCleanup(self._delete_tmpdir, d)
181 a = ansible_common.AnsibleCommon({})
182 a._gen_ansible_playbook_file(["a"], d)
184 @mock.patch.object(tempfile, 'NamedTemporaryFile')
185 @mock.patch.object(moves.builtins, 'open')
186 def test__gen_ansible_inventory_file(self, *args):
188 "name": "name", "user": "user", "password": "PASS",
191 d = tempfile.mkdtemp()
192 self.addCleanup(self._delete_tmpdir, d)
193 a = ansible_common.AnsibleCommon(nodes)
194 a.gen_inventory_ini_dict()
195 inv_context = a._gen_ansible_inventory_file(d)
198 inv_context.write_func(c)
199 self.assertIn("ansible_ssh_pass=PASS", c.getvalue())
201 @mock.patch.object(tempfile, 'NamedTemporaryFile')
202 @mock.patch.object(moves.builtins, 'open')
203 def test__gen_ansible_playbook_file_list_multiple(self, *args):
204 d = tempfile.mkdtemp()
205 self.addCleanup(self._delete_tmpdir, d)
206 a = ansible_common.AnsibleCommon({})
207 a._gen_ansible_playbook_file(["a", "b"], d)
209 @mock.patch.object(tempfile, 'NamedTemporaryFile')
210 @mock.patch.object(subprocess, 'Popen')
211 @mock.patch.object(moves.builtins, 'open')
212 def test_do_install_tmp_dir(self, _, mock_popen, *args):
213 mock_popen.return_value.communicate.return_value = "", ""
214 mock_popen.return_value.wait.return_value = 0
215 d = tempfile.mkdtemp()
216 self.addCleanup(self._delete_tmpdir, d)
217 a = ansible_common.AnsibleCommon({})
220 @mock.patch.object(tempfile, 'NamedTemporaryFile')
221 @mock.patch.object(moves.builtins, 'open')
222 @mock.patch.object(subprocess, 'Popen')
223 def test_execute_ansible_check(self, mock_popen, *args):
224 mock_popen.return_value.communicate.return_value = "", ""
225 mock_popen.return_value.wait.return_value = 0
226 d = tempfile.mkdtemp()
227 self.addCleanup(self._delete_tmpdir, d)
228 a = ansible_common.AnsibleCommon({})
229 a.execute_ansible('', d, ansible_check=True, verbose=True)
231 def test_get_sut_info(self):
232 d = tempfile.mkdtemp()
233 a = ansible_common.AnsibleCommon({})
234 self.addCleanup(self._delete_tmpdir, d)
235 with mock.patch.object(a, '_exec_get_sut_info_cmd'):
238 def test_get_sut_info_not_exist(self):
239 a = ansible_common.AnsibleCommon({})
240 with self.assertRaises(OSError):
241 a.get_sut_info('/hello/world')