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