projects / osm / RO.git / blob
? search:
summary | shortlog | log | commit | commitdiff | tree
history | raw | HEAD
System tests fix for openstack. Flavor and Network related system tests are fixed...
[osm/RO.git] / test / test_RO.py
1 #!/usr/bin/env python2
2 # -*- coding: utf-8 -*-
3
4 ##
5 # Copyright 2017
6 # This file is part of openmano
7 # All Rights Reserved.
8 #
9 # Licensed under the Apache License, Version 2.0 (the "License"); you may
10 # not use this file except in compliance with the License. You may obtain
11 # a copy of the License at
12 #
13 # http://www.apache.org/licenses/LICENSE-2.0
14 #
15 # Unless required by applicable law or agreed to in writing, software
16 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
17 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
18 # License for the specific language governing permissions and limitations
19 # under the License.
20 #
21 ##
22
23 """
24 Module for testing openmano functionality. It uses openmanoclient.py for invoking openmano
25 """
26
27 import logging
28 import os
29 import argcomplete
30 import unittest
31 import string
32 import inspect
33 import random
34 # import traceback
35 import glob
36 import yaml
37 import sys
38 import time
39 import uuid
40 from argparse import ArgumentParser
41
42 __author__ = "Pablo Montes, Alfonso Tierno"
43 __date__ = "$16-Feb-2017 17:08:16$"
44 __version__ = "0.1.0"
45 version_date = "Oct 2017"
46
47 test_config = {} # used for global variables with the test configuration
48
49
50 class test_base(unittest.TestCase):
51 test_index = 1
52 test_text = None
53
54 @classmethod
55 def setUpClass(cls):
56 logger.info("{}. {}".format(test_config["test_number"], cls.__name__))
57
58 @classmethod
59 def tearDownClass(cls):
60 test_config["test_number"] += 1
61
62 def tearDown(self):
63 exec_info = sys.exc_info()
64 if exec_info == (None, None, None):
65 logger.info(self.__class__.test_text+" -> TEST OK")
66 else:
67 logger.warning(self.__class__.test_text+" -> TEST NOK")
68 logger.critical("Traceback error",exc_info=True)
69
70
71 def check_instance_scenario_active(uuid):
72 instance = test_config["client"].get_instance(uuid=uuid)
73
74 for net in instance['nets']:
75 status = net['status']
76 if status != 'ACTIVE':
77 return (False, status)
78
79 for vnf in instance['vnfs']:
80 for vm in vnf['vms']:
81 status = vm['status']
82 if status != 'ACTIVE':
83 return (False, status)
84
85 return (True, None)
86
87
88 '''
89 IMPORTANT NOTE
90 All unittest classes for code based tests must have prefix 'test_' in order to be taken into account for tests
91 '''
92 class test_VIM_datacenter_tenant_operations(test_base):
93 tenant_name = None
94
95 def test_000_create_RO_tenant(self):
96 self.__class__.tenant_name = _get_random_string(20)
97 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
98 inspect.currentframe().f_code.co_name)
99 self.__class__.test_index += 1
100 tenant = test_config["client"].create_tenant(name=self.__class__.tenant_name,
101 description=self.__class__.tenant_name)
102 logger.debug("{}".format(tenant))
103 self.assertEqual(tenant.get('tenant', {}).get('name', ''), self.__class__.tenant_name)
104
105 def test_010_list_RO_tenant(self):
106 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
107 inspect.currentframe().f_code.co_name)
108 self.__class__.test_index += 1
109 tenant = test_config["client"].get_tenant(name=self.__class__.tenant_name)
110 logger.debug("{}".format(tenant))
111 self.assertEqual(tenant.get('tenant', {}).get('name', ''), self.__class__.tenant_name)
112
113 def test_020_delete_RO_tenant(self):
114 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
115 inspect.currentframe().f_code.co_name)
116 self.__class__.test_index += 1
117 tenant = test_config["client"].delete_tenant(name=self.__class__.tenant_name)
118 logger.debug("{}".format(tenant))
119 assert('deleted' in tenant.get('result',""))
120
121
122 class test_VIM_datacenter_operations(test_base):
123 datacenter_name = None
124
125 def test_000_create_datacenter(self):
126 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
127 inspect.currentframe().f_code.co_name)
128 self.__class__.datacenter_name = _get_random_string(20)
129 self.__class__.test_index += 1
130 self.datacenter = test_config["client"].create_datacenter(name=self.__class__.datacenter_name,
131 vim_url="http://fakeurl/fake")
132 logger.debug("{}".format(self.datacenter))
133 self.assertEqual (self.datacenter.get('datacenter', {}).get('name',''), self.__class__.datacenter_name)
134
135 def test_010_list_datacenter(self):
136 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
137 inspect.currentframe().f_code.co_name)
138
139 self.__class__.test_index += 1
140 self.datacenter = test_config["client"].get_datacenter(all_tenants=True, name=self.__class__.datacenter_name)
141 logger.debug("{}".format(self.datacenter))
142 self.assertEqual (self.datacenter.get('datacenter', {}).get('name', ''), self.__class__.datacenter_name)
143
144 def test_020_attach_datacenter(self):
145 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
146 inspect.currentframe().f_code.co_name)
147
148 self.__class__.test_index += 1
149 self.datacenter = test_config["client"].attach_datacenter(name=self.__class__.datacenter_name,
150 vim_tenant_name='fake')
151 logger.debug("{}".format(self.datacenter))
152 assert ('uuid' in self.datacenter.get('datacenter', {}))
153
154 def test_030_list_attached_datacenter(self):
155 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
156 inspect.currentframe().f_code.co_name)
157
158 self.__class__.test_index += 1
159 self.datacenter = test_config["client"].get_datacenter(all_tenants=False, name=self.__class__.datacenter_name)
160 logger.debug("{}".format(self.datacenter))
161 self.assertEqual (self.datacenter.get('datacenter', {}).get('name', ''), self.__class__.datacenter_name)
162
163 def test_040_detach_datacenter(self):
164 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
165 inspect.currentframe().f_code.co_name)
166
167 self.__class__.test_index += 1
168 self.datacenter = test_config["client"].detach_datacenter(name=self.__class__.datacenter_name)
169 logger.debug("{}".format(self.datacenter))
170 assert ('detached' in self.datacenter.get('result', ""))
171
172 def test_050_delete_datacenter(self):
173 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
174 inspect.currentframe().f_code.co_name)
175
176 self.__class__.test_index += 1
177 self.datacenter = test_config["client"].delete_datacenter(name=self.__class__.datacenter_name)
178 logger.debug("{}".format(self.datacenter))
179 assert('deleted' in self.datacenter.get('result',""))
180
181
182 class test_VIM_network_operations(test_base):
183 vim_network_name = None
184 vim_network_uuid = None
185
186 def test_000_create_VIM_network(self):
187 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
188 inspect.currentframe().f_code.co_name)
189 self.__class__.vim_network_name = _get_random_string(20)
190 self.__class__.test_index += 1
191 network = test_config["client"].vim_action("create", "networks", name=self.__class__.vim_network_name)
192 logger.debug("{}".format(network))
193 self.__class__.vim_network_uuid = network["network"]["id"]
194 self.assertEqual(network.get('network', {}).get('name', ''), self.__class__.vim_network_name)
195
196 def test_010_list_VIM_networks(self):
197 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
198 inspect.currentframe().f_code.co_name)
199 self.__class__.test_index += 1
200 networks = test_config["client"].vim_action("list", "networks")
201 logger.debug("{}".format(networks))
202
203 def test_020_get_VIM_network_by_uuid(self):
204 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
205 inspect.currentframe().f_code.co_name)
206
207 self.__class__.test_index += 1
208 network = test_config["client"].vim_action("show", "networks", uuid=self.__class__.vim_network_uuid)
209 logger.debug("{}".format(network))
210 self.assertEqual(network.get('network', {}).get('name', ''), self.__class__.vim_network_name)
211
212 def test_030_delete_VIM_network_by_uuid(self):
213 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
214 inspect.currentframe().f_code.co_name)
215
216 self.__class__.test_index += 1
217 network = test_config["client"].vim_action("delete", "networks", uuid=self.__class__.vim_network_uuid)
218 logger.debug("{}".format(network))
219 assert ('deleted' in network.get('result', ""))
220
221
222 class test_VIM_image_operations(test_base):
223
224 def test_000_list_VIM_images(self):
225 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
226 inspect.currentframe().f_code.co_name)
227 self.__class__.test_index += 1
228 images = test_config["client"].vim_action("list", "images")
229 logger.debug("{}".format(images))
230
231 '''
232 The following is a non critical test that will fail most of the times.
233 In case of OpenStack datacenter these tests will only success if RO has access to the admin endpoint
234 This test will only be executed in case it is specifically requested by the user
235 '''
236 class test_VIM_tenant_operations(test_base):
237 vim_tenant_name = None
238 vim_tenant_uuid = None
239
240 @classmethod
241 def setUpClass(cls):
242 test_base.setUpClass(cls)
243 logger.warning("In case of OpenStack datacenter these tests will only success "
244 "if RO has access to the admin endpoint")
245
246 def test_000_create_VIM_tenant(self):
247 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
248 inspect.currentframe().f_code.co_name)
249 self.__class__.vim_tenant_name = _get_random_string(20)
250 self.__class__.test_index += 1
251 tenant = test_config["client"].vim_action("create", "tenants", name=self.__class__.vim_tenant_name)
252 logger.debug("{}".format(tenant))
253 self.__class__.vim_tenant_uuid = tenant["tenant"]["id"]
254 self.assertEqual(tenant.get('tenant', {}).get('name', ''), self.__class__.vim_tenant_name)
255
256 def test_010_list_VIM_tenants(self):
257 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
258 inspect.currentframe().f_code.co_name)
259 self.__class__.test_index += 1
260 tenants = test_config["client"].vim_action("list", "tenants")
261 logger.debug("{}".format(tenants))
262
263 def test_020_get_VIM_tenant_by_uuid(self):
264 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
265 inspect.currentframe().f_code.co_name)
266
267 self.__class__.test_index += 1
268 tenant = test_config["client"].vim_action("show", "tenants", uuid=self.__class__.vim_tenant_uuid)
269 logger.debug("{}".format(tenant))
270 self.assertEqual(tenant.get('tenant', {}).get('name', ''), self.__class__.vim_tenant_name)
271
272 def test_030_delete_VIM_tenant_by_uuid(self):
273 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
274 inspect.currentframe().f_code.co_name)
275
276 self.__class__.test_index += 1
277 tenant = test_config["client"].vim_action("delete", "tenants", uuid=self.__class__.vim_tenant_uuid)
278 logger.debug("{}".format(tenant))
279 assert ('deleted' in tenant.get('result', ""))
280
281
282 class test_vimconn_connect(test_base):
283
284 def test_000_connect(self):
285 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
286 self.__class__.test_index,
287 inspect.currentframe().f_code.co_name)
288
289 self.__class__.test_index += 1
290 if test_config['vimtype'] == 'vmware':
291 vca_object = test_config["vim_conn"].connect()
292 logger.debug("{}".format(vca_object))
293 self.assertIsNotNone(vca_object)
294 elif test_config['vimtype'] == 'openstack':
295 test_config["vim_conn"]._reload_connection()
296 network_list = test_config["vim_conn"].get_network_list()
297 logger.debug("{}".format(network_list))
298 self.assertIsNotNone(network_list)
299
300 class test_vimconn_new_network(test_base):
301 network_name = None
302
303 def test_000_new_network(self):
304 self.__class__.network_name = _get_random_string(20)
305 network_type = 'bridge'
306
307 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
308 self.__class__.test_index, inspect.currentframe().f_code.co_name)
309 self.__class__.test_index += 1
310
311 network = test_config["vim_conn"].new_network(net_name=self.__class__.network_name,
312 net_type=network_type)
313 self.__class__.network_id = network
314 logger.debug("{}".format(network))
315
316 network_list = test_config["vim_conn"].get_network_list()
317 for net in network_list:
318 if self.__class__.network_name in net.get('name'):
319 self.assertIn(self.__class__.network_name, net.get('name'))
320 self.assertEqual(net.get('type'), network_type)
321
322 # Deleting created network
323 result = test_config["vim_conn"].delete_network(self.__class__.network_id)
324 if result:
325 logger.info("Network id {} sucessfully deleted".format(self.__class__.network_id))
326 else:
327 logger.info("Failed to delete network id {}".format(self.__class__.network_id))
328
329 def test_010_new_network_by_types(self):
330 delete_net_ids = []
331 network_types = ['data','bridge','mgmt']
332 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
333 self.__class__.test_index,
334 inspect.currentframe().f_code.co_name)
335 self.__class__.test_index += 1
336 for net_type in network_types:
337 self.__class__.network_name = _get_random_string(20)
338 network_id = test_config["vim_conn"].new_network(net_name=self.__class__.network_name,
339 net_type=net_type)
340
341 delete_net_ids.append(network_id)
342 logger.debug("{}".format(network_id))
343
344 network_list = test_config["vim_conn"].get_network_list()
345 for net in network_list:
346 if self.__class__.network_name in net.get('name'):
347 self.assertIn(self.__class__.network_name, net.get('name'))
348 if net_type in net.get('type'):
349 self.assertEqual(net.get('type'), net_type)
350 else:
351 self.assertNotEqual(net.get('type'), net_type)
352
353 # Deleting created network
354 for net_id in delete_net_ids:
355 result = test_config["vim_conn"].delete_network(net_id)
356 if result:
357 logger.info("Network id {} sucessfully deleted".format(net_id))
358 else:
359 logger.info("Failed to delete network id {}".format(net_id))
360
361 def test_020_new_network_by_ipprofile(self):
362 test_directory_content = os.listdir(test_config["test_directory"])
363
364 for dir_name in test_directory_content:
365 if dir_name == 'simple_multi_vnfc':
366 self.__class__.scenario_test_path = test_config["test_directory"] + '/'+ dir_name
367 vnfd_files = glob.glob(self.__class__.scenario_test_path+'/vnfd_*.yaml')
368 break
369
370 for vnfd in vnfd_files:
371 with open(vnfd, 'r') as stream:
372 vnf_descriptor = yaml.load(stream)
373
374 internal_connections_list = vnf_descriptor['vnf']['internal-connections']
375 for item in internal_connections_list:
376 if 'ip-profile' in item:
377 version = item['ip-profile']['ip-version']
378 dhcp_count = item['ip-profile']['dhcp']['count']
379 dhcp_enabled = item['ip-profile']['dhcp']['enabled']
380 dhcp_start_address = item['ip-profile']['dhcp']['start-address']
381 subnet_address = item['ip-profile']['subnet-address']
382
383
384 self.__class__.network_name = _get_random_string(20)
385 ip_profile = {'dhcp_count': dhcp_count,
386 'dhcp_enabled': dhcp_enabled,
387 'dhcp_start_address': dhcp_start_address,
388 'ip_version': version,
389 'subnet_address': subnet_address
390 }
391 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
392 self.__class__.test_index,
393 inspect.currentframe().f_code.co_name)
394 self.__class__.test_index += 1
395 network = test_config["vim_conn"].new_network(net_name=self.__class__.network_name,
396 net_type='mgmt',
397 ip_profile=ip_profile)
398 self.__class__.network_id = network
399 logger.debug("{}".format(network))
400
401 network_list = test_config["vim_conn"].get_network_list()
402 for net in network_list:
403 if self.__class__.network_name in net.get('name'):
404 self.assertIn(self.__class__.network_name, net.get('name'))
405
406 # Deleting created network
407 result = test_config["vim_conn"].delete_network(self.__class__.network_id)
408 if result:
409 logger.info("Network id {} sucessfully deleted".format(self.__class__.network_id))
410 else:
411 logger.info("Failed to delete network id {}".format(self.__class__.network_id))
412
413 def test_030_new_network_by_isshared(self):
414 self.__class__.network_name = _get_random_string(20)
415 shared = True
416 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
417 self.__class__.test_index,
418 inspect.currentframe().f_code.co_name)
419 self.__class__.test_index += 1
420 network = test_config["vim_conn"].new_network(net_name=self.__class__.network_name,
421 net_type='bridge',
422 shared=shared)
423 self.__class__.network_id = network
424 logger.debug("{}".format(network))
425
426 network_list = test_config["vim_conn"].get_network_list()
427 for net in network_list:
428 if self.__class__.network_name in net.get('name'):
429 self.assertIn(self.__class__.network_name, net.get('name'))
430 self.assertEqual(net.get('shared'), shared)
431
432 # Deleting created network
433 result = test_config["vim_conn"].delete_network(self.__class__.network_id)
434 if result:
435 logger.info("Network id {} sucessfully deleted".format(self.__class__.network_id))
436 else:
437 logger.info("Failed to delete network id {}".format(self.__class__.network_id))
438
439 def test_040_new_network_by_negative(self):
440 self.__class__.network_name = _get_random_string(20)
441 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
442 self.__class__.test_index,
443 inspect.currentframe().f_code.co_name)
444 self.__class__.test_index += 1
445 network = test_config["vim_conn"].new_network(net_name=self.__class__.network_name,
446 net_type='unknowntype')
447 self.__class__.network_id = network
448 logger.debug("{}".format(network))
449 network_list = test_config["vim_conn"].get_network_list()
450 for net in network_list:
451 if self.__class__.network_name in net.get('name'):
452 self.assertIn(self.__class__.network_name, net.get('name'))
453
454 # Deleting created network
455 result = test_config["vim_conn"].delete_network(self.__class__.network_id)
456 if result:
457 logger.info("Network id {} sucessfully deleted".format(self.__class__.network_id))
458 else:
459 logger.info("Failed to delete network id {}".format(self.__class__.network_id))
460
461 def test_050_refresh_nets_status(self):
462 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
463 self.__class__.test_index,
464 inspect.currentframe().f_code.co_name)
465 self.__class__.test_index += 1
466 # creating new network
467 network_name = _get_random_string(20)
468 net_type = 'bridge'
469 network_id = test_config["vim_conn"].new_network(net_name=network_name,
470 net_type=net_type)
471 # refresh net status
472 net_dict = test_config["vim_conn"].refresh_nets_status([network_id])
473 for attr in net_dict[network_id]:
474 if attr == 'status':
475 self.assertEqual(net_dict[network_id][attr], 'ACTIVE')
476
477 # Deleting created network
478 result = test_config["vim_conn"].delete_network(network_id)
479 if result:
480 logger.info("Network id {} sucessfully deleted".format(network_id))
481 else:
482 logger.info("Failed to delete network id {}".format(network_id))
483
484 def test_060_refresh_nets_status_negative(self):
485 unknown_net_id = str(uuid.uuid4())
486 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
487 self.__class__.test_index,
488 inspect.currentframe().f_code.co_name)
489 self.__class__.test_index += 1
490
491 # refresh net status
492 net_dict = test_config["vim_conn"].refresh_nets_status([unknown_net_id])
493 if test_config['vimtype'] == 'openstack':
494 self.assertEqual(net_dict[unknown_net_id]['status'], 'DELETED')
495 else:
496 # TODO : Fix vmware connector to return status DELETED as per vimconn.py
497 self.assertEqual(net_dict, {})
498
499 class test_vimconn_get_network_list(test_base):
500 network_name = None
501
502 def setUp(self):
503 # creating new network
504 self.__class__.network_name = _get_random_string(20)
505 self.__class__.net_type = 'bridge'
506 network = test_config["vim_conn"].new_network(net_name=self.__class__.network_name,
507 net_type=self.__class__.net_type)
508 self.__class__.network_id = network
509 logger.debug("{}".format(network))
510
511 def tearDown(self):
512 test_base.tearDown(self)
513
514 # Deleting created network
515 result = test_config["vim_conn"].delete_network(self.__class__.network_id)
516 if result:
517 logger.info("Network id {} sucessfully deleted".format(self.__class__.network_id))
518 else:
519 logger.info("Failed to delete network id {}".format(self.__class__.network_id))
520
521 def test_000_get_network_list(self):
522 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
523 self.__class__.test_index,
524 inspect.currentframe().f_code.co_name)
525 self.__class__.test_index += 1
526
527 network_list = test_config["vim_conn"].get_network_list()
528 for net in network_list:
529 if self.__class__.network_name in net.get('name'):
530 self.assertIn(self.__class__.network_name, net.get('name'))
531 self.assertEqual(net.get('type'), self.__class__.net_type)
532 self.assertEqual(net.get('status'), 'ACTIVE')
533 self.assertEqual(net.get('shared'), False)
534
535 def test_010_get_network_list_by_name(self):
536 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
537 self.__class__.test_index,
538 inspect.currentframe().f_code.co_name)
539 self.__class__.test_index += 1
540
541 if test_config['vimtype'] == 'openstack':
542 network_name = test_config['vim_conn'].get_network(self.__class__.network_id)['name']
543 else:
544 network_name = test_config['vim_conn'].get_network_name_by_id(self.__class__.network_id)
545
546 # find network from list by it's name
547 new_network_list = test_config["vim_conn"].get_network_list({'name': network_name})
548 for list_item in new_network_list:
549 if self.__class__.network_name in list_item.get('name'):
550 self.assertEqual(network_name, list_item.get('name'))
551 self.assertEqual(list_item.get('type'), self.__class__.net_type)
552 self.assertEqual(list_item.get('status'), 'ACTIVE')
553
554 def test_020_get_network_list_by_id(self):
555 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
556 self.__class__.test_index,
557 inspect.currentframe().f_code.co_name)
558 self.__class__.test_index += 1
559
560 # find network from list by it's id
561 new_network_list = test_config["vim_conn"].get_network_list({'id':self.__class__.network_id})
562 for list_item in new_network_list:
563 if self.__class__.network_id in list_item.get('id'):
564 self.assertEqual(self.__class__.network_id, list_item.get('id'))
565 self.assertEqual(list_item.get('type'), self.__class__.net_type)
566 self.assertEqual(list_item.get('status'), 'ACTIVE')
567
568 def test_030_get_network_list_by_shared(self):
569 Shared = False
570 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
571 self.__class__.test_index,
572 inspect.currentframe().f_code.co_name)
573 self.__class__.test_index += 1
574
575 if test_config['vimtype'] == 'openstack':
576 network_name = test_config['vim_conn'].get_network(self.__class__.network_id)['name']
577 else:
578 network_name = test_config['vim_conn'].get_network_name_by_id(self.__class__.network_id)
579 # find network from list by it's shared value
580 new_network_list = test_config["vim_conn"].get_network_list({'shared':Shared,
581 'name':network_name})
582 for list_item in new_network_list:
583 if list_item.get('shared') == Shared:
584 self.assertEqual(list_item.get('shared'), Shared)
585 self.assertEqual(list_item.get('type'), self.__class__.net_type)
586 self.assertEqual(network_name, list_item.get('name'))
587
588 def test_040_get_network_list_by_tenant_id(self):
589 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
590 self.__class__.test_index,
591 inspect.currentframe().f_code.co_name)
592 self.__class__.test_index += 1
593
594 tenant_list = test_config["vim_conn"].get_tenant_list()
595 if test_config['vimtype'] == 'openstack':
596 network_name = test_config['vim_conn'].get_network(self.__class__.network_id)['name']
597 else:
598 network_name = test_config['vim_conn'].get_network_name_by_id(self.__class__.network_id)
599
600 for tenant_item in tenant_list:
601 if test_config['tenant'] == tenant_item.get('name'):
602 # find network from list by it's tenant id
603 tenant_id = tenant_item.get('id')
604 new_network_list = test_config["vim_conn"].get_network_list({'tenant_id':tenant_id,
605 'name':network_name})
606 for list_item in new_network_list:
607 self.assertEqual(tenant_id, list_item.get('tenant_id'))
608 self.assertEqual(network_name, list_item.get('name'))
609 self.assertEqual(list_item.get('type'), self.__class__.net_type)
610 self.assertEqual(list_item.get('status'), 'ACTIVE')
611
612 def test_050_get_network_list_by_status(self):
613 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
614 self.__class__.test_index,
615 inspect.currentframe().f_code.co_name)
616 self.__class__.test_index += 1
617 status = 'ACTIVE'
618
619 if test_config['vimtype'] == 'openstack':
620 network_name = test_config['vim_conn'].get_network(self.__class__.network_id)['name']
621 else:
622 network_name = test_config['vim_conn'].get_network_name_by_id(self.__class__.network_id)
623
624 # find network from list by it's status
625 new_network_list = test_config["vim_conn"].get_network_list({'status':status,
626 'name': network_name})
627 for list_item in new_network_list:
628 self.assertIn(self.__class__.network_name, list_item.get('name'))
629 self.assertEqual(list_item.get('type'), self.__class__.net_type)
630 self.assertEqual(list_item.get('status'), status)
631
632 def test_060_get_network_list_by_negative(self):
633 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
634 self.__class__.test_index,
635 inspect.currentframe().f_code.co_name)
636 self.__class__.test_index += 1
637
638 network_list = test_config["vim_conn"].get_network_list({'name': 'unknown_name'})
639 self.assertEqual(network_list, [])
640
641 class test_vimconn_get_network(test_base):
642 network_name = None
643
644 def setUp(self):
645 # creating new network
646 self.__class__.network_name = _get_random_string(20)
647 self.__class__.net_type = 'bridge'
648 network = test_config["vim_conn"].new_network(net_name=self.__class__.network_name,
649 net_type=self.__class__.net_type)
650 self.__class__.network_id = network
651 logger.debug("{}".format(network))
652
653 def tearDown(self):
654 test_base.tearDown(self)
655
656 # Deleting created network
657 result = test_config["vim_conn"].delete_network(self.__class__.network_id)
658 if result:
659 logger.info("Network id {} sucessfully deleted".format(self.__class__.network_id))
660 else:
661 logger.info("Failed to delete network id {}".format(self.__class__.network_id))
662
663 def test_000_get_network(self):
664 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
665 self.__class__.test_index,
666 inspect.currentframe().f_code.co_name)
667 self.__class__.test_index += 1
668
669 network_info = test_config["vim_conn"].get_network(self.__class__.network_id)
670 self.assertEqual(network_info.get('status'), 'ACTIVE')
671 self.assertIn(self.__class__.network_name, network_info.get('name'))
672 self.assertEqual(network_info.get('type'), self.__class__.net_type)
673 self.assertEqual(network_info.get('id'), self.__class__.network_id)
674
675 def test_010_get_network_negative(self):
676 Non_exist_id = str(uuid.uuid4())
677 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
678 self.__class__.test_index,
679 inspect.currentframe().f_code.co_name)
680 self.__class__.test_index += 1
681 with self.assertRaises(Exception) as context:
682 test_config["vim_conn"].get_network(Non_exist_id)
683
684 self.assertEqual((context.exception).http_code, 404)
685
686 class test_vimconn_delete_network(test_base):
687 network_name = None
688
689 def test_000_delete_network(self):
690 # Creating network
691 self.__class__.network_name = _get_random_string(20)
692 self.__class__.net_type = 'bridge'
693 network = test_config["vim_conn"].new_network(net_name=self.__class__.network_name,
694 net_type=self.__class__.net_type)
695 self.__class__.network_id = network
696 logger.debug("{}".format(network))
697
698 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
699 self.__class__.test_index,
700 inspect.currentframe().f_code.co_name)
701 self.__class__.test_index += 1
702
703 result = test_config["vim_conn"].delete_network(self.__class__.network_id)
704 if result:
705 logger.info("Network id {} sucessfully deleted".format(self.__class__.network_id))
706 else:
707 logger.info("Failed to delete network id {}".format(self.__class__.network_id))
708 time.sleep(5)
709 # after deleting network we check in network list
710 network_list = test_config["vim_conn"].get_network_list({ 'id':self.__class__.network_id })
711 self.assertEqual(network_list, [])
712
713 def test_010_delete_network_negative(self):
714 Non_exist_id = str(uuid.uuid4())
715
716 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
717 self.__class__.test_index,
718 inspect.currentframe().f_code.co_name)
719 self.__class__.test_index += 1
720
721 with self.assertRaises(Exception) as context:
722 test_config["vim_conn"].delete_network(Non_exist_id)
723
724 self.assertEqual((context.exception).http_code, 404)
725
726 class test_vimconn_get_flavor(test_base):
727
728 def test_000_get_flavor(self):
729 test_directory_content = os.listdir(test_config["test_directory"])
730
731 for dir_name in test_directory_content:
732 if dir_name == 'simple_linux':
733 self.__class__.scenario_test_path = test_config["test_directory"] + '/'+ dir_name
734 vnfd_files = glob.glob(self.__class__.scenario_test_path+'/vnfd_*.yaml')
735 break
736
737 for vnfd in vnfd_files:
738 with open(vnfd, 'r') as stream:
739 vnf_descriptor = yaml.load(stream)
740
741 vnfc_list = vnf_descriptor['vnf']['VNFC']
742 for item in vnfc_list:
743 if 'ram' in item and 'vcpus' in item and 'disk' in item:
744 ram = item['ram']
745 vcpus = item['vcpus']
746 disk = item['disk']
747
748 flavor_data = {
749 'name' : _get_random_string(20),
750 'ram': ram,
751 'vcpus': vcpus,
752 'disk': disk
753 }
754
755 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
756 self.__class__.test_index,
757 inspect.currentframe().f_code.co_name)
758 self.__class__.test_index += 1
759 # create new flavor
760 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
761 # get flavor by id
762 result = test_config["vim_conn"].get_flavor(flavor_id)
763 self.assertEqual(ram, result['ram'])
764 self.assertEqual(vcpus, result['vcpus'])
765 self.assertEqual(disk, result['disk'])
766
767 # delete flavor
768 result = test_config["vim_conn"].delete_flavor(flavor_id)
769 if result:
770 logger.info("Flavor id {} sucessfully deleted".format(result))
771 else:
772 logger.info("Failed to delete flavor id {}".format(result))
773
774 def test_010_get_flavor_negative(self):
775 Non_exist_flavor_id = str(uuid.uuid4())
776
777 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
778 self.__class__.test_index,
779 inspect.currentframe().f_code.co_name)
780 self.__class__.test_index += 1
781
782 with self.assertRaises(Exception) as context:
783 test_config["vim_conn"].get_flavor(Non_exist_flavor_id)
784
785 self.assertEqual((context.exception).http_code, 404)
786
787 class test_vimconn_new_flavor(test_base):
788 flavor_id = None
789
790 def test_000_new_flavor(self):
791 flavor_data = {'name': _get_random_string(20), 'ram': 1024, 'vpcus': 1, 'disk': 10}
792
793 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
794 self.__class__.test_index,
795 inspect.currentframe().f_code.co_name)
796 self.__class__.test_index += 1
797
798 # create new flavor
799 self.__class__.flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
800 self.assertIsInstance(self.__class__.flavor_id, (str, unicode))
801 self.assertIsInstance(uuid.UUID(self.__class__.flavor_id), uuid.UUID)
802
803 def test_010_delete_flavor(self):
804 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
805 self.__class__.test_index,
806 inspect.currentframe().f_code.co_name)
807 self.__class__.test_index += 1
808
809 # delete flavor
810 result = test_config["vim_conn"].delete_flavor(self.__class__.flavor_id)
811 if result:
812 logger.info("Flavor id {} sucessfully deleted".format(result))
813 else:
814 logger.error("Failed to delete flavor id {}".format(result))
815 raise Exception ("Failed to delete created flavor")
816
817 def test_020_new_flavor_negative(self):
818 Invalid_flavor_data = {'ram': '1024', 'vcpus': 2.0, 'disk': 2.0}
819
820 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
821 self.__class__.test_index,
822 inspect.currentframe().f_code.co_name)
823 self.__class__.test_index += 1
824
825 with self.assertRaises(Exception) as context:
826 test_config["vim_conn"].new_flavor(Invalid_flavor_data)
827
828 self.assertEqual((context.exception).http_code, 400)
829
830 def test_030_delete_flavor_negative(self):
831 Non_exist_flavor_id = str(uuid.uuid4())
832
833 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
834 self.__class__.test_index,
835 inspect.currentframe().f_code.co_name)
836 self.__class__.test_index += 1
837
838 with self.assertRaises(Exception) as context:
839 test_config["vim_conn"].delete_flavor(Non_exist_flavor_id)
840
841 self.assertEqual((context.exception).http_code, 404)
842
843 class test_vimconn_new_image(test_base):
844
845 def test_000_new_image(self):
846 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
847 self.__class__.test_index,
848 inspect.currentframe().f_code.co_name)
849 self.__class__.test_index += 1
850
851 image_path = test_config['image_path']
852 if image_path:
853 self.__class__.image_id = test_config["vim_conn"].new_image({ 'name': 'TestImage', 'location' : image_path })
854 time.sleep(20)
855 self.assertEqual(type(self.__class__.image_id),str)
856 self.assertIsInstance(uuid.UUID(self.__class__.image_id),uuid.UUID)
857 else:
858 self.skipTest("Skipping test as image file not present at RO container")
859
860 def test_010_new_image_negative(self):
861 Non_exist_image_path = '/temp1/cirros.ovf'
862
863 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
864 self.__class__.test_index,
865 inspect.currentframe().f_code.co_name)
866 self.__class__.test_index += 1
867
868 with self.assertRaises(Exception) as context:
869 test_config["vim_conn"].new_image({ 'name': 'TestImage', 'location' : Non_exist_image_path })
870
871 self.assertEqual((context.exception).http_code, 400)
872
873 def test_020_delete_image(self):
874 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
875 self.__class__.test_index,
876 inspect.currentframe().f_code.co_name)
877 self.__class__.test_index += 1
878
879 image_id = test_config["vim_conn"].delete_image(self.__class__.image_id)
880 self.assertEqual(type(image_id),str)
881
882 def test_030_delete_image_negative(self):
883 Non_exist_image_id = str(uuid.uuid4())
884
885 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
886 self.__class__.test_index,
887 inspect.currentframe().f_code.co_name)
888 self.__class__.test_index += 1
889
890 with self.assertRaises(Exception) as context:
891 test_config["vim_conn"].delete_image(Non_exist_image_id)
892
893 self.assertEqual((context.exception).http_code, 404)
894
895 class test_vimconn_get_image_id_from_path(test_base):
896
897 def test_000_get_image_id_from_path(self):
898 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
899 self.__class__.test_index,
900 inspect.currentframe().f_code.co_name)
901 self.__class__.test_index += 1
902
903 image_path = test_config['image_path']
904 if image_path:
905 image_id = test_config["vim_conn"].get_image_id_from_path( image_path )
906 self.assertEqual(type(image_id),str)
907 else:
908 self.skipTest("Skipping test as image file not present at RO container")
909
910 def test_010_get_image_id_from_path_negative(self):
911 Non_exist_image_path = '/temp1/cirros.ovf'
912
913 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
914 self.__class__.test_index,
915 inspect.currentframe().f_code.co_name)
916 self.__class__.test_index += 1
917
918 with self.assertRaises(Exception) as context:
919 test_config["vim_conn"].new_image({ 'name': 'TestImage', 'location' : Non_exist_image_path })
920
921 self.assertEqual((context.exception).http_code, 400)
922
923 class test_vimconn_get_image_list(test_base):
924 image_name = None
925 image_id = None
926
927 def test_000_get_image_list(self):
928 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
929 self.__class__.test_index,
930 inspect.currentframe().f_code.co_name)
931 self.__class__.test_index += 1
932 image_list = test_config["vim_conn"].get_image_list()
933
934 for item in image_list:
935 if 'name' in item:
936 self.__class__.image_name = item['name']
937 self.__class__.image_id = item['id']
938 self.assertEqual(type(self.__class__.image_name),str)
939 self.assertEqual(type(self.__class__.image_id),str)
940
941 def test_010_get_image_list_by_name(self):
942 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
943 self.__class__.test_index,
944 inspect.currentframe().f_code.co_name)
945 self.__class__.test_index += 1
946
947 image_list = test_config["vim_conn"].get_image_list({'name': self.__class__.image_name})
948
949 for item in image_list:
950 self.assertEqual(type(item['id']), str)
951 self.assertEqual(item['id'], self.__class__.image_id)
952 self.assertEqual(type(item['name']), str)
953 self.assertEqual(item['name'], self.__class__.image_name)
954
955 def test_020_get_image_list_by_id(self):
956 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
957 self.__class__.test_index,
958 inspect.currentframe().f_code.co_name)
959 self.__class__.test_index += 1
960
961 filter_image_list = test_config["vim_conn"].get_image_list({'id': self.__class__.image_id})
962
963 for item1 in filter_image_list:
964 self.assertEqual(type(item1.get('id')), str)
965 self.assertEqual(item1.get('id'), self.__class__.image_id)
966 self.assertEqual(type(item1.get('name')), str)
967 self.assertEqual(item1.get('name'), self.__class__.image_name)
968
969 def test_030_get_image_list_negative(self):
970 Non_exist_image_id = uuid.uuid4()
971 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
972 self.__class__.test_index,
973 inspect.currentframe().f_code.co_name)
974 self.__class__.test_index += 1
975 image_list = test_config["vim_conn"].get_image_list({'name': 'Unknown_name', 'id': Non_exist_image_id})
976
977 self.assertIsNotNone(image_list, None)
978 self.assertEqual(image_list, [])
979
980 class test_vimconn_new_vminstance(test_base):
981 network_name = None
982 net_type = None
983 network_id = None
984 image_id = None
985 instance_id = None
986
987 def setUp(self):
988 # create network
989 self.__class__.network_name = _get_random_string(20)
990 self.__class__.net_type = 'bridge'
991
992 self.__class__.network_id = test_config["vim_conn"].new_network(net_name=self.__class__.network_name,
993 net_type=self.__class__.net_type)
994
995 def tearDown(self):
996 test_base.tearDown(self)
997 # Deleting created network
998 result = test_config["vim_conn"].delete_network(self.__class__.network_id)
999 if result:
1000 logger.info("Network id {} sucessfully deleted".format(self.__class__.network_id))
1001 else:
1002 logger.info("Failed to delete network id {}".format(self.__class__.network_id))
1003
1004 def test_000_new_vminstance(self):
1005 vpci = "0000:00:11.0"
1006 name = "eth0"
1007
1008 flavor_data = {'ram': 1024, 'vcpus': 1, 'disk': 10}
1009
1010 # create new flavor
1011 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
1012
1013 # find image name and image id
1014 if test_config['image_name']:
1015 image_list = test_config['vim_conn'].get_image_list({'name': test_config['image_name']})
1016 if len(image_list) == 0:
1017 raise Exception("Image {} is not found at VIM".format(test_config['image_name']))
1018 else:
1019 self.__class__.image_id = image_list[0]['id']
1020 else:
1021 image_list = test_config['vim_conn'].get_image_list()
1022 if len(image_list) == 0:
1023 raise Exception("Not found any image at VIM")
1024 else:
1025 self.__class__.image_id = image_list[0]['id']
1026
1027 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1028 self.__class__.test_index,
1029 inspect.currentframe().f_code.co_name)
1030 self.__class__.test_index += 1
1031
1032 net_list = [{'use': self.__class__.net_type, 'name': name, 'floating_ip': False, 'vpci': vpci, 'port_security': True, 'type': 'virtual', 'net_id': self.__class__.network_id}]
1033
1034 self.__class__.instance_id, _ = test_config["vim_conn"].new_vminstance(name='Test1_vm', image_id=self.__class__.image_id, flavor_id=flavor_id, net_list=net_list)
1035
1036 self.assertEqual(type(self.__class__.instance_id),str)
1037
1038 def test_010_new_vminstance_by_model(self):
1039 flavor_data = {'ram': 1024, 'vcpus': 2, 'disk': 10}
1040 model_name = 'e1000'
1041 name = 'eth0'
1042
1043 # create new flavor
1044 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
1045
1046 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1047 self.__class__.test_index,
1048 inspect.currentframe().f_code.co_name)
1049 self.__class__.test_index += 1
1050
1051 net_list = [{'use': self.__class__.net_type, 'name': name, 'floating_ip': False, 'port_security': True, 'model': model_name, 'type': 'virtual', 'net_id': self.__class__.network_id}]
1052
1053 instance_id, _ = test_config["vim_conn"].new_vminstance(name='Test1_vm', image_id=self.__class__.image_id,
1054 flavor_id=flavor_id,
1055 net_list=net_list)
1056 self.assertEqual(type(instance_id),str)
1057 # Deleting created vm instance
1058 logger.info("Deleting created vm intance")
1059 test_config["vim_conn"].delete_vminstance(instance_id)
1060 time.sleep(10)
1061
1062 def test_020_new_vminstance_by_net_use(self):
1063 flavor_data = {'ram': 1024, 'vcpus': 2, 'disk': 10}
1064 net_use = 'data'
1065 name = 'eth0'
1066
1067 # create new flavor
1068 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
1069
1070 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1071 self.__class__.test_index,
1072 inspect.currentframe().f_code.co_name)
1073 self.__class__.test_index += 1
1074
1075 net_list = [{'use': net_use, 'name': name, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self.__class__.network_id}]
1076
1077 instance_id, _ = test_config["vim_conn"].new_vminstance(name='Test1_vm', image_id=self.__class__.image_id,
1078 flavor_id=flavor_id,
1079 net_list=net_list)
1080 self.assertEqual(type(instance_id),str)
1081 # Deleting created vm instance
1082 logger.info("Deleting created vm intance")
1083 test_config["vim_conn"].delete_vminstance(instance_id)
1084 time.sleep(10)
1085
1086 def test_030_new_vminstance_by_net_type(self):
1087 flavor_data = {'ram': 1024, 'vcpus': 2, 'disk': 10}
1088 _type = 'VF'
1089 name = 'eth0'
1090
1091 # create new flavor
1092 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
1093
1094 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1095 self.__class__.test_index,
1096 inspect.currentframe().f_code.co_name)
1097 self.__class__.test_index += 1
1098
1099 net_list = [{'use': self.__class__.net_type, 'name': name, 'floating_ip': False, 'port_security': True, 'type': _type, 'net_id': self.__class__.network_id}]
1100
1101 instance_id, _ = test_config["vim_conn"].new_vminstance(name='Test1_vm', image_id=self.__class__.image_id,
1102 flavor_id=flavor_id,
1103 net_list=net_list)
1104 self.assertEqual(type(instance_id),str)
1105 # Deleting created vm instance
1106 logger.info("Deleting created vm intance")
1107 test_config["vim_conn"].delete_vminstance(instance_id)
1108 time.sleep(10)
1109
1110 def test_040_new_vminstance_by_cloud_config(self):
1111 flavor_data = {'ram': 1024, 'vcpus': 2, 'disk': 10}
1112 name = 'eth0'
1113 user_name = 'test_user'
1114
1115 key_pairs = ['ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCy2w9GHMKKNkpCmrDK2ovc3XBYDETuLWwaW24S+feHhLBQiZlzh3gSQoINlA+2ycM9zYbxl4BGzEzpTVyCQFZv5PidG4m6ox7LR+KYkDcITMyjsVuQJKDvt6oZvRt6KbChcCi0n2JJD/oUiJbBFagDBlRslbaFI2mmqmhLlJ5TLDtmYxzBLpjuX4m4tv+pdmQVfg7DYHsoy0hllhjtcDlt1nn05WgWYRTu7mfQTWfVTavu+OjIX3e0WN6NW7yIBWZcE/Q9lC0II3W7PZDE3QaT55se4SPIO2JTdqsx6XGbekdG1n6adlduOI27sOU5m4doiyJ8554yVbuDB/z5lRBD alfonso.tiernosepulveda@telefonica.com']
1116
1117 users_data = [{'key-pairs': ['ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCy2w9GHMKKNkpCmrDK2ovc3XBYDETuLWwaW24S+feHhLBQiZlzh3gSQoINlA+2ycM9zYbxl4BGzEzpTVyCQFZv5PidG4m6ox7LR+KYkDcITMyjsVuQJKDvt6oZvRt6KbChcCi0n2JJD/oUiJbBFagDBlRslbaFI2mmqmhLlJ5TLDtmYxzBLpjuX4m4tv+pdmQVfg7DYHsoy0hllhjtcDlt1nn05WgWYRTu7mfQTWfVTavu+OjIX3e0WN6NW7yIBWZcE/Q9lC0II3W7PZDE3QaT55se4SPIO2JTdqsx6XGbekdG1n6adlduOI27sOU5m4doiyJ8554yVbuDB/z5lRBD alfonso.tiernosepulveda@telefonica.com'], 'name': user_name}]
1118
1119 cloud_data = {'config-files': [{'content': 'auto enp0s3\niface enp0s3 inet dhcp\n', 'dest': '/etc/network/interfaces.d/enp0s3.cfg', 'owner': 'root:root', 'permissions': '0644'}, {'content': '#! /bin/bash\nls -al >> /var/log/osm.log\n', 'dest': '/etc/rc.local', 'permissions': '0755'}, {'content': 'file content', 'dest': '/etc/test_delete'}], 'boot-data-drive': True, 'key-pairs': key_pairs, 'users': users_data }
1120
1121 # create new flavor
1122 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
1123
1124 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1125 self.__class__.test_index,
1126 inspect.currentframe().f_code.co_name)
1127 self.__class__.test_index += 1
1128
1129 net_list = [{'use': self.__class__.net_type, 'name': name, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self.__class__.network_id}]
1130
1131 instance_id, _ = test_config["vim_conn"].new_vminstance(name='Cloud_vm', image_id=self.__class__.image_id,
1132 flavor_id=flavor_id,
1133 net_list=net_list,
1134 cloud_config=cloud_data)
1135 self.assertEqual(type(instance_id),str)
1136 # Deleting created vm instance
1137 logger.info("Deleting created vm intance")
1138 test_config["vim_conn"].delete_vminstance(instance_id)
1139 time.sleep(10)
1140
1141 def test_050_new_vminstance_by_disk_list(self):
1142 flavor_data = {'ram': 1024, 'vcpus': 2, 'disk': 10}
1143 name = 'eth0'
1144
1145 device_data = [{'image_id': self.__class__.image_id, 'size': '5'}]
1146
1147 # create new flavor
1148 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
1149
1150 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1151 self.__class__.test_index,
1152 inspect.currentframe().f_code.co_name)
1153 self.__class__.test_index += 1
1154
1155 net_list = [{'use': self.__class__.net_type, 'name': name, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self.__class__.network_id}]
1156
1157 instance_id, _ = test_config["vim_conn"].new_vminstance(name='VM_test1', image_id=self.__class__.image_id,
1158 flavor_id=flavor_id,
1159 net_list=net_list,
1160 disk_list=device_data)
1161 self.assertEqual(type(instance_id),str)
1162 # Deleting created vm instance
1163 logger.info("Deleting created vm intance")
1164 test_config["vim_conn"].delete_vminstance(instance_id)
1165 time.sleep(10)
1166
1167 def test_060_new_vminstance_negative(self):
1168 unknown_flavor_id = str(uuid.uuid4())
1169 unknown_image_id = str(uuid.uuid4())
1170 name = 'eth2'
1171
1172 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1173 self.__class__.test_index,
1174 inspect.currentframe().f_code.co_name)
1175 self.__class__.test_index += 1
1176
1177 net_list = [{'use': self.__class__.net_type, 'name': name, 'floating_ip': False, 'port_security': True, 'type': 'virtual', 'net_id': self.__class__.network_id}]
1178
1179 with self.assertRaises(Exception) as context:
1180 test_config["vim_conn"].new_vminstance(name='Test1_vm', image_id=unknown_image_id,
1181 flavor_id=unknown_flavor_id,
1182 net_list=net_list)
1183 self.assertEqual((context.exception).http_code, 404)
1184
1185 def test_070_get_vminstance(self):
1186 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1187 self.__class__.test_index,
1188 inspect.currentframe().f_code.co_name)
1189 self.__class__.test_index += 1
1190
1191 # Get instance by its id
1192 vm_info = test_config["vim_conn"].get_vminstance(self.__class__.instance_id)
1193
1194 if test_config['vimtype'] == 'vmware':
1195 for attr in vm_info:
1196 if attr == 'status':
1197 self.assertEqual(vm_info[attr], 'ACTIVE')
1198 if attr == 'hostId':
1199 self.assertEqual(type(vm_info[attr]), str)
1200 if attr == 'interfaces':
1201 self.assertEqual(type(vm_info[attr]), list)
1202 self.assertEqual(vm_info[attr][0]['IsConnected'], 'true')
1203 if attr == 'IsEnabled':
1204 self.assertEqual(vm_info[attr], 'true')
1205
1206 def test_080_get_vminstance_negative(self):
1207 unknown_instance_id = str(uuid.uuid4())
1208
1209 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1210 self.__class__.test_index,
1211 inspect.currentframe().f_code.co_name)
1212 self.__class__.test_index += 1
1213
1214 with self.assertRaises(Exception) as context:
1215 test_config["vim_conn"].get_vminstance(unknown_instance_id)
1216
1217 self.assertEqual((context.exception).http_code, 404)
1218
1219 def test_090_refresh_vms_status(self):
1220 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1221 self.__class__.test_index,
1222 inspect.currentframe().f_code.co_name)
1223 self.__class__.test_index += 1
1224 vm_list = []
1225 vm_list.append(self.__class__.instance_id)
1226
1227 # refresh vm status
1228 vm_info = test_config["vim_conn"].refresh_vms_status(vm_list)
1229 for attr in vm_info[self.__class__.instance_id]:
1230 if attr == 'status':
1231 self.assertEqual(vm_info[self.__class__.instance_id][attr], 'ACTIVE')
1232 if attr == 'interfaces':
1233 self.assertEqual(type(vm_info[self.__class__.instance_id][attr]), list)
1234
1235 def test_100_refresh_vms_status_negative(self):
1236 unknown_id = str(uuid.uuid4())
1237
1238 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1239 self.__class__.test_index,
1240 inspect.currentframe().f_code.co_name)
1241 self.__class__.test_index += 1
1242
1243 vm_dict = test_config["vim_conn"].refresh_vms_status([unknown_id])
1244 self.assertEqual(vm_dict, {})
1245
1246 def test_110_action_vminstance(self):
1247 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1248 self.__class__.test_index,
1249 inspect.currentframe().f_code.co_name)
1250 self.__class__.test_index += 1
1251
1252 action_list = ['shutdown','start','shutoff','rebuild','pause','resume']
1253 # various action on vminstace
1254 for action in action_list:
1255 instance_id = test_config["vim_conn"].action_vminstance(self.__class__.instance_id,
1256 { action: None})
1257 self.assertEqual(instance_id, self.__class__.instance_id)
1258
1259 def test_120_action_vminstance_negative(self):
1260 non_exist_id = str(uuid.uuid4())
1261 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1262 self.__class__.test_index,
1263 inspect.currentframe().f_code.co_name)
1264 self.__class__.test_index += 1
1265
1266 action = 'start'
1267 with self.assertRaises(Exception) as context:
1268 test_config["vim_conn"].action_vminstance(non_exist_id, { action: None})
1269
1270 self.assertEqual((context.exception).http_code, 400)
1271
1272 def test_130_delete_vminstance(self):
1273 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1274 self.__class__.test_index,
1275 inspect.currentframe().f_code.co_name)
1276 self.__class__.test_index += 1
1277
1278 # Deleting created vm instance
1279 logger.info("Deleting created vm instance")
1280 test_config["vim_conn"].delete_vminstance(self.__class__.instance_id)
1281 time.sleep(10)
1282
1283 class test_vimconn_get_tenant_list(test_base):
1284 tenant_id = None
1285
1286 def test_000_get_tenant_list(self):
1287 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1288 self.__class__.test_index,
1289 inspect.currentframe().f_code.co_name)
1290 self.__class__.test_index += 1
1291
1292 # Getting tenant list
1293 tenant_list = test_config["vim_conn"].get_tenant_list()
1294
1295 for item in tenant_list:
1296 if test_config['tenant'] == item['name']:
1297 self.__class__.tenant_id = item['id']
1298 self.assertEqual(type(item['name']), str)
1299 self.assertEqual(type(item['id']), str)
1300
1301 def test_010_get_tenant_list_by_id(self):
1302 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1303 self.__class__.test_index,
1304 inspect.currentframe().f_code.co_name)
1305 self.__class__.test_index += 1
1306
1307 # Getting filter tenant list by its id
1308 filter_tenant_list = test_config["vim_conn"].get_tenant_list({'id': self.__class__.tenant_id})
1309
1310 for item in filter_tenant_list:
1311 self.assertEqual(type(item['id']), str)
1312 self.assertEqual(item['id'], self.__class__.tenant_id)
1313
1314 def test_020_get_tenant_list_by_name(self):
1315 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1316 self.__class__.test_index,
1317 inspect.currentframe().f_code.co_name)
1318 self.__class__.test_index += 1
1319
1320 # Getting filter tenant list by its name
1321 filter_tenant_list = test_config["vim_conn"].get_tenant_list({'name': test_config['tenant']})
1322
1323 for item in filter_tenant_list:
1324 self.assertEqual(type(item['name']), str)
1325 self.assertEqual(item['name'], test_config['tenant'])
1326
1327 def test_030_get_tenant_list_by_name_and_id(self):
1328 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1329 self.__class__.test_index,
1330 inspect.currentframe().f_code.co_name)
1331 self.__class__.test_index += 1
1332
1333 # Getting filter tenant list by its name and id
1334 filter_tenant_list = test_config["vim_conn"].get_tenant_list({'name': test_config['tenant'],
1335 'id': self.__class__.tenant_id})
1336
1337 for item in filter_tenant_list:
1338 self.assertEqual(type(item['name']), str)
1339 self.assertEqual(type(item['id']), str)
1340 self.assertEqual(item['name'], test_config['tenant'])
1341 self.assertEqual(item['id'], self.__class__.tenant_id)
1342
1343 def test_040_get_tenant_list_negative(self):
1344 non_exist_tenant_name = "Tenant_123"
1345 non_exist_tenant_id = "kjhgrt456-45345kjhdfgnbdk-34dsfjdfg"
1346 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1347 self.__class__.test_index,
1348 inspect.currentframe().f_code.co_name)
1349 self.__class__.test_index += 1
1350
1351 filter_tenant_list = test_config["vim_conn"].get_tenant_list({'name': non_exist_tenant_name,
1352 'id': non_exist_tenant_id})
1353
1354 self.assertEqual(filter_tenant_list, [])
1355
1356 class test_vimconn_new_tenant(test_base):
1357 tenant_id = None
1358
1359 def test_000_new_tenant(self):
1360 tenant_name = _get_random_string(20)
1361 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1362 self.__class__.test_index,
1363 inspect.currentframe().f_code.co_name)
1364 self.__class__.test_index += 1
1365
1366 self.__class__.tenant_id = test_config["vim_conn"].new_tenant(tenant_name)
1367 time.sleep(15)
1368
1369 self.assertEqual(type(self.__class__.tenant_id), str)
1370
1371 def test_010_new_tenant_negative(self):
1372 Invalid_tenant_name = 10121
1373 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1374 self.__class__.test_index,
1375 inspect.currentframe().f_code.co_name)
1376 self.__class__.test_index += 1
1377
1378 with self.assertRaises(Exception) as context:
1379 test_config["vim_conn"].new_tenant(Invalid_tenant_name)
1380
1381 self.assertEqual((context.exception).http_code, 400)
1382
1383 def test_020_delete_tenant(self):
1384 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1385 self.__class__.test_index,
1386 inspect.currentframe().f_code.co_name)
1387 self.__class__.test_index += 1
1388
1389 tenant_id = test_config["vim_conn"].delete_tenant(self.__class__.tenant_id)
1390 self.assertEqual(type(tenant_id), str)
1391
1392 def test_030_delete_tenant_negative(self):
1393 Non_exist_tenant_name = 'Test_30_tenant'
1394 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1395 self.__class__.test_index,
1396 inspect.currentframe().f_code.co_name)
1397 self.__class__.test_index += 1
1398
1399 with self.assertRaises(Exception) as context:
1400 test_config["vim_conn"].delete_tenant(Non_exist_tenant_name)
1401
1402 self.assertEqual((context.exception).http_code, 404)
1403
1404
1405 '''
1406 IMPORTANT NOTE
1407 The following unittest class does not have the 'test_' on purpose. This test is the one used for the
1408 scenario based tests.
1409 '''
1410 class descriptor_based_scenario_test(test_base):
1411 test_index = 0
1412 scenario_test_path = None
1413
1414 @classmethod
1415 def setUpClass(cls):
1416 cls.test_index = 1
1417 cls.to_delete_list = []
1418 cls.scenario_uuids = []
1419 cls.instance_scenario_uuids = []
1420 cls.scenario_test_path = test_config["test_directory"] + '/' + test_config["test_folder"]
1421 logger.info("{}. {} {}".format(test_config["test_number"], cls.__name__, test_config["test_folder"]))
1422
1423 @classmethod
1424 def tearDownClass(cls):
1425 test_config["test_number"] += 1
1426
1427 def test_000_load_scenario(self):
1428 self.__class__.test_text = "{}.{}. TEST {} {}".format(test_config["test_number"], self.__class__.test_index,
1429 inspect.currentframe().f_code.co_name,
1430 test_config["test_folder"])
1431 self.__class__.test_index += 1
1432 # load VNFD and NSD
1433 descriptor_files = glob.glob(self.__class__.scenario_test_path+'/*.yaml')
1434 vnf_descriptors = []
1435 scenario_descriptors = []
1436 for descriptor_file in descriptor_files:
1437 with open(descriptor_file, 'r') as stream:
1438 descriptor = yaml.load(stream)
1439 if "vnf" in descriptor or "vnfd:vnfd-catalog" in descriptor or "vnfd-catalog" in descriptor:
1440 vnf_descriptors.append(descriptor)
1441 else:
1442 scenario_descriptors.append(descriptor)
1443
1444 scenario_file = glob.glob(self.__class__.scenario_test_path + '/scenario_*.yaml')
1445 if not vnf_descriptors or not scenario_descriptors or len(scenario_descriptors) > 1:
1446 raise Exception("Test '{}' not valid. It must contain an scenario file and at least one vnfd file'".format(
1447 test_config["test_folder"]))
1448
1449 # load all vnfd
1450 for vnf_descriptor in vnf_descriptors:
1451 logger.debug("VNF descriptor: {}".format(vnf_descriptor))
1452 vnf = test_config["client"].create_vnf(descriptor=vnf_descriptor, image_name=test_config["image_name"])
1453 logger.debug(vnf)
1454 if 'vnf' in vnf:
1455 vnf_uuid = vnf['vnf']['uuid']
1456 else:
1457 vnf_uuid = vnf['vnfd'][0]['uuid']
1458 self.__class__.to_delete_list.insert(0, {"item": "vnf", "function": test_config["client"].delete_vnf,
1459 "params": {"uuid": vnf_uuid}})
1460
1461 # load the scenario definition
1462 for scenario_descriptor in scenario_descriptors:
1463 # networks = scenario_descriptor['scenario']['networks']
1464 # networks[test_config["mgmt_net"]] = networks.pop('mgmt')
1465 logger.debug("Scenario descriptor: {}".format(scenario_descriptor))
1466 scenario = test_config["client"].create_scenario(descriptor=scenario_descriptor)
1467 logger.debug(scenario)
1468 if 'scenario' in scenario:
1469 scenario_uuid = scenario['scenario']['uuid']
1470 else:
1471 scenario_uuid = scenario['nsd'][0]['uuid']
1472 self.__class__.to_delete_list.insert(0, {"item": "scenario",
1473 "function": test_config["client"].delete_scenario,
1474 "params": {"uuid": scenario_uuid}})
1475 self.__class__.scenario_uuids.append(scenario_uuid)
1476
1477 def test_010_instantiate_scenario(self):
1478 self.__class__.test_text = "{}.{}. TEST {} {}".format(test_config["test_number"], self.__class__.test_index,
1479 inspect.currentframe().f_code.co_name,
1480 test_config["test_folder"])
1481 self.__class__.test_index += 1
1482 for scenario_uuid in self.__class__.scenario_uuids:
1483 instance_descriptor = {
1484 "instance":{
1485 "name": self.__class__.test_text,
1486 "scenario": scenario_uuid,
1487 "networks":{
1488 "mgmt": {"sites": [ { "netmap-use": test_config["mgmt_net"]} ]}
1489 }
1490 }
1491 }
1492 instance = test_config["client"].create_instance(instance_descriptor)
1493 self.__class__.instance_scenario_uuids.append(instance['uuid'])
1494 logger.debug(instance)
1495 self.__class__.to_delete_list.insert(0, {"item": "instance",
1496 "function": test_config["client"].delete_instance,
1497 "params": {"uuid": instance['uuid']}})
1498
1499 def test_020_check_deployent(self):
1500 self.__class__.test_text = "{}.{}. TEST {} {}".format(test_config["test_number"], self.__class__.test_index,
1501 inspect.currentframe().f_code.co_name,
1502 test_config["test_folder"])
1503 self.__class__.test_index += 1
1504
1505 if test_config["manual"]:
1506 raw_input('Scenario has been deployed. Perform manual check and press any key to resume')
1507 return
1508
1509 keep_waiting = test_config["timeout"]
1510 pending_instance_scenario_uuids = list(self.__class__.instance_scenario_uuids) # make a copy
1511 while pending_instance_scenario_uuids:
1512 index = 0
1513 while index < len(pending_instance_scenario_uuids):
1514 result = check_instance_scenario_active(pending_instance_scenario_uuids[index])
1515 if result[0]:
1516 del pending_instance_scenario_uuids[index]
1517 break
1518 elif 'ERROR' in result[1]:
1519 msg = 'Got error while waiting for the instance to get active: '+result[1]
1520 logging.error(msg)
1521 raise Exception(msg)
1522 index += 1
1523
1524 if keep_waiting >= 5:
1525 time.sleep(5)
1526 keep_waiting -= 5
1527 elif keep_waiting > 0:
1528 time.sleep(keep_waiting)
1529 keep_waiting = 0
1530 else:
1531 msg = 'Timeout reached while waiting instance scenario to get active'
1532 logging.error(msg)
1533 raise Exception(msg)
1534
1535 def test_030_clean_deployment(self):
1536 self.__class__.test_text = "{}.{}. TEST {} {}".format(test_config["test_number"], self.__class__.test_index,
1537 inspect.currentframe().f_code.co_name,
1538 test_config["test_folder"])
1539 self.__class__.test_index += 1
1540 #At the moment if you delete an scenario right after creating it, in openstack datacenters
1541 #sometimes scenario ports get orphaned. This sleep is just a dirty workaround
1542 time.sleep(5)
1543 for item in self.__class__.to_delete_list:
1544 response = item["function"](**item["params"])
1545 logger.debug(response)
1546
1547
1548 def _get_random_string(maxLength):
1549 '''generates a string with random characters string.letters and string.digits
1550 with a random length up to maxLength characters. If maxLength is <15 it will be changed automatically to 15
1551 '''
1552 prefix = 'testing_'
1553 min_string = 15
1554 minLength = min_string - len(prefix)
1555 if maxLength < min_string: maxLength = min_string
1556 maxLength -= len(prefix)
1557 length = random.randint(minLength,maxLength)
1558 return 'testing_'+"".join([random.choice(string.letters+string.digits) for i in xrange(length)])
1559
1560
1561 def test_vimconnector(args):
1562 global test_config
1563 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + "/osm_ro")
1564 test_config['vimtype'] = args.vimtype
1565 if args.vimtype == "vmware":
1566 import vimconn_vmware as vim
1567
1568 test_config["test_directory"] = os.path.dirname(__file__) + "/RO_tests"
1569
1570 tenant_name = args.tenant_name
1571 test_config['tenant'] = tenant_name
1572 config_params = yaml.load(args.config_param)
1573 org_name = config_params.get('orgname')
1574 org_user = config_params.get('user')
1575 org_passwd = config_params.get('passwd')
1576 vim_url = args.endpoint_url
1577 test_config['image_path'] = args.image_path
1578 test_config['image_name'] = args.image_name
1579
1580 # vmware connector obj
1581 test_config['vim_conn'] = vim.vimconnector(name=org_name, tenant_name=tenant_name, user=org_user,passwd=org_passwd, url=vim_url, config=config_params)
1582
1583 elif args.vimtype == "aws":
1584 import vimconn_aws as vim
1585 elif args.vimtype == "openstack":
1586 import vimconn_openstack as vim
1587
1588 test_config["test_directory"] = os.path.dirname(__file__) + "/RO_tests"
1589
1590 tenant_name = args.tenant_name
1591 test_config['tenant'] = tenant_name
1592 config_params = yaml.load(args.config_param)
1593 os_user = config_params.get('user')
1594 os_passwd = config_params.get('passwd')
1595 vim_url = args.endpoint_url
1596 test_config['image_path'] = args.image_path
1597 test_config['image_name'] = args.image_name
1598
1599 # openstack connector obj
1600 vim_persistent_info = {}
1601 test_config['vim_conn'] = vim.vimconnector(
1602 uuid="test-uuid-1", name="VIO-openstack",
1603 tenant_id=None, tenant_name=tenant_name,
1604 url=vim_url, url_admin=None,
1605 user=os_user, passwd=os_passwd,
1606 config=config_params, persistent_info=vim_persistent_info
1607 )
1608 test_config['vim_conn'].debug = "true"
1609
1610 elif args.vimtype == "openvim":
1611 import vimconn_openvim as vim
1612 else:
1613 logger.critical("vimtype '{}' not supported".format(args.vimtype))
1614 sys.exit(1)
1615 executed = 0
1616 failed = 0
1617 clsmembers = inspect.getmembers(sys.modules[__name__], inspect.isclass)
1618 # If only want to obtain a tests list print it and exit
1619 if args.list_tests:
1620 tests_names = []
1621 for cls in clsmembers:
1622 if cls[0].startswith('test_vimconn'):
1623 tests_names.append(cls[0])
1624
1625 msg = "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names))
1626 print(msg)
1627 logger.info(msg)
1628 sys.exit(0)
1629
1630 # Create the list of tests to be run
1631 code_based_tests = []
1632 if args.tests:
1633 for test in args.tests:
1634 for t in test.split(','):
1635 matches_code_based_tests = [item for item in clsmembers if item[0] == t]
1636 if len(matches_code_based_tests) > 0:
1637 code_based_tests.append(matches_code_based_tests[0][1])
1638 else:
1639 logger.critical("Test '{}' is not among the possible ones".format(t))
1640 sys.exit(1)
1641 if not code_based_tests:
1642 # include all tests
1643 for cls in clsmembers:
1644 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
1645 if cls[0].startswith('test_vimconn'):
1646 code_based_tests.append(cls[1])
1647
1648 logger.debug("tests to be executed: {}".format(code_based_tests))
1649
1650 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1651 # This is handled in the tests using logging.
1652 stream = open('/dev/null', 'w')
1653
1654 # Run code based tests
1655 basic_tests_suite = unittest.TestSuite()
1656 for test in code_based_tests:
1657 basic_tests_suite.addTest(unittest.makeSuite(test))
1658 result = unittest.TextTestRunner(stream=stream, failfast=failfast).run(basic_tests_suite)
1659 executed += result.testsRun
1660 failed += len(result.failures) + len(result.errors)
1661 if failfast and failed:
1662 sys.exit(1)
1663 if len(result.failures) > 0:
1664 logger.debug("failures : {}".format(result.failures))
1665 if len(result.errors) > 0:
1666 logger.debug("errors : {}".format(result.errors))
1667 return executed, failed
1668
1669
1670 def test_vim(args):
1671 global test_config
1672 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + "/osm_ro")
1673 import openmanoclient
1674 executed = 0
1675 failed = 0
1676 test_config["client"] = openmanoclient.openmanoclient(
1677 endpoint_url=args.endpoint_url,
1678 tenant_name=args.tenant_name,
1679 datacenter_name=args.datacenter,
1680 debug=args.debug, logger=test_config["logger_name"])
1681 clsmembers = inspect.getmembers(sys.modules[__name__], inspect.isclass)
1682 # If only want to obtain a tests list print it and exit
1683 if args.list_tests:
1684 tests_names = []
1685 for cls in clsmembers:
1686 if cls[0].startswith('test_VIM'):
1687 tests_names.append(cls[0])
1688
1689 msg = "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names)) +\
1690 "\nNOTE: The test test_VIM_tenant_operations will fail in case the used datacenter is type OpenStack " \
1691 "unless RO has access to the admin endpoint. Therefore this test is excluded by default"
1692 print(msg)
1693 logger.info(msg)
1694 sys.exit(0)
1695
1696 # Create the list of tests to be run
1697 code_based_tests = []
1698 if args.tests:
1699 for test in args.tests:
1700 for t in test.split(','):
1701 matches_code_based_tests = [item for item in clsmembers if item[0] == t]
1702 if len(matches_code_based_tests) > 0:
1703 code_based_tests.append(matches_code_based_tests[0][1])
1704 else:
1705 logger.critical("Test '{}' is not among the possible ones".format(t))
1706 sys.exit(1)
1707 if not code_based_tests:
1708 # include all tests
1709 for cls in clsmembers:
1710 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
1711 if cls[0].startswith('test_VIM') and cls[0] != 'test_VIM_tenant_operations':
1712 code_based_tests.append(cls[1])
1713
1714 logger.debug("tests to be executed: {}".format(code_based_tests))
1715
1716 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1717 # This is handled in the tests using logging.
1718 stream = open('/dev/null', 'w')
1719
1720 # Run code based tests
1721 basic_tests_suite = unittest.TestSuite()
1722 for test in code_based_tests:
1723 basic_tests_suite.addTest(unittest.makeSuite(test))
1724 result = unittest.TextTestRunner(stream=stream, failfast=failfast).run(basic_tests_suite)
1725 executed += result.testsRun
1726 failed += len(result.failures) + len(result.errors)
1727 if failfast and failed:
1728 sys.exit(1)
1729 if len(result.failures) > 0:
1730 logger.debug("failures : {}".format(result.failures))
1731 if len(result.errors) > 0:
1732 logger.debug("errors : {}".format(result.errors))
1733 return executed, failed
1734
1735
1736 def test_deploy(args):
1737 global test_config
1738 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + "/osm_ro")
1739 import openmanoclient
1740 executed = 0
1741 failed = 0
1742 test_config["test_directory"] = os.path.dirname(__file__) + "/RO_tests"
1743 test_config["image_name"] = args.image_name
1744 test_config["mgmt_net"] = args.mgmt_net
1745 test_config["manual"] = args.manual
1746 test_directory_content = os.listdir(test_config["test_directory"])
1747 # If only want to obtain a tests list print it and exit
1748 if args.list_tests:
1749 msg = "the 'deploy' set tests are:\n\t" + ', '.join(sorted(test_directory_content))
1750 print(msg)
1751 # logger.info(msg)
1752 sys.exit(0)
1753
1754 descriptor_based_tests = []
1755 # Create the list of tests to be run
1756 code_based_tests = []
1757 if args.tests:
1758 for test in args.tests:
1759 for t in test.split(','):
1760 if t in test_directory_content:
1761 descriptor_based_tests.append(t)
1762 else:
1763 logger.critical("Test '{}' is not among the possible ones".format(t))
1764 sys.exit(1)
1765 if not descriptor_based_tests:
1766 # include all tests
1767 descriptor_based_tests = test_directory_content
1768
1769 logger.debug("tests to be executed: {}".format(code_based_tests))
1770
1771 # import openmanoclient from relative path
1772 test_config["client"] = openmanoclient.openmanoclient(
1773 endpoint_url=args.endpoint_url,
1774 tenant_name=args.tenant_name,
1775 datacenter_name=args.datacenter,
1776 debug=args.debug, logger=test_config["logger_name"])
1777
1778 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1779 # This is handled in the tests using logging.
1780 stream = open('/dev/null', 'w')
1781 # This scenario based tests are defined as directories inside the directory defined in 'test_directory'
1782 for test in descriptor_based_tests:
1783 test_config["test_folder"] = test
1784 test_suite = unittest.TestSuite()
1785 test_suite.addTest(unittest.makeSuite(descriptor_based_scenario_test))
1786 result = unittest.TextTestRunner(stream=stream, failfast=False).run(test_suite)
1787 executed += result.testsRun
1788 failed += len(result.failures) + len(result.errors)
1789 if failfast and failed:
1790 sys.exit(1)
1791 if len(result.failures) > 0:
1792 logger.debug("failures : {}".format(result.failures))
1793 if len(result.errors) > 0:
1794 logger.debug("errors : {}".format(result.errors))
1795
1796 return executed, failed
1797
1798 if __name__=="__main__":
1799
1800 parser = ArgumentParser(description='Test RO module')
1801 parser.add_argument('-v','--version', action='version', help="Show current version",
1802 version='%(prog)s version ' + __version__ + ' ' + version_date)
1803
1804 # Common parameters
1805 parent_parser = ArgumentParser(add_help=False)
1806 parent_parser.add_argument('--failfast', help='Stop when a test fails rather than execute all tests',
1807 dest='failfast', action="store_true", default=False)
1808 parent_parser.add_argument('--failed', help='Set logs to show only failed tests. --debug disables this option',
1809 dest='failed', action="store_true", default=False)
1810 default_logger_file = os.path.dirname(__file__)+'/'+os.path.splitext(os.path.basename(__file__))[0]+'.log'
1811 parent_parser.add_argument('--list-tests', help='List all available tests', dest='list_tests', action="store_true",
1812 default=False)
1813 parent_parser.add_argument('--logger_file', dest='logger_file', default=default_logger_file,
1814 help='Set the logger file. By default '+default_logger_file)
1815 parent_parser.add_argument("-t", '--tenant', dest='tenant_name', default="osm",
1816 help="Set the openmano tenant to use for the test. By default 'osm'")
1817 parent_parser.add_argument('--debug', help='Set logs to debug level', dest='debug', action="store_true")
1818 parent_parser.add_argument('--timeout', help='Specify the instantiation timeout in seconds. By default 300',
1819 dest='timeout', type=int, default=300)
1820 parent_parser.add_argument('--test', '--tests', help='Specify the tests to run', dest='tests', action="append")
1821
1822 subparsers = parser.add_subparsers(help='test sets')
1823
1824 # Deployment test set
1825 # -------------------
1826 deploy_parser = subparsers.add_parser('deploy', parents=[parent_parser],
1827 help="test deployment using descriptors at RO_test folder ")
1828 deploy_parser.set_defaults(func=test_deploy)
1829
1830 # Mandatory arguments
1831 mandatory_arguments = deploy_parser.add_argument_group('mandatory arguments')
1832 mandatory_arguments.add_argument('-d', '--datacenter', required=True, help='Set the datacenter to test')
1833 mandatory_arguments.add_argument("-i", '--image-name', required=True, dest="image_name",
1834 help='Image name available at datacenter used for the tests')
1835 mandatory_arguments.add_argument("-n", '--mgmt-net-name', required=True, dest='mgmt_net',
1836 help='Set the vim management network to use for tests')
1837
1838 # Optional arguments
1839 deploy_parser.add_argument('-m', '--manual-check', dest='manual', action="store_true", default=False,
1840 help='Pause execution once deployed to allow manual checking of the '
1841 'deployed instance scenario')
1842 deploy_parser.add_argument('-u', '--url', dest='endpoint_url', default='http://localhost:9090/openmano',
1843 help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
1844
1845 # Vimconn test set
1846 # -------------------
1847 vimconn_parser = subparsers.add_parser('vimconn', parents=[parent_parser], help="test vimconnector plugin")
1848 vimconn_parser.set_defaults(func=test_vimconnector)
1849 # Mandatory arguments
1850 mandatory_arguments = vimconn_parser.add_argument_group('mandatory arguments')
1851 mandatory_arguments.add_argument('--vimtype', choices=['vmware', 'aws', 'openstack', 'openvim'], required=True,
1852 help='Set the vimconnector type to test')
1853 mandatory_arguments.add_argument('-c', '--config', dest='config_param', required=True,
1854 help='Set the vimconnector specific config parameters in dictionary format')
1855 mandatory_arguments.add_argument('-u', '--url', dest='endpoint_url',required=True, help="Set the vim connector url or Host IP")
1856 # Optional arguments
1857 vimconn_parser.add_argument('-i', '--image-path', dest='image_path', help="Provide image path present at RO container")
1858 vimconn_parser.add_argument('-n', '--image-name', dest='image_name', help="Provide image name for test")
1859 # TODO add optional arguments for vimconn tests
1860 # vimconn_parser.add_argument("-i", '--image-name', dest='image_name', help='<HELP>'))
1861
1862 # Datacenter test set
1863 # -------------------
1864 vimconn_parser = subparsers.add_parser('vim', parents=[parent_parser], help="test vim")
1865 vimconn_parser.set_defaults(func=test_vim)
1866
1867 # Mandatory arguments
1868 mandatory_arguments = vimconn_parser.add_argument_group('mandatory arguments')
1869 mandatory_arguments.add_argument('-d', '--datacenter', required=True, help='Set the datacenter to test')
1870
1871 # Optional arguments
1872 vimconn_parser.add_argument('-u', '--url', dest='endpoint_url', default='http://localhost:9090/openmano',
1873 help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
1874
1875 argcomplete.autocomplete(parser)
1876 args = parser.parse_args()
1877 # print str(args)
1878 test_config = {}
1879
1880 # default logger level is INFO. Options --debug and --failed override this, being --debug prioritary
1881 logger_level = 'INFO'
1882 if args.debug:
1883 logger_level = 'DEBUG'
1884 elif args.failed:
1885 logger_level = 'WARNING'
1886 logger_name = os.path.basename(__file__)
1887 test_config["logger_name"] = logger_name
1888 logger = logging.getLogger(logger_name)
1889 logger.setLevel(logger_level)
1890 failfast = args.failfast
1891
1892 # Configure a logging handler to store in a logging file
1893 if args.logger_file:
1894 fileHandler = logging.FileHandler(args.logger_file)
1895 formatter_fileHandler = logging.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s')
1896 fileHandler.setFormatter(formatter_fileHandler)
1897 logger.addHandler(fileHandler)
1898
1899 # Configure a handler to print to stdout
1900 consoleHandler = logging.StreamHandler(sys.stdout)
1901 formatter_consoleHandler = logging.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s')
1902 consoleHandler.setFormatter(formatter_consoleHandler)
1903 logger.addHandler(consoleHandler)
1904
1905 logger.debug('Program started with the following arguments: ' + str(args))
1906
1907 # set test config parameters
1908 test_config["timeout"] = args.timeout
1909 test_config["test_number"] = 1
1910
1911 executed, failed = args.func(args)
1912
1913 # Log summary
1914 logger.warning("Total number of tests: {}; Total number of failures/errors: {}".format(executed, failed))
1915 sys.exit(1 if failed else 0)