yardstick/tests/unit/common/test_ansible_common.py

   1 # Copyright (c) 2016-2017 Intel Corporation
   2 #
   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
   6 #
   7 #      http://www.apache.org/licenses/LICENSE-2.0
   8 #
   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.
  14
  15 import collections
  16 import shutil
  17 import subprocess
  18 import tempfile
  19
  20 import mock
  21 from six import moves
  22 from six.moves import configparser
  23
  24 from yardstick.common import ansible_common
  25 from yardstick.tests.unit import base as ut_base
  26
  27
  28 class OverwriteDictTestCase(ut_base.BaseUnitTestCase):
  29
  30     def test_overwrite_dict_cfg(self):
  31         c = configparser.ConfigParser(allow_no_value=True)
  32         d = {
  33             "section_a": "empty_value",
  34             "section_b": {"key_c": "Val_d", "key_d": "VAL_D"},
  35             "section_c": ["key_c", "key_d"],
  36         }
  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"))
  45
  46
  47 class FilenameGeneratorTestCase(ut_base.BaseUnitTestCase):
  48
  49     @mock.patch.object(tempfile, 'NamedTemporaryFile')
  50     def test__handle_existing_file(self, _):
  51         ansible_common.FileNameGenerator._handle_existing_file('/dev/null')
  52
  53     def test_get_generator_from_file(self):
  54         ansible_common.FileNameGenerator.get_generator_from_filename(
  55             '/dev/null', '', '', '')
  56
  57     def test_get_generator_from_file_middle(self):
  58         ansible_common.FileNameGenerator.get_generator_from_filename(
  59             '/dev/null', '', '', 'null')
  60
  61     def test_get_generator_from_file_prefix(self):
  62         ansible_common.FileNameGenerator.get_generator_from_filename(
  63             '/dev/null', '', 'null', 'middle')
  64
  65
  66 class AnsibleNodeTestCase(ut_base.BaseUnitTestCase):
  67
  68     def test_ansible_node_len(self):
  69         self.assertEqual(0, len(ansible_common.AnsibleNode()))
  70
  71     def test_ansible_node_repr(self):
  72         self.assertEqual('AnsibleNode<{}>', repr(ansible_common.AnsibleNode()))
  73
  74     def test_ansible_node_iter(self):
  75         node = ansible_common.AnsibleNode(data={'a': 1, 'b': 2, 'c': 3})
  76         for key in node:
  77             self.assertIn(key, ('a', 'b', 'c'))
  78
  79     def test_is_role(self):
  80         node = ansible_common.AnsibleNode()
  81         self.assertFalse(node.is_role('', default='foo'))
  82
  83     def test_ansible_node_get_tuple(self):
  84         node = ansible_common.AnsibleNode({'name': 'name'})
  85         self.assertEqual(node.get_tuple(), ('name', node))
  86
  87     def test_gen_inventory_line(self):
  88         a = ansible_common.AnsibleNode(collections.defaultdict(str))
  89         self.assertEqual(a.gen_inventory_line(), "")
  90
  91     def test_ansible_node_delitem(self):
  92         node = ansible_common.AnsibleNode({'name': 'name'})
  93         self.assertEqual(1, len(node))
  94         del node['name']
  95         self.assertEqual(0, len(node))
  96
  97     def test_ansible_node_getattr(self):
  98         node = ansible_common.AnsibleNode({'name': 'name'})
  99         self.assertIsNone(getattr(node, 'nosuch', None))
 100
 101
 102 class AnsibleNodeDictTestCase(ut_base.BaseUnitTestCase):
 103
 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))
 108
 109     def test_ansible_node_dict_repr(self):
 110         n = ansible_common.AnsibleNode
 111         a = ansible_common.AnsibleNodeDict(n, {})
 112         self.assertEqual('{}', repr(a))
 113
 114     def test_ansible_node_dict_get(self):
 115         n = ansible_common.AnsibleNode
 116         a = ansible_common.AnsibleNodeDict(n, {})
 117         self.assertIsNone(a.get(""))
 118
 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(""), [])
 123
 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",
 128             "role": "role",
 129         }])
 130         self.assertEqual(a.gen_all_inventory_lines(),
 131                          ["name ansible_ssh_pass=PASS ansible_user=user"])
 132
 133
 134 class AnsibleCommonTestCase(ut_base.BaseUnitTestCase):
 135
 136     @staticmethod
 137     def _delete_tmpdir(dir):
 138         shutil.rmtree(dir)
 139
 140     def test_get_timeouts(self):
 141         self.assertAlmostEqual(
 142             ansible_common.AnsibleCommon.get_timeout(-100), 1200.0)
 143
 144     def test_reset(self):
 145         a = ansible_common.AnsibleCommon({})
 146         a.reset()
 147
 148     def test_do_install_no_dir(self):
 149         a = ansible_common.AnsibleCommon({})
 150         self.assertRaises(OSError, a.do_install, '', '')
 151
 152     def test_gen_inventory_dict(self):
 153         nodes = [{
 154             "name": "name", "user": "user", "password": "PASS",
 155             "role": "role",
 156         }]
 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'],
 161             'role': ['name']
 162         })
 163
 164     def test_deploy_dir(self):
 165         a = ansible_common.AnsibleCommon({})
 166         self.assertRaises(ValueError, getattr, a, "deploy_dir")
 167
 168     def test_deploy_dir_set(self):
 169         a = ansible_common.AnsibleCommon({})
 170         a.deploy_dir = ""
 171
 172     def test_deploy_dir_set_get(self):
 173         a = ansible_common.AnsibleCommon({})
 174         a.deploy_dir = "d"
 175         self.assertEqual(a.deploy_dir, "d")
 176
 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)
 183
 184     @mock.patch.object(tempfile, 'NamedTemporaryFile')
 185     @mock.patch.object(moves.builtins, 'open')
 186     def test__gen_ansible_inventory_file(self, *args):
 187         nodes = [{
 188             "name": "name", "user": "user", "password": "PASS",
 189             "role": "role",
 190         }]
 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)
 196         with inv_context:
 197             c = moves.StringIO()
 198             inv_context.write_func(c)
 199             self.assertIn("ansible_ssh_pass=PASS", c.getvalue())
 200
 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)
 208
 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({})
 218         a.do_install('', d)
 219
 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)
 230
 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'):
 236             a.get_sut_info(d)
 237
 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')