# http://www.apache.org/licenses/LICENSE-2.0
import logging
-import mock
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'}],
}
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):
tosca_file=None)
self.assertEqual(resp, self.createvnfd)
- @mock.patch('functest.utils.openstack_tacker.logger.error')
- def test_create_vnfd_default(self, mock_logger_error):
+ 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')) \
resp = openstack_tacker.create_vnfd(self.tacker_client,
tosca_file=self.tosca_file)
m.assert_called_once_with(self.tosca_file)
- mock_logger_error.assert_called_once_with('1')
self.assertEqual(resp, self.createvnfd)
- @mock.patch('functest.utils.openstack_tacker.logger.exception')
- def test_create_vnfd_exception(self, mock_logger_excep):
+ @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_excep.assert_called_once_with(test_utils.
+ mock_logger_error.assert_called_once_with(test_utils.
SubstrMatch("Error"
" [create"
"_vnfd("
vnfd_name=self.vnfd)
self.assertEqual(resp, self.vnfd)
- # TODO: Exception('You need to provide an VNFD'
- # 'id or name') AssertionError
+ 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):
"client"))
self.assertIsNone(resp)
- # TODO: wait_for_vnf
+ 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',
vnf_name=self.vnf)
self.assertEqual(resp, self.vnf)
- # TODO: Exception('You need to provide an VNF'
- # 'classifier id or name') AssertionError
+ 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):
sfc_name=self.sfc)
self.assertEqual(resp, self.sfc)
- # TODO: Exception('You need to provide an SFC'
- # 'id or name') AssertionError
+ 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):
sfc_clf_name=cl)
self.assertEqual(resp, cl)
- # TODO: Exception('You need to provide an SFC'
- # 'classifier id or name') AssertionError
+ 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):
if __name__ == "__main__":
+ logging.disable(logging.CRITICAL)
unittest.main(verbosity=2)