functest/tests/unit/ci/test_check_deployment.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2017 Ericsson and others.
   4 #
   5 # All rights reserved. This program and the accompanying materials
   6 # are made available under the terms of the Apache License, Version 2.0
   7 # which accompanies this distribution, and is available at
   8 # http://www.apache.org/licenses/LICENSE-2.0
   9
  10 # pylint: disable=missing-docstring
  11
  12 import socket
  13 import unittest
  14
  15 import logging
  16 import mock
  17
  18 from functest.ci import check_deployment
  19
  20 __author__ = "Jose Lausuch <jose.lausuch@ericsson.com>"
  21
  22
  23 class CheckDeploymentTesting(unittest.TestCase):
  24     """The super class which testing classes could inherit."""
  25     # pylint: disable=missing-docstring,too-many-public-methods
  26
  27     logging.disable(logging.CRITICAL)
  28
  29     def setUp(self):
  30         self.client_test = mock.Mock()
  31         self.deployment = check_deployment.CheckDeployment()
  32         self.service_test = 'compute'
  33         self.rc_file = self.deployment.rc_file
  34         self.endpoint_test = 'http://192.168.0.6:5000/v3'
  35         creds_attr = {'auth_url': self.endpoint_test,
  36                       'proxy_settings': ''}
  37         proxy_attr = {'host': '192.168.0.1', 'port': '5000'}
  38         proxy_settings = mock.Mock()
  39         proxy_settings.configure_mock(**proxy_attr)
  40         self.os_creds = mock.Mock()
  41         self.os_creds.configure_mock(**creds_attr)
  42         self.os_creds.proxy_settings = proxy_settings
  43         self.deployment.os_creds = self.os_creds
  44
  45     @mock.patch('socket.socket.connect', side_effect=TypeError)
  46     def test_verify_connectivity_ko1(self, *args):
  47         self.assertFalse(check_deployment.verify_connectivity("127.0.0.1"))
  48         args[0].assert_called_once_with(None, 80)
  49
  50     @mock.patch('socket.socket.connect', side_effect=socket.error)
  51     def test_verify_connectivity_ko2(self, *args):
  52         self.assertFalse(
  53             check_deployment.verify_connectivity("http://127.0.0.1"))
  54         args[0].assert_called_once_with("127.0.0.1", 80)
  55
  56     @mock.patch('socket.socket.connect', side_effect=socket.error)
  57     def test_verify_connectivity_ko3(self, *args):
  58         self.assertFalse(
  59             check_deployment.verify_connectivity("https://127.0.0.1"))
  60         args[0].assert_called_once_with("127.0.0.1", 443)
  61
  62     @mock.patch('socket.socket.connect')
  63     def test_verify_connectivity(self, *args):
  64         self.assertTrue(
  65             check_deployment.verify_connectivity("https://127.0.0.1"))
  66         args[0].assert_called_once_with("127.0.0.1", 443)
  67
  68     @mock.patch('snaps.openstack.utils.keystone_utils.keystone_session',
  69                 return_value=mock.Mock(
  70                     get_token=mock.Mock(side_effect=Exception)))
  71     def test_get_auth_token_ko(self, *args):
  72         with self.assertRaises(Exception):
  73             check_deployment.get_auth_token(self.os_creds)
  74         args[0].assert_called_once_with(self.os_creds)
  75
  76     @mock.patch('snaps.openstack.utils.keystone_utils.keystone_session',
  77                 return_value=mock.Mock(
  78                     get_token=mock.Mock(return_value="foo")))
  79     def test_get_auth_token(self, *args):
  80         self.assertEqual(check_deployment.get_auth_token(self.os_creds), "foo")
  81         args[0].assert_called_once_with(self.os_creds)
  82
  83     @mock.patch('six.moves.builtins.open',
  84                 mock.mock_open(read_data='OS_AUTH_URL'))
  85     @mock.patch('functest.ci.check_deployment.os.path.isfile', returns=True)
  86     def test_check_rc(self, *args):
  87         self.deployment.check_rc()
  88         args[0].assert_called_once_with(self.rc_file)
  89
  90     @mock.patch('functest.ci.check_deployment.os.path.isfile',
  91                 return_value=False)
  92     def test_check_rc_missing_file(self, *args):
  93         with self.assertRaises(Exception) as context:
  94             self.deployment.check_rc()
  95         args[0].assert_called_once_with(self.rc_file)
  96         msg = 'RC file {} does not exist!'.format(self.rc_file)
  97         self.assertTrue(msg in str(context.exception))
  98
  99     @mock.patch('six.moves.builtins.open',
 100                 mock.mock_open(read_data='test'))
 101     @mock.patch('functest.ci.check_deployment.os.path.isfile',
 102                 return_value=True)
 103     def test_check_rc_missing_os_auth(self, *args):
 104         with self.assertRaises(Exception) as context:
 105             self.deployment.check_rc()
 106         args[0].assert_called_once_with(self.rc_file)
 107         msg = 'OS_AUTH_URL not defined in {}.'.format(self.rc_file)
 108         self.assertTrue(msg in str(context.exception))
 109
 110     @mock.patch('functest.ci.check_deployment.get_auth_token',
 111                 return_value='gAAAAABaOhXGS')
 112     @mock.patch('functest.ci.check_deployment.verify_connectivity',
 113                 return_value=True)
 114     def test_check_auth_endpoint(self, *args):
 115         self.deployment.check_auth_endpoint()
 116         args[0].assert_called_once_with(self.endpoint_test)
 117         args[1].assert_called_once_with(mock.ANY)
 118
 119     @mock.patch('functest.ci.check_deployment.verify_connectivity',
 120                 return_value=False)
 121     def test_check_auth_endpoint_ko(self, *args):
 122         with self.assertRaises(Exception) as context:
 123             self.deployment.check_auth_endpoint()
 124         msg = "OS_AUTH_URL {} is not reachable.".format(self.os_creds.auth_url)
 125         args[0].assert_called_once_with(self.os_creds.auth_url)
 126         self.assertTrue(msg in str(context.exception))
 127
 128     @mock.patch('functest.ci.check_deployment.verify_connectivity',
 129                 return_value=True)
 130     @mock.patch('functest.ci.check_deployment.keystone_utils.get_endpoint')
 131     def test_check_public_endpoint(self, *args):
 132         args[0].return_value = self.endpoint_test
 133         self.deployment.check_public_endpoint()
 134         args[0].assert_called_once_with(
 135             mock.ANY, 'identity', interface='public')
 136         args[1].assert_called_once_with(self.endpoint_test)
 137
 138     @mock.patch('functest.ci.check_deployment.verify_connectivity',
 139                 return_value=False)
 140     @mock.patch('functest.ci.check_deployment.keystone_utils.get_endpoint')
 141     def test_check_public_endpoint_ko(self, *args):
 142         args[0].return_value = self.endpoint_test
 143         with self.assertRaises(Exception) as context:
 144             self.deployment.check_public_endpoint()
 145         args[0].assert_called_once_with(
 146             mock.ANY, 'identity', interface='public')
 147         args[1].assert_called_once_with(self.endpoint_test)
 148         msg = "Public endpoint {} is not reachable.".format(self.endpoint_test)
 149         self.assertTrue(msg in str(context.exception))
 150
 151     @mock.patch('functest.ci.check_deployment.verify_connectivity',
 152                 return_value=True)
 153     @mock.patch('functest.ci.check_deployment.keystone_utils.get_endpoint')
 154     def test_check_service_endpoint(self, *args):
 155         self.deployment.check_service_endpoint(self.service_test)
 156         args[0].assert_called_once_with(
 157             mock.ANY, self.service_test, interface='public')
 158         args[1].assert_called_once_with(args[0].return_value)
 159
 160     @mock.patch('functest.ci.check_deployment.verify_connectivity',
 161                 return_value=False)
 162     @mock.patch('functest.ci.check_deployment.keystone_utils.get_endpoint')
 163     def test_check_service_endpoint_ko(self, *args):
 164         args[0].return_value = self.endpoint_test
 165         with self.assertRaises(Exception) as context:
 166             self.deployment.check_service_endpoint(self.service_test)
 167         msg = "{} endpoint {} is not reachable.".format(
 168             self.service_test, self.endpoint_test)
 169         self.assertTrue(msg in str(context.exception))
 170         args[0].assert_called_once_with(
 171             mock.ANY, self.service_test, interface='public')
 172         args[1].assert_called_once_with(args[0].return_value)
 173
 174     @mock.patch('functest.ci.check_deployment.nova_utils.nova_client')
 175     def test_check_nova(self, mock_method):
 176         self.deployment.check_nova()
 177         mock_method.assert_called_once_with(mock.ANY)
 178
 179     @mock.patch('functest.ci.check_deployment.nova_utils.nova_client',
 180                 return_value=mock.Mock(
 181                     servers=mock.Mock(list=mock.Mock(side_effect=Exception))))
 182     def test_check_nova_fail(self, mock_method):
 183         with self.assertRaises(Exception):
 184             self.deployment.check_nova()
 185         mock_method.assert_called_once_with(mock.ANY)
 186
 187     @mock.patch('functest.ci.check_deployment.neutron_utils.neutron_client')
 188     def test_check_neutron(self, mock_method):
 189         self.deployment.check_neutron()
 190         mock_method.assert_called_once_with(mock.ANY)
 191
 192     @mock.patch('functest.ci.check_deployment.neutron_utils.neutron_client',
 193                 return_value=mock.Mock(
 194                     list_networks=mock.Mock(side_effect=Exception)))
 195     def test_check_neutron_fail(self, mock_method):
 196         with self.assertRaises(Exception):
 197             self.deployment.check_neutron()
 198         mock_method.assert_called_once_with(mock.ANY)
 199
 200     @mock.patch('functest.ci.check_deployment.glance_utils.glance_client')
 201     def test_check_glance(self, mock_method):
 202         self.deployment.check_glance()
 203         mock_method.assert_called_once_with(mock.ANY)
 204
 205     @mock.patch('functest.ci.check_deployment.glance_utils.glance_client',
 206                 return_value=mock.Mock(
 207                     images=mock.Mock(list=mock.Mock(side_effect=Exception))))
 208     def test_check_glance_fail(self, mock_method):
 209         with self.assertRaises(Exception):
 210             self.deployment.check_glance()
 211         mock_method.assert_called_once_with(mock.ANY)
 212
 213     @mock.patch('functest.ci.check_deployment.LOGGER.info')
 214     @mock.patch('functest.opnfv_tests.openstack.snaps.snaps_utils.'
 215                 'get_ext_net_name', return_value='ext-net')
 216     def test_check_extnet(self, *args):
 217         self.deployment.check_ext_net()
 218         args[0].assert_called_once_with(mock.ANY)
 219         args[1].assert_called_once_with(
 220             "External network found: %s", "ext-net")
 221
 222     @mock.patch('functest.opnfv_tests.openstack.snaps.snaps_utils.'
 223                 'get_ext_net_name', return_value='')
 224     def test_check_extnet_none(self, mock_getext):
 225         with self.assertRaises(Exception) as context:
 226             self.deployment.check_ext_net()
 227         self.assertTrue(mock_getext.called)
 228         msg = 'ERROR: No external networks in the deployment.'
 229         self.assertTrue(msg in str(context.exception))
 230
 231     @mock.patch('functest.ci.check_deployment.CheckDeployment.check_rc',
 232                 side_effect=Exception)
 233     def test_check_all_exc1(self, *args):
 234         with self.assertRaises(Exception):
 235             self.deployment.check_all()
 236         args[0].assert_called_once_with()
 237
 238     @mock.patch('snaps.openstack.tests.openstack_tests.get_credentials',
 239                 side_effect=Exception)
 240     @mock.patch('functest.ci.check_deployment.CheckDeployment.check_rc')
 241     def test_check_all_exc2(self, *args):
 242         with self.assertRaises(Exception):
 243             self.deployment.check_all()
 244         args[0].assert_called_once_with()
 245         args[1].assert_called_once_with(
 246             os_env_file=self.rc_file, proxy_settings_str=None,
 247             ssh_proxy_cmd=None)
 248
 249     @mock.patch('snaps.openstack.tests.openstack_tests.get_credentials',
 250                 return_value=None)
 251     @mock.patch('functest.ci.check_deployment.CheckDeployment.check_rc')
 252     def test_check_all_exc3(self, *args):
 253         with self.assertRaises(Exception):
 254             self.deployment.check_all()
 255         args[0].assert_called_once_with()
 256         args[1].assert_called_once_with(
 257             os_env_file=self.rc_file, proxy_settings_str=None,
 258             ssh_proxy_cmd=None)
 259
 260     @mock.patch('functest.ci.check_deployment.CheckDeployment.check_ext_net')
 261     @mock.patch('functest.ci.check_deployment.CheckDeployment.check_glance')
 262     @mock.patch('functest.ci.check_deployment.CheckDeployment.check_neutron')
 263     @mock.patch('functest.ci.check_deployment.CheckDeployment.check_nova')
 264     @mock.patch(
 265         'functest.ci.check_deployment.CheckDeployment.check_service_endpoint')
 266     @mock.patch(
 267         'functest.ci.check_deployment.CheckDeployment.check_public_endpoint')
 268     @mock.patch(
 269         'functest.ci.check_deployment.CheckDeployment.check_auth_endpoint')
 270     @mock.patch('snaps.openstack.tests.openstack_tests.get_credentials')
 271     @mock.patch('functest.ci.check_deployment.CheckDeployment.check_rc')
 272     def test_check_all(self, *args):
 273         self.assertEqual(self.deployment.check_all(), 0)
 274         for i in [0, 2, 3, 5, 6, 7, 8]:
 275             args[i].assert_called_once_with()
 276         args[1].assert_called_once_with(
 277             os_env_file=self.rc_file, proxy_settings_str=None,
 278             ssh_proxy_cmd=None)
 279         calls = [mock.call('compute'), mock.call('network'),
 280                  mock.call('image')]
 281         args[4].assert_has_calls(calls)
 282
 283
 284 if __name__ == "__main__":
 285     logging.disable(logging.CRITICAL)
 286     unittest.main(verbosity=2)