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