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
  16 from __future__ import absolute_import
  17
  18 import os
  19 import tempfile
  20 import shutil
  21 from collections import defaultdict
  22
  23 import mock
  24 import unittest
  25
  26 from six.moves.configparser import ConfigParser
  27 from six.moves import StringIO
  28
  29 from yardstick.common import ansible_common
  30
  31 PREFIX = 'yardstick.common.ansible_common'
  32
  33
  34 class OverwriteDictTestCase(unittest.TestCase):
  35     def test_overwrite_dict_cfg(self):
  36         c = ConfigParser(allow_no_value=True)
  37         d = {
  38             "section_a": "empty_value",
  39             "section_b": {"key_c": "Val_d", "key_d": "VAL_D"},
  40             "section_c": ["key_c", "key_d"],
  41         }
  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"))
  49
  50
  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")
  55
  56     def test_get_generator_from_file(self):
  57         ansible_common.FileNameGenerator.get_generator_from_filename("/dev/null", "", "", "")
  58
  59     def test_get_generator_from_file_middle(self):
  60         ansible_common.FileNameGenerator.get_generator_from_filename("/dev/null", "", "",
  61                                                                      "null")
  62
  63     def test_get_generator_from_file_prefix(self):
  64         ansible_common.FileNameGenerator.get_generator_from_filename("/dev/null", "", "null",
  65                                                                      "middle")
  66
  67
  68 class AnsibleNodeTestCase(unittest.TestCase):
  69     def test_ansible_node(self):
  70         ansible_common.AnsibleNode()
  71
  72     def test_ansible_node_len(self):
  73         a = ansible_common.AnsibleNode()
  74         len(a)
  75
  76     def test_ansible_node_repr(self):
  77         a = ansible_common.AnsibleNode()
  78         repr(a)
  79
  80     def test_ansible_node_iter(self):
  81         a = ansible_common.AnsibleNode()
  82         for _ in a:
  83             pass
  84
  85     def test_is_role(self):
  86         a = ansible_common.AnsibleNode()
  87         self.assertFalse(a.is_role("", default="foo"))
  88
  89     def test_ansible_node_get_tuple(self):
  90         a = ansible_common.AnsibleNode({"name": "name"})
  91         self.assertEqual(a.get_tuple(), ('name', a))
  92
  93     def test_gen_inventory_line(self):
  94         a = ansible_common.AnsibleNode(defaultdict(str))
  95         self.assertEqual(a.gen_inventory_line(), "")
  96
  97     def test_ansible_node_delitem(self):
  98         a = ansible_common.AnsibleNode({"name": "name"})
  99         del a['name']
 100
 101     def test_ansible_node_getattr(self):
 102         a = ansible_common.AnsibleNode({"name": "name"})
 103         self.assertIsNone(getattr(a, "nosuch", None))
 104
 105
 106 class AnsibleNodeDictTestCase(unittest.TestCase):
 107     def test_ansible_node_dict(self):
 108         n = ansible_common.AnsibleNode
 109         ansible_common.AnsibleNodeDict(n, {})
 110
 111     def test_ansible_node_dict_len(self):
 112         n = ansible_common.AnsibleNode
 113         a = ansible_common.AnsibleNodeDict(n, {})
 114         len(a)
 115
 116     def test_ansible_node_dict_repr(self):
 117         n = ansible_common.AnsibleNode
 118         a = ansible_common.AnsibleNodeDict(n, {})
 119         repr(a)
 120
 121     def test_ansible_node_dict_iter(self):
 122         n = ansible_common.AnsibleNode
 123         a = ansible_common.AnsibleNodeDict(n, {})
 124         for _ in a:
 125             pass
 126
 127     def test_ansible_node_dict_get(self):
 128         n = ansible_common.AnsibleNode
 129         a = ansible_common.AnsibleNodeDict(n, {})
 130         self.assertIsNone(a.get(""))
 131
 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(""), [])
 136
 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",
 141             "role": "role",
 142         }])
 143         self.assertEqual(a.gen_all_inventory_lines(),
 144                          ["name ansible_ssh_pass=PASS ansible_user=user"])
 145
 146
 147 class AnsibleCommonTestCase(unittest.TestCase):
 148     def test_get_timeouts(self):
 149         self.assertAlmostEqual(ansible_common.AnsibleCommon.get_timeout(-100), 1200.0)
 150
 151     def test__init__(self):
 152         ansible_common.AnsibleCommon({})
 153
 154     def test_reset(self):
 155         a = ansible_common.AnsibleCommon({})
 156         a.reset()
 157
 158     def test_do_install_no_dir(self):
 159         a = ansible_common.AnsibleCommon({})
 160         self.assertRaises(OSError, a.do_install, '', '')
 161
 162     def test_gen_inventory_dict(self):
 163         nodes = [{
 164             "name": "name", "user": "user", "password": "PASS",
 165             "role": "role",
 166         }]
 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'],
 171             'role': ['name']
 172         })
 173
 174     def test_deploy_dir(self):
 175         a = ansible_common.AnsibleCommon({})
 176         self.assertRaises(ValueError, getattr, a, "deploy_dir")
 177
 178     def test_deploy_dir_set(self):
 179         a = ansible_common.AnsibleCommon({})
 180         a.deploy_dir = ""
 181
 182     def test_deploy_dir_set_get(self):
 183         a = ansible_common.AnsibleCommon({})
 184         a.deploy_dir = "d"
 185         self.assertEqual(a.deploy_dir, "d")
 186
 187     @mock.patch('{}.open'.format(PREFIX))
 188     def test__gen_ansible_playbook_file_list(self, _):
 189         d = tempfile.mkdtemp()
 190         try:
 191             a = ansible_common.AnsibleCommon({})
 192             a._gen_ansible_playbook_file(["a"], d)
 193         finally:
 194             os.rmdir(d)
 195
 196     @mock.patch('{}.NamedTemporaryFile'.format(PREFIX))
 197     @mock.patch('{}.open'.format(PREFIX))
 198     def test__gen_ansible_inventory_file(self, _, __):
 199         nodes = [{
 200             "name": "name", "user": "user", "password": "PASS",
 201             "role": "role",
 202         }]
 203         d = tempfile.mkdtemp()
 204         try:
 205             a = ansible_common.AnsibleCommon(nodes)
 206             a.gen_inventory_ini_dict()
 207             inv_context = a._gen_ansible_inventory_file(d)
 208             with inv_context:
 209                 c = StringIO()
 210                 inv_context.write_func(c)
 211                 self.assertIn("ansible_ssh_pass=PASS", c.getvalue())
 212         finally:
 213             os.rmdir(d)
 214
 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()
 219         try:
 220             a = ansible_common.AnsibleCommon({})
 221             a._gen_ansible_playbook_file(["a", "b"], d)
 222         finally:
 223             os.rmdir(d)
 224
 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()
 232         try:
 233             a = ansible_common.AnsibleCommon({})
 234             a.do_install('', d)
 235         finally:
 236             os.rmdir(d)
 237
 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()
 245         try:
 246             a = ansible_common.AnsibleCommon({})
 247             a.execute_ansible('', d, ansible_check=True, verbose=True)
 248         finally:
 249             os.rmdir(d)
 250
 251     def test_get_sut_info(self):
 252         d = tempfile.mkdtemp()
 253         a = ansible_common.AnsibleCommon({})
 254         try:
 255             a.get_sut_info(d)
 256         finally:
 257             shutil.rmtree(d)
 258
 259     def test_get_sut_info_not_exist(self):
 260         a = ansible_common.AnsibleCommon({})
 261         try:
 262             a.get_sut_info('/hello/world')
 263         except OSError:
 264             pass