functest/tests/unit/utils/test_openstack_tacker.py


#!/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):

    logging.disable(logging.CRITICAL)

    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__":
    unittest.main(verbosity=2)