projects / osm / RO.git / blob
? search:


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