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     def test_create_vnfd_default(self):
 150         with mock.patch.object(self.tacker_client, 'create_vnfd',
 151                                return_value=self.createvnfd), \
 152                 mock.patch('__builtin__.open', mock.mock_open(read_data='1')) \
 153                 as m:
 154             resp = openstack_tacker.create_vnfd(self.tacker_client,
 155                                                 tosca_file=self.tosca_file)
 156             m.assert_called_once_with(self.tosca_file)
 157             self.assertEqual(resp, self.createvnfd)
 158
 159     @mock.patch('functest.utils.openstack_tacker.logger.error')
 160     def test_create_vnfd_exception(self, mock_logger_error):
 161         with mock.patch.object(self.tacker_client, 'create_vnfd',
 162                                side_effect=Exception):
 163             resp = openstack_tacker.create_vnfd(self.tacker_client,
 164                                                 tosca_file=self.tosca_file)
 165             mock_logger_error.assert_called_once_with(test_utils.
 166                                                       SubstrMatch("Error"
 167                                                                   " [create"
 168                                                                   "_vnfd("
 169                                                                   "tacker_"
 170                                                                   "client, "
 171                                                                   "'%s')]"
 172                                                                   % self.
 173                                                                   tosca_file))
 174             self.assertIsNone(resp)
 175
 176     def test_delete_vnfd(self):
 177         with mock.patch('functest.utils.openstack_tacker.get_vnfd_id',
 178                         return_value=self.vnfd), \
 179                 mock.patch.object(self.tacker_client, 'delete_vnfd',
 180                                   return_value=self.vnfd):
 181             resp = openstack_tacker.delete_vnfd(self.tacker_client,
 182                                                 vnfd_id='vnfd_id',
 183                                                 vnfd_name=self.vnfd)
 184             self.assertEqual(resp, self.vnfd)
 185
 186     # TODO: Exception('You need to provide an VNFD'
 187     #                 'id or name') AssertionError
 188
 189     @mock.patch('functest.utils.openstack_tacker.logger.error')
 190     def test_delete_vnfd_exception(self, mock_logger_error):
 191         with mock.patch('functest.utils.openstack_tacker.get_vnfd_id',
 192                         return_value=self.vnfd), \
 193                 mock.patch.object(self.tacker_client, 'delete_vnfd',
 194                                   side_effect=Exception):
 195             resp = openstack_tacker.delete_vnfd(self.tacker_client,
 196                                                 vnfd_id=None,
 197                                                 vnfd_name=None)
 198             self.assertIsNone(resp)
 199             self.assertTrue(mock_logger_error.called)
 200
 201     def test_list_vnfs(self):
 202         with mock.patch.object(self.tacker_client, 'list_vnfs',
 203                                return_value=self.vnflist):
 204             resp = openstack_tacker.list_vnfs(self.tacker_client,
 205                                               verbose=False)
 206             self.assertEqual(resp, ['test_vnf1', 'test_vnf2'])
 207
 208     def test_list_vnfs_verbose(self):
 209         with mock.patch.object(self.tacker_client, 'list_vnfs',
 210                                return_value=self.vnflist):
 211             resp = openstack_tacker.list_vnfs(self.tacker_client,
 212                                               verbose=True)
 213             self.assertEqual(resp, self.vnflist)
 214
 215     @mock.patch('functest.utils.openstack_tacker.logger.error')
 216     def test_list_vnfs_exception(self, mock_logger_error):
 217         with mock.patch.object(self.tacker_client, 'list_vnfs',
 218                                side_effect=Exception):
 219             resp = openstack_tacker.list_vnfs(self.tacker_client,
 220                                               verbose=False)
 221             mock_logger_error.assert_called_once_with(test_utils.
 222                                                       SubstrMatch("Error"
 223                                                                   " [list"
 224                                                                   "_vnfs("
 225                                                                   "tacker_"
 226                                                                   "client)]:"))
 227             self.assertIsNone(resp)
 228
 229     def test_create_vnf_default(self):
 230         with mock.patch.object(self.tacker_client, 'create_vnf',
 231                                return_value=self.createvnf), \
 232                 mock.patch('functest.utils.openstack_tacker.get_vnfd_id',
 233                            return_value=self.vnf):
 234             resp = openstack_tacker.create_vnf(self.tacker_client,
 235                                                vnf_name=self.vnf,
 236                                                vnfd_id='vnfd_id',
 237                                                vnfd_name=self.vnfd)
 238             self.assertEqual(resp, self.createvnf)
 239
 240     @mock.patch('functest.utils.openstack_tacker.logger.error')
 241     def test_create_vnf_exception(self, mock_logger_error):
 242         with mock.patch.object(self.tacker_client, 'create_vnf',
 243                                side_effect=Exception):
 244             resp = openstack_tacker.create_vnf(self.tacker_client,
 245                                                vnf_name=self.vnf,
 246                                                vnfd_id='vnfd_id',
 247                                                vnfd_name=self.vnfd)
 248             mock_logger_error.assert_called_once_with(test_utils.
 249                                                       SubstrMatch("error"
 250                                                                   " [create"
 251                                                                   "_vnf("
 252                                                                   "tacker_"
 253                                                                   "client"))
 254             self.assertIsNone(resp)
 255
 256     # TODO: wait_for_vnf
 257
 258     def test_delete_vnf(self):
 259         with mock.patch('functest.utils.openstack_tacker.get_vnf_id',
 260                         return_value=self.vnf), \
 261                 mock.patch.object(self.tacker_client, 'delete_vnf',
 262                                   return_value=self.vnf):
 263             resp = openstack_tacker.delete_vnf(self.tacker_client,
 264                                                vnf_id='vnf_id',
 265                                                vnf_name=self.vnf)
 266             self.assertEqual(resp, self.vnf)
 267
 268     # TODO: Exception('You need to provide an VNF'
 269     #                 'classifier id or name') AssertionError
 270
 271     @mock.patch('functest.utils.openstack_tacker.logger.error')
 272     def test_delete_vnf_exception(self, mock_logger_error):
 273         with mock.patch('functest.utils.openstack_tacker.get_vnf_id',
 274                         return_value=self.vnf), \
 275                 mock.patch.object(self.tacker_client, 'delete_vnf',
 276                                   side_effect=Exception):
 277             resp = openstack_tacker.delete_vnf(self.tacker_client,
 278                                                vnf_id=None,
 279                                                vnf_name=None)
 280             self.assertIsNone(resp)
 281             self.assertTrue(mock_logger_error.called)
 282
 283     def test_list_sfcs(self):
 284         with mock.patch.object(self.tacker_client, 'list_sfcs',
 285                                return_value=self.sfclist):
 286             resp = openstack_tacker.list_sfcs(self.tacker_client,
 287                                               verbose=False)
 288             self.assertEqual(resp, ['test_sfc1', 'test_sfc2'])
 289
 290     def test_list_sfcs_verbose(self):
 291         with mock.patch.object(self.tacker_client, 'list_sfcs',
 292                                return_value=self.sfclist):
 293             resp = openstack_tacker.list_sfcs(self.tacker_client,
 294                                               verbose=True)
 295             self.assertEqual(resp, self.sfclist)
 296
 297     @mock.patch('functest.utils.openstack_tacker.logger.error')
 298     def test_list_sfcs_exception(self, mock_logger_error):
 299         with mock.patch.object(self.tacker_client, 'list_sfcs',
 300                                side_effect=Exception):
 301             resp = openstack_tacker.list_sfcs(self.tacker_client,
 302                                               verbose=False)
 303             mock_logger_error.assert_called_once_with(test_utils.
 304                                                       SubstrMatch("Error"
 305                                                                   " [list"
 306                                                                   "_sfcs("
 307                                                                   "tacker_"
 308                                                                   "client)]:"))
 309             self.assertIsNone(resp)
 310
 311     def test_create_sfc_default(self):
 312         with mock.patch.object(self.tacker_client, 'create_sfc',
 313                                return_value=self.createsfc), \
 314                 mock.patch('functest.utils.openstack_tacker.get_vnf_id',
 315                            return_value=self.vnf):
 316             resp = openstack_tacker.create_sfc(self.tacker_client,
 317                                                sfc_name=self.sfc,
 318                                                chain_vnf_ids=['chain_vnf_id'],
 319                                                chain_vnf_names=[self.vnf])
 320             self.assertEqual(resp, self.createsfc)
 321
 322     @mock.patch('functest.utils.openstack_tacker.logger.error')
 323     def test_create_sfc_exception(self, mock_logger_error):
 324         with mock.patch.object(self.tacker_client, 'create_sfc',
 325                                side_effect=Exception):
 326             resp = openstack_tacker.create_sfc(self.tacker_client,
 327                                                sfc_name=self.sfc,
 328                                                chain_vnf_ids=['chain_vnf_id'],
 329                                                chain_vnf_names=[self.vnf])
 330             mock_logger_error.assert_called_once_with(test_utils.
 331                                                       SubstrMatch("error"
 332                                                                   " [create"
 333                                                                   "_sfc("
 334                                                                   "tacker_"
 335                                                                   "client"))
 336             self.assertIsNone(resp)
 337
 338     def test_delete_sfc(self):
 339         with mock.patch('functest.utils.openstack_tacker.get_sfc_id',
 340                         return_value=self.sfc), \
 341                 mock.patch.object(self.tacker_client, 'delete_sfc',
 342                                   return_value=self.sfc):
 343             resp = openstack_tacker.delete_sfc(self.tacker_client,
 344                                                sfc_id='sfc_id',
 345                                                sfc_name=self.sfc)
 346             self.assertEqual(resp, self.sfc)
 347
 348     # TODO: Exception('You need to provide an SFC'
 349     #                 'id or name') AssertionError
 350
 351     @mock.patch('functest.utils.openstack_tacker.logger.error')
 352     def test_delete_sfc_exception(self, mock_logger_error):
 353         with mock.patch('functest.utils.openstack_tacker.get_sfc_id',
 354                         return_value=self.sfc), \
 355                 mock.patch.object(self.tacker_client, 'delete_sfc',
 356                                   side_effect=Exception):
 357             resp = openstack_tacker.delete_sfc(self.tacker_client,
 358                                                sfc_id=None,
 359                                                sfc_name=None)
 360             self.assertIsNone(resp)
 361             self.assertTrue(mock_logger_error.called)
 362
 363     def test_list_sfc_classifiers(self):
 364         with mock.patch.object(self.tacker_client, 'list_sfc_classifiers',
 365                                return_value=self.sfc_classifierlist):
 366             resp = openstack_tacker.list_sfc_classifiers(self.tacker_client,
 367                                                          verbose=False)
 368             self.assertEqual(resp, ['test_sfc_cl1', 'test_sfc_cl2'])
 369
 370     def test_list_sfc_classifiers_verbose(self):
 371         with mock.patch.object(self.tacker_client, 'list_sfc_classifiers',
 372                                return_value=self.sfc_classifierlist):
 373             resp = openstack_tacker.list_sfc_classifiers(self.tacker_client,
 374                                                          verbose=True)
 375             self.assertEqual(resp, self.sfc_classifierlist)
 376
 377     @mock.patch('functest.utils.openstack_tacker.logger.error')
 378     def test_list_sfc_classifiers_exception(self, mock_logger_error):
 379         with mock.patch.object(self.tacker_client, 'list_sfc_classifiers',
 380                                side_effect=Exception):
 381             resp = openstack_tacker.list_sfc_classifiers(self.tacker_client,
 382                                                          verbose=False)
 383             mock_logger_error.assert_called_once_with(test_utils.
 384                                                       SubstrMatch("Error"
 385                                                                   " [list"
 386                                                                   "_sfc_cl"
 387                                                                   "assifiers("
 388                                                                   "tacker_"
 389                                                                   "client)]:"))
 390             self.assertIsNone(resp)
 391
 392     def test_create_sfc_classifier_default(self):
 393         with mock.patch.object(self.tacker_client, 'create_sfc_classifier',
 394                                return_value=self.createsfc_clf), \
 395                 mock.patch('functest.utils.openstack_tacker.get_sfc_id',
 396                            return_value=self.sfc):
 397             cl = self.sfc_clf
 398             resp = openstack_tacker.create_sfc_classifier(self.tacker_client,
 399                                                           sfc_clf_name=cl,
 400                                                           sfc_id='sfc_id',
 401                                                           sfc_name=self.sfc)
 402             self.assertEqual(resp, self.createsfc_clf)
 403
 404     @mock.patch('functest.utils.openstack_tacker.logger.error')
 405     def test_sfc_classifier_exception(self, mock_logger_error):
 406         with mock.patch.object(self.tacker_client, 'create_sfc_classifier',
 407                                side_effect=Exception):
 408             cl = self.sfc_clf
 409             resp = openstack_tacker.create_sfc_classifier(self.tacker_client,
 410                                                           sfc_clf_name=cl,
 411                                                           sfc_id='sfc_id',
 412                                                           sfc_name=self.sfc)
 413             mock_logger_error.assert_called_once_with(test_utils.
 414                                                       SubstrMatch("error"
 415                                                                   " [create"
 416                                                                   "_sfc_cl"
 417                                                                   "assifier("
 418                                                                   "tacker_"
 419                                                                   "client"))
 420             self.assertIsNone(resp)
 421
 422     def test_delete_sfc_classifier(self):
 423         with mock.patch('functest.utils.openstack_tacker.get_sfc_'
 424                         'classifier_id',
 425                         return_value=self.sfc_clf), \
 426                 mock.patch.object(self.tacker_client, 'delete_sfc_classifier',
 427                                   return_value=self.sfc_clf):
 428             cl = self.sfc_clf
 429             resp = openstack_tacker.delete_sfc_classifier(self.tacker_client,
 430                                                           sfc_clf_id='sfc_id',
 431                                                           sfc_clf_name=cl)
 432             self.assertEqual(resp, cl)
 433
 434     # TODO: Exception('You need to provide an SFC'
 435     #                 'classifier id or name') AssertionError
 436
 437     @mock.patch('functest.utils.openstack_tacker.logger.error')
 438     def test_delete_sfc_classifier_exception(self, mock_logger_error):
 439         with mock.patch('functest.utils.openstack_tacker.get_sfc_'
 440                         'classifier_id',
 441                         return_value=self.sfc_clf), \
 442                 mock.patch.object(self.tacker_client, 'delete_sfc_classifier',
 443                                   side_effect=Exception):
 444             cl = self.sfc_clf
 445             resp = openstack_tacker.delete_sfc_classifier(self.tacker_client,
 446                                                           sfc_clf_id='sfc_id',
 447                                                           sfc_clf_name=cl)
 448             self.assertIsNone(resp)
 449             self.assertTrue(mock_logger_error.called)
 450
 451
 452 if __name__ == "__main__":
 453     unittest.main(verbosity=2)