Merge "Remove tacker library from functest"

2 files changed, 0 insertions, 841 deletions

diff --git a/functest/tests/unit/utils/test_openstack_tacker.py b/functest/tests/unit/utils/test_openstack_tacker.py
deleted file mode 100644
index 3c0fc3d0..00000000
--- a/functest/tests/unit/utils/test_openstack_tacker.py
+++ /dev/null
@@ -1,524 +0,0 @@
-#!/usr/bin/env python
-
-# All rights reserved. This program and the accompanying materials
-# are made available under the terms of the Apache License, Version 2.0
-# which accompanies this distribution, and is available at
-# http://www.apache.org/licenses/LICENSE-2.0
-
-import logging
-import unittest
-
-import mock
-from tackerclient.v1_0 import client as tackerclient
-
-from functest.utils import openstack_tacker
-from functest.tests.unit import test_utils
-
-
-class OSTackerTesting(unittest.TestCase):
-
-    def setUp(self):
-        self.tacker_client = mock.Mock()
-        self.getresponse = {'vnfds': [{'id': 'test_id'}],
-                            'vnfs': [{'id': 'test_id'}],
-                            'sfcs': [{'id': 'test_id'}]}
-        self.vnfdlist = {'vnfds': [{'id': 'test_vnfd1'}, {'id': 'test_vnfd2'}]}
-        self.vnflist = {'vnfs': [{'id': 'test_vnf1'}, {'id': 'test_vnf2'}]}
-        self.sfclist = {'sfcs': [{'id': 'test_sfc1'}, {'id': 'test_sfc2'}]}
-        self.sfc_classifierlist = {'sfc_classifiers': [{'id': 'test_sfc_cl1'},
-                                   {'id': 'test_sfc_cl2'}]}
-
-        self.createvnfd = {"vnfd": {"attributes": {"vnfd": 'vnfd_body'}}}
-        self.createvnf = {"vnf": {"attributes": {"vnf": 'vnf_body'}}}
-        self.createsfc = {"sfc": {"attributes": {"sfc": 'sfc_body'}}}
-        self.createsfc_clf = {"sfc_classifier": {"attributes":
-                                                 {"sfc_clf": 'sfc_clf_body'}}}
-
-        self.resource_type = 'vnfd'
-        self.resource_name = 'resource_name'
-        self.tosca_file = 'test_tosca_file'
-        self.vnfd = 'test_vnfd'
-        self.vnf = 'test_vnf'
-        self.sfc = 'test_sfc'
-        self.sfc_clf = 'test_sfc_clf'
-
-    def _get_creds(self):
-        cred_dict = {
-            'OS_USERNAME': 'username',
-            'OS_PASSWORD': 'password',
-            'OS_AUTH_URL': 'auth_url',
-            'OS_TENANT_NAME': 'tenant_name',
-            'OS_USER_DOMAIN_NAME': 'user_domain_name',
-            'OS_PROJECT_DOMAIN_NAME': 'project_domain_name',
-            'OS_PROJECT_NAME': 'project_name',
-            'OS_ENDPOINT_TYPE': 'endpoint_type',
-            'OS_REGION_NAME': 'region_name'
-        }
-        return cred_dict
-
-    def test_get_tacker_client(self):
-        with mock.patch('functest.utils.openstack_tacker.'
-                        'os_utils.get_session'):
-            tackerclient.Client = mock.Mock
-            ret = openstack_tacker.get_tacker_client()
-            self.assertTrue(isinstance(ret, mock.Mock))
-
-    def test_get_id_from_name(self):
-        with mock.patch.object(self.tacker_client, 'get',
-                               return_value=self.getresponse):
-            resp = openstack_tacker.get_id_from_name(self.tacker_client,
-                                                     self.resource_type,
-                                                     self.resource_name)
-            self.assertEqual(resp, 'test_id')
-
-    @mock.patch('functest.utils.openstack_tacker.logger.error')
-    def test_get_id_from_name_exception(self, mock_logger_error):
-        with mock.patch.object(self.tacker_client, 'get',
-                               side_effect=Exception):
-            resp = openstack_tacker.get_id_from_name(self.tacker_client,
-                                                     self.resource_type,
-                                                     self.resource_name)
-            self.assertIsNone(resp)
-            mock_logger_error.assert_called_once_with(test_utils.
-                                                      SubstrMatch("Error [get"
-                                                                  "_id_from_"
-                                                                  "name(tacker"
-                                                                  "_client"
-                                                                  ", resource_"
-                                                                  "type, "
-                                                                  "resource_"
-                                                                  "name)]:"))
-
-    @mock.patch('functest.utils.openstack_tacker.get_id_from_name')
-    def test_get_vnfd_id(self, mock_get_id_from_name):
-        openstack_tacker.get_vnfd_id(self.tacker_client, self.resource_name)
-        mock_get_id_from_name.assert_called_once_with(self.tacker_client,
-                                                      'vnfd',
-                                                      self.resource_name)
-
-    @mock.patch('functest.utils.openstack_tacker.get_id_from_name')
-    def test_get_vnf_id(self, mock_get_id_from_name):
-        openstack_tacker.get_vnf_id(self.tacker_client, self.resource_name)
-        mock_get_id_from_name.assert_called_once_with(self.tacker_client,
-                                                      'vnf',
-                                                      self.resource_name)
-
-    @mock.patch('functest.utils.openstack_tacker.get_id_from_name')
-    def test_get_sfc_id(self, mock_get_id_from_name):
-        openstack_tacker.get_sfc_id(self.tacker_client, self.resource_name)
-        mock_get_id_from_name.assert_called_once_with(self.tacker_client,
-                                                      'sfc',
-                                                      self.resource_name)
-
-    @mock.patch('functest.utils.openstack_tacker.get_id_from_name')
-    def test_get_sfc_classifier_id(self, mock_get_id_from_name):
-        openstack_tacker.get_sfc_classifier_id(self.tacker_client,
-                                               self.resource_name)
-        mock_get_id_from_name.assert_called_once_with(self.tacker_client,
-                                                      'sfc-classifier',
-                                                      self.resource_name)
-
-    def test_list_vnfds(self):
-        with mock.patch.object(self.tacker_client, 'list_vnfds',
-                               return_value=self.vnfdlist):
-            resp = openstack_tacker.list_vnfds(self.tacker_client,
-                                               verbose=False)
-            self.assertEqual(resp, ['test_vnfd1', 'test_vnfd2'])
-
-    def test_list_vnfds_verbose(self):
-        with mock.patch.object(self.tacker_client, 'list_vnfds',
-                               return_value=self.vnfdlist):
-            resp = openstack_tacker.list_vnfds(self.tacker_client,
-                                               verbose=True)
-            self.assertEqual(resp, self.vnfdlist)
-
-    @mock.patch('functest.utils.openstack_tacker.logger.error')
-    def test_list_vnfds_exception(self, mock_logger_error):
-        with mock.patch.object(self.tacker_client, 'list_vnfds',
-                               side_effect=Exception):
-            resp = openstack_tacker.list_vnfds(self.tacker_client,
-                                               verbose=False)
-            mock_logger_error.assert_called_once_with(test_utils.
-                                                      SubstrMatch("Error"
-                                                                  " [list"
-                                                                  "_vnfds("
-                                                                  "tacker_"
-                                                                  "client)]:"))
-            self.assertIsNone(resp)
-
-    def test_create_vnfd_missing_file(self):
-        with mock.patch.object(self.tacker_client, 'create_vnfd',
-                               return_value=self.createvnfd):
-            resp = openstack_tacker.create_vnfd(self.tacker_client,
-                                                tosca_file=None)
-            self.assertEqual(resp, self.createvnfd)
-
-    def test_create_vnfd_default(self):
-        with mock.patch.object(self.tacker_client, 'create_vnfd',
-                               return_value=self.createvnfd), \
-                mock.patch('__builtin__.open', mock.mock_open(read_data='1')) \
-                as m:
-            resp = openstack_tacker.create_vnfd(self.tacker_client,
-                                                tosca_file=self.tosca_file)
-            m.assert_called_once_with(self.tosca_file)
-            self.assertEqual(resp, self.createvnfd)
-
-    @mock.patch('functest.utils.openstack_tacker.logger.error')
-    def test_create_vnfd_exception(self, mock_logger_error):
-        with mock.patch.object(self.tacker_client, 'create_vnfd',
-                               side_effect=Exception):
-            resp = openstack_tacker.create_vnfd(self.tacker_client,
-                                                tosca_file=self.tosca_file)
-            mock_logger_error.assert_called_once_with(test_utils.
-                                                      SubstrMatch("Error"
-                                                                  " [create"
-                                                                  "_vnfd("
-                                                                  "tacker_"
-                                                                  "client, "
-                                                                  "'%s')]"
-                                                                  % self.
-                                                                  tosca_file))
-            self.assertIsNone(resp)
-
-    def test_delete_vnfd(self):
-        with mock.patch('functest.utils.openstack_tacker.get_vnfd_id',
-                        return_value=self.vnfd), \
-                mock.patch.object(self.tacker_client, 'delete_vnfd',
-                                  return_value=self.vnfd):
-            resp = openstack_tacker.delete_vnfd(self.tacker_client,
-                                                vnfd_id='vnfd_id',
-                                                vnfd_name=self.vnfd)
-            self.assertEqual(resp, self.vnfd)
-
-    def test_delete_vnfd_missing_vnfd_name(self):
-        with mock.patch('functest.utils.openstack_tacker.get_vnfd_id',
-                        return_value=self.vnfd), \
-                self.assertRaises(Exception) as context:
-            resp = openstack_tacker.delete_vnfd(self.tacker_client,
-                                                vnfd_id=None,
-                                                vnfd_name=None)
-            self.assertIsNone(resp)
-            msg = 'You need to provide VNFD id or VNFD name'
-            self.assertTrue(msg in context)
-
-    @mock.patch('functest.utils.openstack_tacker.logger.error')
-    def test_delete_vnfd_exception(self, mock_logger_error):
-        with mock.patch('functest.utils.openstack_tacker.get_vnfd_id',
-                        return_value=self.vnfd), \
-                mock.patch.object(self.tacker_client, 'delete_vnfd',
-                                  side_effect=Exception):
-            resp = openstack_tacker.delete_vnfd(self.tacker_client,
-                                                vnfd_id=None,
-                                                vnfd_name=None)
-            self.assertIsNone(resp)
-            self.assertTrue(mock_logger_error.called)
-
-    def test_list_vnfs(self):
-        with mock.patch.object(self.tacker_client, 'list_vnfs',
-                               return_value=self.vnflist):
-            resp = openstack_tacker.list_vnfs(self.tacker_client,
-                                              verbose=False)
-            self.assertEqual(resp, ['test_vnf1', 'test_vnf2'])
-
-    def test_list_vnfs_verbose(self):
-        with mock.patch.object(self.tacker_client, 'list_vnfs',
-                               return_value=self.vnflist):
-            resp = openstack_tacker.list_vnfs(self.tacker_client,
-                                              verbose=True)
-            self.assertEqual(resp, self.vnflist)
-
-    @mock.patch('functest.utils.openstack_tacker.logger.error')
-    def test_list_vnfs_exception(self, mock_logger_error):
-        with mock.patch.object(self.tacker_client, 'list_vnfs',
-                               side_effect=Exception):
-            resp = openstack_tacker.list_vnfs(self.tacker_client,
-                                              verbose=False)
-            mock_logger_error.assert_called_once_with(test_utils.
-                                                      SubstrMatch("Error"
-                                                                  " [list"
-                                                                  "_vnfs("
-                                                                  "tacker_"
-                                                                  "client)]:"))
-            self.assertIsNone(resp)
-
-    def test_create_vnf_default(self):
-        with mock.patch.object(self.tacker_client, 'create_vnf',
-                               return_value=self.createvnf), \
-                mock.patch('functest.utils.openstack_tacker.get_vnfd_id',
-                           return_value=self.vnf):
-            resp = openstack_tacker.create_vnf(self.tacker_client,
-                                               vnf_name=self.vnf,
-                                               vnfd_id='vnfd_id',
-                                               vnfd_name=self.vnfd)
-            self.assertEqual(resp, self.createvnf)
-
-    @mock.patch('functest.utils.openstack_tacker.logger.error')
-    def test_create_vnf_exception(self, mock_logger_error):
-        with mock.patch.object(self.tacker_client, 'create_vnf',
-                               side_effect=Exception):
-            resp = openstack_tacker.create_vnf(self.tacker_client,
-                                               vnf_name=self.vnf,
-                                               vnfd_id='vnfd_id',
-                                               vnfd_name=self.vnfd)
-            mock_logger_error.assert_called_once_with(test_utils.
-                                                      SubstrMatch("error"
-                                                                  " [create"
-                                                                  "_vnf("
-                                                                  "tacker_"
-                                                                  "client"))
-            self.assertIsNone(resp)
-
-    def test_wait_for_vnf_vnf_retrieval_failed(self):
-        with mock.patch('functest.utils.openstack_tacker.get_vnf',
-                        return_value=None), \
-                self.assertRaises(Exception) as context:
-            openstack_tacker.wait_for_vnf(self.tacker_client,
-                                          vnf_id='vnf_id',
-                                          vnf_name='vnf_name')
-            msg = ("Could not retrieve VNF - id='vnf_id', "
-                   "name='vnf_name'")
-            self.assertTrue(msg in context)
-        with mock.patch('functest.utils.openstack_tacker.get_vnf',
-                        side_effect=Exception):
-            ret = openstack_tacker.wait_for_vnf(self.tacker_client,
-                                                vnf_id='vnf_id',
-                                                vnf_name='vnf_name')
-            self.assertEqual(ret, None)
-
-    def test_wait_for_vnf_vnf_status_error(self):
-        vnf = {'id': 'vnf_id',
-               'status': 'ERROR'}
-        with mock.patch('functest.utils.openstack_tacker.get_vnf',
-                        return_value=vnf), \
-                self.assertRaises(Exception) as context:
-            openstack_tacker.wait_for_vnf(self.tacker_client,
-                                          vnf_id='vnf_id',
-                                          vnf_name='vnf_name')
-            msg = ('Error when booting vnf vnf_id')
-            self.assertTrue(msg in context)
-
-    def test_wait_for_vnf_vnf_timeout(self):
-        vnf = {'id': 'vnf_id',
-               'status': 'PENDING_CREATE'}
-        with mock.patch('functest.utils.openstack_tacker.get_vnf',
-                        return_value=vnf), \
-                self.assertRaises(Exception) as context:
-            openstack_tacker.wait_for_vnf(self.tacker_client,
-                                          vnf_id='vnf_id',
-                                          vnf_name='vnf_name',
-                                          timeout=2)
-            msg = ('Timeout when booting vnf vnf_id')
-            self.assertTrue(msg in context)
-
-    def test_delete_vnf(self):
-        with mock.patch('functest.utils.openstack_tacker.get_vnf_id',
-                        return_value=self.vnf), \
-                mock.patch.object(self.tacker_client, 'delete_vnf',
-                                  return_value=self.vnf):
-            resp = openstack_tacker.delete_vnf(self.tacker_client,
-                                               vnf_id='vnf_id',
-                                               vnf_name=self.vnf)
-            self.assertEqual(resp, self.vnf)
-
-    def test_delete_vnf_missing_vnf_name(self):
-        with self.assertRaises(Exception) as context:
-            openstack_tacker.delete_vnf(self.tacker_client,
-                                        vnf_id=None,
-                                        vnf_name=None)
-            msg = 'You need to provide a VNF id or name'
-            self.assertTrue(msg in context)
-
-    @mock.patch('functest.utils.openstack_tacker.logger.error')
-    def test_delete_vnf_exception(self, mock_logger_error):
-        with mock.patch('functest.utils.openstack_tacker.get_vnf_id',
-                        return_value=self.vnf), \
-                mock.patch.object(self.tacker_client, 'delete_vnf',
-                                  side_effect=Exception):
-            resp = openstack_tacker.delete_vnf(self.tacker_client,
-                                               vnf_id=None,
-                                               vnf_name=None)
-            self.assertIsNone(resp)
-            self.assertTrue(mock_logger_error.called)
-
-    def test_list_sfcs(self):
-        with mock.patch.object(self.tacker_client, 'list_sfcs',
-                               return_value=self.sfclist):
-            resp = openstack_tacker.list_sfcs(self.tacker_client,
-                                              verbose=False)
-            self.assertEqual(resp, ['test_sfc1', 'test_sfc2'])
-
-    def test_list_sfcs_verbose(self):
-        with mock.patch.object(self.tacker_client, 'list_sfcs',
-                               return_value=self.sfclist):
-            resp = openstack_tacker.list_sfcs(self.tacker_client,
-                                              verbose=True)
-            self.assertEqual(resp, self.sfclist)
-
-    @mock.patch('functest.utils.openstack_tacker.logger.error')
-    def test_list_sfcs_exception(self, mock_logger_error):
-        with mock.patch.object(self.tacker_client, 'list_sfcs',
-                               side_effect=Exception):
-            resp = openstack_tacker.list_sfcs(self.tacker_client,
-                                              verbose=False)
-            mock_logger_error.assert_called_once_with(test_utils.
-                                                      SubstrMatch("Error"
-                                                                  " [list"
-                                                                  "_sfcs("
-                                                                  "tacker_"
-                                                                  "client)]:"))
-            self.assertIsNone(resp)
-
-    def test_create_sfc_default(self):
-        with mock.patch.object(self.tacker_client, 'create_sfc',
-                               return_value=self.createsfc), \
-                mock.patch('functest.utils.openstack_tacker.get_vnf_id',
-                           return_value=self.vnf):
-            resp = openstack_tacker.create_sfc(self.tacker_client,
-                                               sfc_name=self.sfc,
-                                               chain_vnf_ids=['chain_vnf_id'],
-                                               chain_vnf_names=[self.vnf])
-            self.assertEqual(resp, self.createsfc)
-
-    @mock.patch('functest.utils.openstack_tacker.logger.error')
-    def test_create_sfc_exception(self, mock_logger_error):
-        with mock.patch.object(self.tacker_client, 'create_sfc',
-                               side_effect=Exception):
-            resp = openstack_tacker.create_sfc(self.tacker_client,
-                                               sfc_name=self.sfc,
-                                               chain_vnf_ids=['chain_vnf_id'],
-                                               chain_vnf_names=[self.vnf])
-            mock_logger_error.assert_called_once_with(test_utils.
-                                                      SubstrMatch("error"
-                                                                  " [create"
-                                                                  "_sfc("
-                                                                  "tacker_"
-                                                                  "client"))
-            self.assertIsNone(resp)
-
-    def test_delete_sfc(self):
-        with mock.patch('functest.utils.openstack_tacker.get_sfc_id',
-                        return_value=self.sfc), \
-                mock.patch.object(self.tacker_client, 'delete_sfc',
-                                  return_value=self.sfc):
-            resp = openstack_tacker.delete_sfc(self.tacker_client,
-                                               sfc_id='sfc_id',
-                                               sfc_name=self.sfc)
-            self.assertEqual(resp, self.sfc)
-
-    def test_delete_sfc_missing_sfc_name(self):
-        with self.assertRaises(Exception) as context:
-            openstack_tacker.delete_sfc(self.tacker_client,
-                                        sfc_id=None,
-                                        sfc_name=None)
-            msg = 'You need to provide an SFC id or name'
-            self.assertTrue(msg in context)
-
-    @mock.patch('functest.utils.openstack_tacker.logger.error')
-    def test_delete_sfc_exception(self, mock_logger_error):
-        with mock.patch('functest.utils.openstack_tacker.get_sfc_id',
-                        return_value=self.sfc), \
-                mock.patch.object(self.tacker_client, 'delete_sfc',
-                                  side_effect=Exception):
-            resp = openstack_tacker.delete_sfc(self.tacker_client,
-                                               sfc_id=None,
-                                               sfc_name=None)
-            self.assertIsNone(resp)
-            self.assertTrue(mock_logger_error.called)
-
-    def test_list_sfc_classifiers(self):
-        with mock.patch.object(self.tacker_client, 'list_sfc_classifiers',
-                               return_value=self.sfc_classifierlist):
-            resp = openstack_tacker.list_sfc_classifiers(self.tacker_client,
-                                                         verbose=False)
-            self.assertEqual(resp, ['test_sfc_cl1', 'test_sfc_cl2'])
-
-    def test_list_sfc_classifiers_verbose(self):
-        with mock.patch.object(self.tacker_client, 'list_sfc_classifiers',
-                               return_value=self.sfc_classifierlist):
-            resp = openstack_tacker.list_sfc_classifiers(self.tacker_client,
-                                                         verbose=True)
-            self.assertEqual(resp, self.sfc_classifierlist)
-
-    @mock.patch('functest.utils.openstack_tacker.logger.error')
-    def test_list_sfc_classifiers_exception(self, mock_logger_error):
-        with mock.patch.object(self.tacker_client, 'list_sfc_classifiers',
-                               side_effect=Exception):
-            resp = openstack_tacker.list_sfc_classifiers(self.tacker_client,
-                                                         verbose=False)
-            mock_logger_error.assert_called_once_with(test_utils.
-                                                      SubstrMatch("Error"
-                                                                  " [list"
-                                                                  "_sfc_cl"
-                                                                  "assifiers("
-                                                                  "tacker_"
-                                                                  "client)]:"))
-            self.assertIsNone(resp)
-
-    def test_create_sfc_classifier_default(self):
-        with mock.patch.object(self.tacker_client, 'create_sfc_classifier',
-                               return_value=self.createsfc_clf), \
-                mock.patch('functest.utils.openstack_tacker.get_sfc_id',
-                           return_value=self.sfc):
-            cl = self.sfc_clf
-            resp = openstack_tacker.create_sfc_classifier(self.tacker_client,
-                                                          sfc_clf_name=cl,
-                                                          sfc_id='sfc_id',
-                                                          sfc_name=self.sfc)
-            self.assertEqual(resp, self.createsfc_clf)
-
-    @mock.patch('functest.utils.openstack_tacker.logger.error')
-    def test_sfc_classifier_exception(self, mock_logger_error):
-        with mock.patch.object(self.tacker_client, 'create_sfc_classifier',
-                               side_effect=Exception):
-            cl = self.sfc_clf
-            resp = openstack_tacker.create_sfc_classifier(self.tacker_client,
-                                                          sfc_clf_name=cl,
-                                                          sfc_id='sfc_id',
-                                                          sfc_name=self.sfc)
-            mock_logger_error.assert_called_once_with(test_utils.
-                                                      SubstrMatch("error"
-                                                                  " [create"
-                                                                  "_sfc_cl"
-                                                                  "assifier("
-                                                                  "tacker_"
-                                                                  "client"))
-            self.assertIsNone(resp)
-
-    def test_delete_sfc_classifier(self):
-        with mock.patch('functest.utils.openstack_tacker.get_sfc_'
-                        'classifier_id',
-                        return_value=self.sfc_clf), \
-                mock.patch.object(self.tacker_client, 'delete_sfc_classifier',
-                                  return_value=self.sfc_clf):
-            cl = self.sfc_clf
-            resp = openstack_tacker.delete_sfc_classifier(self.tacker_client,
-                                                          sfc_clf_id='sfc_id',
-                                                          sfc_clf_name=cl)
-            self.assertEqual(resp, cl)
-
-    def test_delete_sfc_classifier_missing_sfc_name(self):
-        with self.assertRaises(Exception) as context:
-            openstack_tacker.delete_vnf(self.tacker_client,
-                                        sfc_clf_id=None,
-                                        sfc_clf_name=None)
-            msg = 'You need to provide an SFCclassifier id or name'
-            self.assertTrue(msg in context)
-
-    @mock.patch('functest.utils.openstack_tacker.logger.error')
-    def test_delete_sfc_classifier_exception(self, mock_logger_error):
-        with mock.patch('functest.utils.openstack_tacker.get_sfc_'
-                        'classifier_id',
-                        return_value=self.sfc_clf), \
-                mock.patch.object(self.tacker_client, 'delete_sfc_classifier',
-                                  side_effect=Exception):
-            cl = self.sfc_clf
-            resp = openstack_tacker.delete_sfc_classifier(self.tacker_client,
-                                                          sfc_clf_id='sfc_id',
-                                                          sfc_clf_name=cl)
-            self.assertIsNone(resp)
-            self.assertTrue(mock_logger_error.called)
-
-
-if __name__ == "__main__":
-    logging.disable(logging.CRITICAL)
-    unittest.main(verbosity=2)
diff --git a/functest/utils/openstack_tacker.py b/functest/utils/openstack_tacker.py
deleted file mode 100644
index 9fd9d5c4..00000000
--- a/functest/utils/openstack_tacker.py
+++ /dev/null
@@ -1,317 +0,0 @@
-###########################################################################
-# Copyright (c) 2016 Ericsson AB and others.
-# Author: George Paraskevopoulos <geopar@intracom-telecom.com>
-#
-# Wrappers for trozet's python-tackerclient v1.0
-# (https://github.com/trozet/python-tackerclient)
-#
-# All rights reserved. This program and the accompanying materials
-# are made available under the terms of the Apache License, Version 2.0
-# which accompanies this distribution, and is available at
-# http://www.apache.org/licenses/LICENSE-2.0
-##########################################################################
-
-import logging
-
-from tackerclient.v1_0 import client as tackerclient
-import functest.utils.openstack_utils as os_utils
-import time
-
-logger = logging.getLogger(__name__)
-
-
-def get_tacker_client(other_creds={}):
-    sess = os_utils.get_session(other_creds)
-    return tackerclient.Client(session=sess)
-
-
-# *********************************************
-#   TACKER
-# *********************************************
-def get_id_from_name(tacker_client, resource_type, resource_name):
-    try:
-        req_params = {'fields': 'id', 'name': resource_name}
-        endpoint = '/{0}s'.format(resource_type)
-        resp = tacker_client.get(endpoint, params=req_params)
-        endpoint = endpoint.replace('-', '_')
-        return resp[endpoint[1:]][0]['id']
-    except Exception, e:
-        logger.error("Error [get_id_from_name(tacker_client, "
-                     "resource_type, resource_name)]: %s" % e)
-        return None
-
-
-def get_vnfd_id(tacker_client, vnfd_name):
-    return get_id_from_name(tacker_client, 'vnfd', vnfd_name)
-
-
-def get_vnf_id(tacker_client, vnf_name, timeout=5):
-    vnf_id = None
-    while vnf_id is None and timeout >= 0:
-        vnf_id = get_id_from_name(tacker_client, 'vnf', vnf_name)
-        if vnf_id is None:
-            logger.info("Could not retrieve ID for vnf with name [%s]."
-                        " Retrying." % vnf_name)
-            time.sleep(1)
-            timeout -= 1
-    return vnf_id
-
-
-def get_sfc_id(tacker_client, sfc_name):
-    return get_id_from_name(tacker_client, 'sfc', sfc_name)
-
-
-def get_sfc_classifier_id(tacker_client, sfc_clf_name):
-    return get_id_from_name(tacker_client, 'sfc-classifier', sfc_clf_name)
-
-
-def list_vnfds(tacker_client, verbose=False):
-    try:
-        vnfds = tacker_client.list_vnfds(retrieve_all=True)
-        if not verbose:
-            vnfds = [vnfd['id'] for vnfd in vnfds['vnfds']]
-        return vnfds
-    except Exception, e:
-        logger.error("Error [list_vnfds(tacker_client)]: %s" % e)
-        return None
-
-
-def create_vnfd(tacker_client, tosca_file=None):
-    try:
-        vnfd_body = {}
-        if tosca_file is not None:
-            with open(tosca_file) as tosca_fd:
-                vnfd_body = tosca_fd.read()
-            logger.info('VNFD template:\n{0}'.format(vnfd_body))
-        return tacker_client.create_vnfd(
-            body={"vnfd": {"attributes": {"vnfd": vnfd_body}}})
-    except Exception, e:
-        logger.error("Error [create_vnfd(tacker_client, '%s')]: %s"
-                     % (tosca_file, e))
-        return None
-
-
-def delete_vnfd(tacker_client, vnfd_id=None, vnfd_name=None):
-    try:
-        vnfd = vnfd_id
-        if vnfd is None:
-            if vnfd_name is None:
-                raise Exception('You need to provide VNFD id or VNFD name')
-            vnfd = get_vnfd_id(tacker_client, vnfd_name)
-        return tacker_client.delete_vnfd(vnfd)
-    except Exception, e:
-        logger.error("Error [delete_vnfd(tacker_client, '%s', '%s')]: %s"
-                     % (vnfd_id, vnfd_name, e))
-        return None
-
-
-def list_vnfs(tacker_client, verbose=False):
-    try:
-        vnfs = tacker_client.list_vnfs(retrieve_all=True)
-        if not verbose:
-            vnfs = [vnf['id'] for vnf in vnfs['vnfs']]
-        return vnfs
-    except Exception, e:
-        logger.error("Error [list_vnfs(tacker_client)]: %s" % e)
-        return None
-
-
-def create_vnf(tacker_client, vnf_name, vnfd_id=None,
-               vnfd_name=None, param_file=None):
-    try:
-        vnf_body = {
-            'vnf': {
-                'attributes': {},
-                'name': vnf_name
-            }
-        }
-        if param_file is not None:
-            params = None
-            with open(param_file) as f:
-                params = f.read()
-            vnf_body['vnf']['attributes']['param_values'] = params
-        if vnfd_id is not None:
-            vnf_body['vnf']['vnfd_id'] = vnfd_id
-        else:
-            if vnfd_name is None:
-                raise Exception('vnfd id or vnfd name is required')
-            vnf_body['vnf']['vnfd_id'] = get_vnfd_id(tacker_client, vnfd_name)
-        return tacker_client.create_vnf(body=vnf_body)
-    except Exception, e:
-        logger.error("error [create_vnf(tacker_client,"
-                     " '%s', '%s', '%s')]: %s"
-                     % (vnf_name, vnfd_id, vnfd_name, e))
-        return None
-
-
-def get_vnf(tacker_client, vnf_id=None, vnf_name=None):
-    try:
-        if vnf_id is None and vnf_name is None:
-            raise Exception('You must specify vnf_id or vnf_name')
-
-        _id = get_vnf_id(tacker_client, vnf_name) if vnf_id is None else vnf_id
-
-        if _id is not None:
-            all_vnfs = list_vnfs(tacker_client, verbose=True)['vnfs']
-            return next((vnf for vnf in all_vnfs if vnf['id'] == _id), None)
-        else:
-            raise Exception('Could not retrieve ID from name [%s]' % vnf_name)
-
-    except Exception, e:
-        logger.error("Could not retrieve VNF [vnf_id=%s, vnf_name=%s] - %s"
-                     % (vnf_id, vnf_name, e))
-        return None
-
-
-def wait_for_vnf(tacker_client, vnf_id=None, vnf_name=None, timeout=60):
-    try:
-        vnf = get_vnf(tacker_client, vnf_id, vnf_name)
-        if vnf is None:
-            raise Exception("Could not retrieve VNF - id='%s', name='%s'"
-                            % vnf_id, vnf_name)
-        logger.info('Waiting for vnf {0}'.format(str(vnf)))
-        while vnf['status'] != 'ACTIVE' and timeout >= 0:
-            if vnf['status'] == 'ERROR':
-                raise Exception('Error when booting vnf %s' % vnf['id'])
-            elif vnf['status'] == 'PENDING_CREATE':
-                time.sleep(3)
-                timeout -= 3
-            vnf = get_vnf(tacker_client, vnf_id, vnf_name)
-
-        if (timeout < 0):
-            raise Exception('Timeout when booting vnf %s' % vnf['id'])
-
-        return vnf['id']
-    except Exception, e:
-        logger.error("error [wait_for_vnf(tacker_client, '%s', '%s')]: %s"
-                     % (vnf_id, vnf_name, e))
-        return None
-
-
-def delete_vnf(tacker_client, vnf_id=None, vnf_name=None):
-    try:
-        vnf = vnf_id
-        if vnf is None:
-            if vnf_name is None:
-                raise Exception('You need to provide a VNF id or name')
-            vnf = get_vnf_id(tacker_client, vnf_name)
-        return tacker_client.delete_vnf(vnf)
-    except Exception, e:
-        logger.error("Error [delete_vnf(tacker_client, '%s', '%s')]: %s"
-                     % (vnf_id, vnf_name, e))
-        return None
-
-
-def list_sfcs(tacker_client, verbose=False):
-    try:
-        sfcs = tacker_client.list_sfcs(retrieve_all=True)
-        if not verbose:
-            sfcs = [sfc['id'] for sfc in sfcs['sfcs']]
-        return sfcs
-    except Exception, e:
-        logger.error("Error [list_sfcs(tacker_client)]: %s" % e)
-        return None
-
-
-def create_sfc(tacker_client, sfc_name,
-               chain_vnf_ids=None,
-               chain_vnf_names=None,
-               symmetrical=False):
-    try:
-        sfc_body = {
-            'sfc': {
-                'attributes': {},
-                'name': sfc_name,
-                'chain': []
-            }
-        }
-        if symmetrical:
-            sfc_body['sfc']['symmetrical'] = True
-        if chain_vnf_ids is not None:
-            sfc_body['sfc']['chain'] = chain_vnf_ids
-        else:
-            if chain_vnf_names is None:
-                raise Exception('You need to provide a chain of VNFs')
-            sfc_body['sfc']['chain'] = [get_vnf_id(tacker_client, name)
-                                        for name in chain_vnf_names]
-        return tacker_client.create_sfc(body=sfc_body)
-    except Exception, e:
-        logger.error("error [create_sfc(tacker_client,"
-                     " '%s', '%s', '%s')]: %s"
-                     % (sfc_name, chain_vnf_ids, chain_vnf_names, e))
-        return None
-
-
-def delete_sfc(tacker_client, sfc_id=None, sfc_name=None):
-    try:
-        sfc = sfc_id
-        if sfc is None:
-            if sfc_name is None:
-                raise Exception('You need to provide an SFC id or name')
-            sfc = get_sfc_id(tacker_client, sfc_name)
-        return tacker_client.delete_sfc(sfc)
-    except Exception, e:
-        logger.error("Error [delete_sfc(tacker_client, '%s', '%s')]: %s"
-                     % (sfc_id, sfc_name, e))
-        return None
-
-
-def list_sfc_classifiers(tacker_client, verbose=False):
-    try:
-        sfc_clfs = tacker_client.list_sfc_classifiers(retrieve_all=True)
-        if not verbose:
-            sfc_clfs = [sfc_clf['id']
-                        for sfc_clf in sfc_clfs['sfc_classifiers']]
-        return sfc_clfs
-    except Exception, e:
-        logger.error("Error [list_sfc_classifiers(tacker_client)]: %s" % e)
-        return None
-
-
-def create_sfc_classifier(tacker_client, sfc_clf_name, sfc_id=None,
-                          sfc_name=None, match={}):
-    # Example match:
-    # match: {
-    #     "source_port": "0",
-    #     "protocol": "6",
-    #     "dest_port": "80"
-    # }
-    try:
-        sfc_clf_body = {
-            'sfc_classifier': {
-                'attributes': {},
-                'name': sfc_clf_name,
-                'match': match,
-                'chain': ''
-            }
-        }
-        if sfc_id is not None:
-            sfc_clf_body['sfc_classifier']['chain'] = sfc_id
-        else:
-            if sfc_name is None:
-                raise Exception('You need to provide an SFC id or name')
-            sfc_clf_body['sfc_classifier']['chain'] = get_sfc_id(
-                tacker_client, sfc_name)
-        return tacker_client.create_sfc_classifier(body=sfc_clf_body)
-    except Exception, e:
-        logger.error("error [create_sfc_classifier(tacker_client,"
-                     " '%s', '%s','%s', '%s')]: '%s'"
-                     % (sfc_clf_name, sfc_id, sfc_name, str(match), e))
-        return None
-
-
-def delete_sfc_classifier(tacker_client,
-                          sfc_clf_id=None,
-                          sfc_clf_name=None):
-    try:
-        sfc_clf = sfc_clf_id
-        if sfc_clf is None:
-            if sfc_clf_name is None:
-                raise Exception('You need to provide an SFC'
-                                'classifier id or name')
-            sfc_clf = get_sfc_classifier_id(tacker_client, sfc_clf_name)
-        return tacker_client.delete_sfc_classifier(sfc_clf)
-    except Exception, e:
-        logger.error("Error [delete_sfc_classifier(tacker_client, '%s', "
-                     "'%s')]: %s" % (sfc_clf_id, sfc_clf_name, e))
-        return None