functest/tests/unit/utils/test_openstack_tacker.py

   1 #!/usr/bin/env python
   2
   3 # All rights reserved. This program and the accompanying materials
   4 # are made available under the terms of the Apache License, Version 2.0
   5 # which accompanies this distribution, and is available at
   6 # http://www.apache.org/licenses/LICENSE-2.0
   7
   8 import logging
   9 import mock
  10 import unittest
  11
  12 from functest.utils import openstack_tacker
  13 from functest.tests.unit import test_utils
  14
  15
  16 class OSTackerTesting(unittest.TestCase):
  17
  18     logging.disable(logging.CRITICAL)
  19
  20     def setUp(self):
  21         self.tacker_client = mock.Mock()
  22         self.getresponse = {'vnfds': [{'id': 'test_id'}],
  23                             'vnfs': [{'id': 'test_id'}],
  24                             'sfcs': [{'id': 'test_id'}]}
  25         self.vnfdlist = {'vnfds': [{'id': 'test_vnfd1'}, {'id': 'test_vnfd2'}]}
  26         self.vnflist = {'vnfs': [{'id': 'test_vnf1'}, {'id': 'test_vnf2'}]}
  27         self.sfclist = {'sfcs': [{'id': 'test_sfc1'}, {'id': 'test_sfc2'}]}
  28         self.sfc_classifierlist = {'sfc_classifiers': [{'id': 'test_sfc_cl1'},
  29                                    {'id': 'test_sfc_cl2'}]}
  30
  31         self.createvnfd = {"vnfd": {"attributes": {"vnfd": 'vnfd_body'}}}
  32         self.createvnf = {"vnf": {"attributes": {"vnf": 'vnf_body'}}}
  33         self.createsfc = {"sfc": {"attributes": {"sfc": 'sfc_body'}}}
  34         self.createsfc_clf = {"sfc_classifier": {"attributes":
  35                                                  {"sfc_clf": 'sfc_clf_body'}}}
  36
  37         self.resource_type = 'vnfd'
  38         self.resource_name = 'resource_name'
  39         self.tosca_file = 'test_tosca_file'
  40         self.vnfd = 'test_vnfd'
  41         self.vnf = 'test_vnf'
  42         self.sfc = 'test_sfc'
  43         self.sfc_clf = 'test_sfc_clf'
  44
  45     def _get_creds(self):
  46         cred_dict = {
  47             'OS_USERNAME': 'username',
  48             'OS_PASSWORD': 'password',
  49             'OS_AUTH_URL': 'auth_url',
  50             'OS_TENANT_NAME': 'tenant_name',
  51             'OS_USER_DOMAIN_NAME': 'user_domain_name',
  52             'OS_PROJECT_DOMAIN_NAME': 'project_domain_name',
  53             'OS_PROJECT_NAME': 'project_name',
  54             'OS_ENDPOINT_TYPE': 'endpoint_type',
  55             'OS_REGION_NAME': 'region_name'
  56         }
  57         return cred_dict
  58
  59     def test_get_id_from_name(self):
  60         with mock.patch.object(self.tacker_client, 'get',
  61                                return_value=self.getresponse):
  62             resp = openstack_tacker.get_id_from_name(self.tacker_client,
  63                                                      self.resource_type,
  64                                                      self.resource_name)
  65             self.assertEqual(resp, 'test_id')
  66
  67     @mock.patch('functest.utils.openstack_tacker.logger.error')
  68     def test_get_id_from_name_exception(self, mock_logger_error):
  69         with mock.patch.object(self.tacker_client, 'get',
  70                                side_effect=Exception):
  71             resp = openstack_tacker.get_id_from_name(self.tacker_client,
  72                                                      self.resource_type,
  73                                                      self.resource_name)
  74             self.assertIsNone(resp)
  75             mock_logger_error.assert_called_once_with(test_utils.
  76                                                       SubstrMatch("Error [get"
  77                                                                   "_id_from_"
  78                                                                   "name(tacker"
  79                                                                   "_client"
  80                                                                   ", resource_"
  81                                                                   "type, "
  82                                                                   "resource_"
  83                                                                   "name)]:"))
  84
  85     @mock.patch('functest.utils.openstack_tacker.get_id_from_name')
  86     def test_get_vnfd_id(self, mock_get_id_from_name):
  87         openstack_tacker.get_vnfd_id(self.tacker_client, self.resource_name)
  88         mock_get_id_from_name.assert_called_once_with(self.tacker_client,
  89                                                       'vnfd',
  90                                                       self.resource_name)
  91
  92     @mock.patch('functest.utils.openstack_tacker.get_id_from_name')
  93     def test_get_vnf_id(self, mock_get_id_from_name):
  94         openstack_tacker.get_vnf_id(self.tacker_client, self.resource_name)
  95         mock_get_id_from_name.assert_called_once_with(self.tacker_client,
  96                                                       'vnf',
  97                                                       self.resource_name)
  98
  99     @mock.patch('functest.utils.openstack_tacker.get_id_from_name')
 100     def test_get_sfc_id(self, mock_get_id_from_name):
 101         openstack_tacker.get_sfc_id(self.tacker_client, self.resource_name)
 102         mock_get_id_from_name.assert_called_once_with(self.tacker_client,
 103                                                       'sfc',
 104                                                       self.resource_name)
 105
 106     @mock.patch('functest.utils.openstack_tacker.get_id_from_name')
 107     def test_get_sfc_classifier_id(self, mock_get_id_from_name):
 108         openstack_tacker.get_sfc_classifier_id(self.tacker_client,
 109                                                self.resource_name)
 110         mock_get_id_from_name.assert_called_once_with(self.tacker_client,
 111                                                       'sfc-classifier',
 112                                                       self.resource_name)
 113
 114     def test_list_vnfds(self):
 115         with mock.patch.object(self.tacker_client, 'list_vnfds',
 116                                return_value=self.vnfdlist):
 117             resp = openstack_tacker.list_vnfds(self.tacker_client,
 118                                                verbose=False)
 119             self.assertEqual(resp, ['test_vnfd1', 'test_vnfd2'])
 120
 121     def test_list_vnfds_verbose(self):
 122         with mock.patch.object(self.tacker_client, 'list_vnfds',
 123                                return_value=self.vnfdlist):
 124             resp = openstack_tacker.list_vnfds(self.tacker_client,
 125                                                verbose=True)
 126             self.assertEqual(resp, self.vnfdlist)
 127
 128     @mock.patch('functest.utils.openstack_tacker.logger.error')
 129     def test_list_vnfds_exception(self, mock_logger_error):
 130         with mock.patch.object(self.tacker_client, 'list_vnfds',
 131                                side_effect=Exception):
 132             resp = openstack_tacker.list_vnfds(self.tacker_client,
 133                                                verbose=False)
 134             mock_logger_error.assert_called_once_with(test_utils.
 135                                                       SubstrMatch("Error"
 136                                                                   " [list"
 137                                                                   "_vnfds("
 138                                                                   "tacker_"
 139                                                                   "client)]:"))
 140             self.assertIsNone(resp)
 141
 142     def test_create_vnfd_missing_file(self):
 143         with mock.patch.object(self.tacker_client, 'create_vnfd',
 144                                return_value=self.createvnfd):
 145             resp = openstack_tacker.create_vnfd(self.tacker_client,
 146                                                 tosca_file=None)
 147             self.assertEqual(resp, self.createvnfd)
 148
 149     @mock.patch('functest.utils.openstack_tacker.logger.error')
 150     def test_create_vnfd_default(self, mock_logger_error):
 151         with mock.patch.object(self.tacker_client, 'create_vnfd',
 152                                return_value=self.createvnfd), \
 153                 mock.patch('__builtin__.open', mock.mock_open(read_data='1')) \
 154                 as m:
 155             resp = openstack_tacker.create_vnfd(self.tacker_client,
 156                                                 tosca_file=self.tosca_file)
 157             m.assert_called_once_with(self.tosca_file)
 158             mock_logger_error.assert_called_once_with('1')
 159             self.assertEqual(resp, self.createvnfd)
 160
 161     @mock.patch('functest.utils.openstack_tacker.logger.exception')
 162     def test_create_vnfd_exception(self, mock_logger_excep):
 163         with mock.patch.object(self.tacker_client, 'create_vnfd',
 164                                side_effect=Exception):
 165             resp = openstack_tacker.create_vnfd(self.tacker_client,
 166                                                 tosca_file=self.tosca_file)
 167             mock_logger_excep.assert_called_once_with(test_utils.
 168                                                       SubstrMatch("Error"
 169                                                                   " [create"
 170                                                                   "_vnfd("
 171                                                                   "tacker_"
 172                                                                   "client, "
 173                                                                   "'%s')]"
 174                                                                   % self.
 175                                                                   tosca_file))
 176             self.assertIsNone(resp)
 177
 178     def test_delete_vnfd(self):
 179         with mock.patch('functest.utils.openstack_tacker.get_vnfd_id',
 180                         return_value=self.vnfd), \
 181                 mock.patch.object(self.tacker_client, 'delete_vnfd',
 182                                   return_value=self.vnfd):
 183             resp = openstack_tacker.delete_vnfd(self.tacker_client,
 184                                                 vnfd_id='vnfd_id',
 185                                                 vnfd_name=self.vnfd)
 186             self.assertEqual(resp, self.vnfd)
 187
 188     # TODO: Exception('You need to provide an VNFD'
 189     #                 'id or name') AssertionError
 190
 191     @mock.patch('functest.utils.openstack_tacker.logger.error')
 192     def test_delete_vnfd_exception(self, mock_logger_error):
 193         with mock.patch('functest.utils.openstack_tacker.get_vnfd_id',
 194                         return_value=self.vnfd), \
 195                 mock.patch.object(self.tacker_client, 'delete_vnfd',
 196                                   side_effect=Exception):
 197             resp = openstack_tacker.delete_vnfd(self.tacker_client,
 198                                                 vnfd_id=None,
 199                                                 vnfd_name=None)
 200             self.assertIsNone(resp)
 201             self.assertTrue(mock_logger_error.called)
 202
 203     def test_list_vnfs(self):
 204         with mock.patch.object(self.tacker_client, 'list_vnfs',
 205                                return_value=self.vnflist):
 206             resp = openstack_tacker.list_vnfs(self.tacker_client,
 207                                               verbose=False)
 208             self.assertEqual(resp, ['test_vnf1', 'test_vnf2'])
 209
 210     def test_list_vnfs_verbose(self):
 211         with mock.patch.object(self.tacker_client, 'list_vnfs',
 212                                return_value=self.vnflist):
 213             resp = openstack_tacker.list_vnfs(self.tacker_client,
 214                                               verbose=True)
 215             self.assertEqual(resp, self.vnflist)
 216
 217     @mock.patch('functest.utils.openstack_tacker.logger.error')
 218     def test_list_vnfs_exception(self, mock_logger_error):
 219         with mock.patch.object(self.tacker_client, 'list_vnfs',
 220                                side_effect=Exception):
 221             resp = openstack_tacker.list_vnfs(self.tacker_client,
 222                                               verbose=False)
 223             mock_logger_error.assert_called_once_with(test_utils.
 224                                                       SubstrMatch("Error"
 225                                                                   " [list"
 226                                                                   "_vnfs("
 227                                                                   "tacker_"
 228                                                                   "client)]:"))
 229             self.assertIsNone(resp)
 230
 231     def test_create_vnf_default(self):
 232         with mock.patch.object(self.tacker_client, 'create_vnf',
 233                                return_value=self.createvnf), \
 234                 mock.patch('functest.utils.openstack_tacker.get_vnfd_id',
 235                            return_value=self.vnf):
 236             resp = openstack_tacker.create_vnf(self.tacker_client,
 237                                                vnf_name=self.vnf,
 238                                                vnfd_id='vnfd_id',
 239                                                vnfd_name=self.vnfd)
 240             self.assertEqual(resp, self.createvnf)
 241
 242     @mock.patch('functest.utils.openstack_tacker.logger.error')
 243     def test_create_vnf_exception(self, mock_logger_error):
 244         with mock.patch.object(self.tacker_client, 'create_vnf',
 245                                side_effect=Exception):
 246             resp = openstack_tacker.create_vnf(self.tacker_client,
 247                                                vnf_name=self.vnf,
 248                                                vnfd_id='vnfd_id',
 249                                                vnfd_name=self.vnfd)
 250             mock_logger_error.assert_called_once_with(test_utils.
 251                                                       SubstrMatch("error"
 252                                                                   " [create"
 253                                                                   "_vnf("
 254                                                                   "tacker_"
 255                                                                   "client"))
 256             self.assertIsNone(resp)
 257
 258     # TODO: wait_for_vnf
 259
 260     def test_delete_vnf(self):
 261         with mock.patch('functest.utils.openstack_tacker.get_vnf_id',
 262                         return_value=self.vnf), \
 263                 mock.patch.object(self.tacker_client, 'delete_vnf',
 264                                   return_value=self.vnf):
 265             resp = openstack_tacker.delete_vnf(self.tacker_client,
 266                                                vnf_id='vnf_id',
 267                                                vnf_name=self.vnf)
 268             self.assertEqual(resp, self.vnf)
 269
 270     # TODO: Exception('You need to provide an VNF'
 271     #                 'classifier id or name') AssertionError
 272
 273     @mock.patch('functest.utils.openstack_tacker.logger.error')
 274     def test_delete_vnf_exception(self, mock_logger_error):
 275         with mock.patch('functest.utils.openstack_tacker.get_vnf_id',
 276                         return_value=self.vnf), \
 277                 mock.patch.object(self.tacker_client, 'delete_vnf',
 278                                   side_effect=Exception):
 279             resp = openstack_tacker.delete_vnf(self.tacker_client,
 280                                                vnf_id=None,
 281                                                vnf_name=None)
 282             self.assertIsNone(resp)
 283             self.assertTrue(mock_logger_error.called)
 284
 285     def test_list_sfcs(self):
 286         with mock.patch.object(self.tacker_client, 'list_sfcs',
 287                                return_value=self.sfclist):
 288             resp = openstack_tacker.list_sfcs(self.tacker_client,
 289                                               verbose=False)
 290             self.assertEqual(resp, ['test_sfc1', 'test_sfc2'])
 291
 292     def test_list_sfcs_verbose(self):
 293         with mock.patch.object(self.tacker_client, 'list_sfcs',
 294                                return_value=self.sfclist):
 295             resp = openstack_tacker.list_sfcs(self.tacker_client,
 296                                               verbose=True)
 297             self.assertEqual(resp, self.sfclist)
 298
 299     @mock.patch('functest.utils.openstack_tacker.logger.error')
 300     def test_list_sfcs_exception(self, mock_logger_error):
 301         with mock.patch.object(self.tacker_client, 'list_sfcs',
 302                                side_effect=Exception):
 303             resp = openstack_tacker.list_sfcs(self.tacker_client,
 304                                               verbose=False)
 305             mock_logger_error.assert_called_once_with(test_utils.
 306                                                       SubstrMatch("Error"
 307                                                                   " [list"
 308                                                                   "_sfcs("
 309                                                                   "tacker_"
 310                                                                   "client)]:"))
 311             self.assertIsNone(resp)
 312
 313     def test_create_sfc_default(self):
 314         with mock.patch.object(self.tacker_client, 'create_sfc',
 315                                return_value=self.createsfc), \
 316                 mock.patch('functest.utils.openstack_tacker.get_vnf_id',
 317                            return_value=self.vnf):
 318             resp = openstack_tacker.create_sfc(self.tacker_client,
 319                                                sfc_name=self.sfc,
 320                                                chain_vnf_ids=['chain_vnf_id'],
 321                                                chain_vnf_names=[self.vnf])
 322             self.assertEqual(resp, self.createsfc)
 323
 324     @mock.patch('functest.utils.openstack_tacker.logger.error')
 325     def test_create_sfc_exception(self, mock_logger_error):
 326         with mock.patch.object(self.tacker_client, 'create_sfc',
 327                                side_effect=Exception):
 328             resp = openstack_tacker.create_sfc(self.tacker_client,
 329                                                sfc_name=self.sfc,
 330                                                chain_vnf_ids=['chain_vnf_id'],
 331                                                chain_vnf_names=[self.vnf])
 332             mock_logger_error.assert_called_once_with(test_utils.
 333                                                       SubstrMatch("error"
 334                                                                   " [create"
 335                                                                   "_sfc("
 336                                                                   "tacker_"
 337                                                                   "client"))
 338             self.assertIsNone(resp)
 339
 340     def test_delete_sfc(self):
 341         with mock.patch('functest.utils.openstack_tacker.get_sfc_id',
 342                         return_value=self.sfc), \
 343                 mock.patch.object(self.tacker_client, 'delete_sfc',
 344                                   return_value=self.sfc):
 345             resp = openstack_tacker.delete_sfc(self.tacker_client,
 346                                                sfc_id='sfc_id',
 347                                                sfc_name=self.sfc)
 348             self.assertEqual(resp, self.sfc)
 349
 350     # TODO: Exception('You need to provide an SFC'
 351     #                 'id or name') AssertionError
 352
 353     @mock.patch('functest.utils.openstack_tacker.logger.error')
 354     def test_delete_sfc_exception(self, mock_logger_error):
 355         with mock.patch('functest.utils.openstack_tacker.get_sfc_id',
 356                         return_value=self.sfc), \
 357                 mock.patch.object(self.tacker_client, 'delete_sfc',
 358                                   side_effect=Exception):
 359             resp = openstack_tacker.delete_sfc(self.tacker_client,
 360                                                sfc_id=None,
 361                                                sfc_name=None)
 362             self.assertIsNone(resp)
 363             self.assertTrue(mock_logger_error.called)
 364
 365     def test_list_sfc_classifiers(self):
 366         with mock.patch.object(self.tacker_client, 'list_sfc_classifiers',
 367                                return_value=self.sfc_classifierlist):
 368             resp = openstack_tacker.list_sfc_classifiers(self.tacker_client,
 369                                                          verbose=False)
 370             self.assertEqual(resp, ['test_sfc_cl1', 'test_sfc_cl2'])
 371
 372     def test_list_sfc_classifiers_verbose(self):
 373         with mock.patch.object(self.tacker_client, 'list_sfc_classifiers',
 374                                return_value=self.sfc_classifierlist):
 375             resp = openstack_tacker.list_sfc_classifiers(self.tacker_client,
 376                                                          verbose=True)
 377             self.assertEqual(resp, self.sfc_classifierlist)
 378
 379     @mock.patch('functest.utils.openstack_tacker.logger.error')
 380     def test_list_sfc_classifiers_exception(self, mock_logger_error):
 381         with mock.patch.object(self.tacker_client, 'list_sfc_classifiers',
 382                                side_effect=Exception):
 383             resp = openstack_tacker.list_sfc_classifiers(self.tacker_client,
 384                                                          verbose=False)
 385             mock_logger_error.assert_called_once_with(test_utils.
 386                                                       SubstrMatch("Error"
 387                                                                   " [list"
 388                                                                   "_sfc_cl"
 389                                                                   "assifiers("
 390                                                                   "tacker_"
 391                                                                   "client)]:"))
 392             self.assertIsNone(resp)
 393
 394     def test_create_sfc_classifier_default(self):
 395         with mock.patch.object(self.tacker_client, 'create_sfc_classifier',
 396                                return_value=self.createsfc_clf), \
 397                 mock.patch('functest.utils.openstack_tacker.get_sfc_id',
 398                            return_value=self.sfc):
 399             cl = self.sfc_clf
 400             resp = openstack_tacker.create_sfc_classifier(self.tacker_client,
 401                                                           sfc_clf_name=cl,
 402                                                           sfc_id='sfc_id',
 403                                                           sfc_name=self.sfc)
 404             self.assertEqual(resp, self.createsfc_clf)
 405
 406     @mock.patch('functest.utils.openstack_tacker.logger.error')
 407     def test_sfc_classifier_exception(self, mock_logger_error):
 408         with mock.patch.object(self.tacker_client, 'create_sfc_classifier',
 409                                side_effect=Exception):
 410             cl = self.sfc_clf
 411             resp = openstack_tacker.create_sfc_classifier(self.tacker_client,
 412                                                           sfc_clf_name=cl,
 413                                                           sfc_id='sfc_id',
 414                                                           sfc_name=self.sfc)
 415             mock_logger_error.assert_called_once_with(test_utils.
 416                                                       SubstrMatch("error"
 417                                                                   " [create"
 418                                                                   "_sfc_cl"
 419                                                                   "assifier("
 420                                                                   "tacker_"
 421                                                                   "client"))
 422             self.assertIsNone(resp)
 423
 424     def test_delete_sfc_classifier(self):
 425         with mock.patch('functest.utils.openstack_tacker.get_sfc_'
 426                         'classifier_id',
 427                         return_value=self.sfc_clf), \
 428                 mock.patch.object(self.tacker_client, 'delete_sfc_classifier',
 429                                   return_value=self.sfc_clf):
 430             cl = self.sfc_clf
 431             resp = openstack_tacker.delete_sfc_classifier(self.tacker_client,
 432                                                           sfc_clf_id='sfc_id',
 433                                                           sfc_clf_name=cl)
 434             self.assertEqual(resp, cl)
 435
 436     # TODO: Exception('You need to provide an SFC'
 437     #                 'classifier id or name') AssertionError
 438
 439     @mock.patch('functest.utils.openstack_tacker.logger.error')
 440     def test_delete_sfc_classifier_exception(self, mock_logger_error):
 441         with mock.patch('functest.utils.openstack_tacker.get_sfc_'
 442                         'classifier_id',
 443                         return_value=self.sfc_clf), \
 444                 mock.patch.object(self.tacker_client, 'delete_sfc_classifier',
 445                                   side_effect=Exception):
 446             cl = self.sfc_clf
 447             resp = openstack_tacker.delete_sfc_classifier(self.tacker_client,
 448                                                           sfc_clf_id='sfc_id',
 449                                                           sfc_clf_name=cl)
 450             self.assertIsNone(resp)
 451             self.assertTrue(mock_logger_error.called)
 452
 453
 454 if __name__ == "__main__":
 455     unittest.main(verbosity=2)