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
  27 from yardstick.common import ansible_common
  28
  29 PREFIX = 'yardstick.common.ansible_common'
  30
  31
  32 class OverwriteDictTestCase(unittest.TestCase):
  33
  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         f = ansible_common.FileNameGenerator._handle_existing_file("/dev/null")
  54
  55     def test_get_generator_from_file(self):
  56         f = ansible_common.FileNameGenerator.get_generator_from_filename("/dev/null", "", "", "")
  57
  58     def test_get_generator_from_file_middle(self):
  59         f = ansible_common.FileNameGenerator.get_generator_from_filename("/dev/null", "", "",
  60                                                                          "null")
  61
  62     def test_get_generator_from_file_prefix(self):
  63         f = 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         a = 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         a = 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
 137 class AnsibleCommonTestCase(unittest.TestCase):
 138     def test_get_timeouts(self):
 139         self.assertAlmostEquals(ansible_common.AnsibleCommon.get_timeout(-100), 1200.0)
 140
 141     def test__init__(self):
 142         a = ansible_common.AnsibleCommon({})
 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         a = ansible_common.AnsibleCommon({})
 154         a.inventory_dict = {}
 155         self.assertIsNone(a.gen_inventory_ini_dict())
 156
 157     def test_deploy_dir(self):
 158         a = ansible_common.AnsibleCommon({})
 159         self.assertRaises(ValueError, getattr, a, "deploy_dir")
 160
 161     def test_deploy_dir_set(self):
 162         a = ansible_common.AnsibleCommon({})
 163         a.deploy_dir = ""
 164
 165     def test_deploy_dir_set_get(self):
 166         a = ansible_common.AnsibleCommon({})
 167         a.deploy_dir = "d"
 168         self.assertEqual(a.deploy_dir, "d")
 169
 170     @mock.patch('{}.open'.format(PREFIX))
 171     def test__gen_ansible_playbook_file_list(self, mock_open):
 172         d = tempfile.mkdtemp()
 173         try:
 174             a = ansible_common.AnsibleCommon({})
 175             a._gen_ansible_playbook_file(["a"], d)
 176         finally:
 177             os.rmdir(d)
 178
 179     @mock.patch('{}.NamedTemporaryFile'.format(PREFIX))
 180     @mock.patch('{}.open'.format(PREFIX))
 181     def test__gen_ansible_playbook_file_list_multiple(self, mock_open, mock_tmp):
 182         d = tempfile.mkdtemp()
 183         try:
 184             a = ansible_common.AnsibleCommon({})
 185             a._gen_ansible_playbook_file(["a", "b"], d)
 186         finally:
 187             os.rmdir(d)
 188
 189     @mock.patch('{}.NamedTemporaryFile'.format(PREFIX))
 190     @mock.patch('{}.Popen'.format(PREFIX))
 191     @mock.patch('{}.open'.format(PREFIX))
 192     def test_do_install_tmp_dir(self, mock_open, mock_popen, mock_tmp):
 193         mock_popen.return_value.communicate.return_value = "", ""
 194         mock_popen.return_value.wait.return_value = 0
 195         d = tempfile.mkdtemp()
 196         try:
 197             a = ansible_common.AnsibleCommon({})
 198             a.do_install('', d)
 199         finally:
 200             os.rmdir(d)
 201
 202     @mock.patch('{}.NamedTemporaryFile'.format(PREFIX))
 203     @mock.patch('{}.Popen'.format(PREFIX))
 204     @mock.patch('{}.open'.format(PREFIX))
 205     def test_execute_ansible_check(self, mock_open, mock_popen, mock_tmp):
 206         mock_popen.return_value.communicate.return_value = "", ""
 207         mock_popen.return_value.wait.return_value = 0
 208         d = tempfile.mkdtemp()
 209         try:
 210             a = ansible_common.AnsibleCommon({})
 211             a.execute_ansible('', d, ansible_check=True, verbose=True)
 212         finally:
 213             os.rmdir(d)