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


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