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