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


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