Revert "Removing deprecated/unused/outdated code"
[osm/RO.git] / RO / 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 # DEBUG WITH PDB
24 from os import getenv
25 if getenv('OSMRO_PDB_DEBUG'):
26 import sys
27 print(sys.path)
28 import pdb
29 pdb.set_trace()
30
31
32 """
33 Module for testing openmano functionality. It uses openmanoclient.py for invoking openmano
34 """
35
36 import logging
37 import os
38 import argcomplete
39 import unittest
40 import string
41 import inspect
42 import random
43 # import traceback
44 import glob
45 import yaml
46 import sys
47 import time
48 import uuid
49 from argparse import ArgumentParser
50
51 __author__ = "Pablo Montes, Alfonso Tierno"
52 __date__ = "$16-Feb-2017 17:08:16$"
53 __version__ = "0.1.0"
54 version_date = "Oct 2017"
55
56 test_config = {} # used for global variables with the test configuration
57
58
59 class test_base(unittest.TestCase):
60 test_index = 1
61 test_text = None
62
63 @classmethod
64 def setUpClass(cls):
65 logger.info("{}. {}".format(test_config["test_number"], cls.__name__))
66
67 @classmethod
68 def tearDownClass(cls):
69 test_config["test_number"] += 1
70
71 def tearDown(self):
72 exec_info = sys.exc_info()
73 if exec_info == (None, None, None):
74 logger.info(self.__class__.test_text+" -> TEST OK")
75 else:
76 logger.warning(self.__class__.test_text+" -> TEST NOK")
77 logger.critical("Traceback error",exc_info=True)
78
79
80 def check_instance_scenario_active(uuid):
81 instance = test_config["client"].get_instance(uuid=uuid)
82
83 for net in instance['nets']:
84 status = net['status']
85 if status != 'ACTIVE':
86 return (False, status)
87
88 for vnf in instance['vnfs']:
89 for vm in vnf['vms']:
90 status = vm['status']
91 if status != 'ACTIVE':
92 return (False, status)
93
94 return (True, None)
95
96
97 '''
98 IMPORTANT NOTE
99 All unittest classes for code based tests must have prefix 'test_' in order to be taken into account for tests
100 '''
101 class test_VIM_datacenter_tenant_operations(test_base):
102 tenant_name = None
103
104 def test_000_create_RO_tenant(self):
105 self.__class__.tenant_name = _get_random_string(20)
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 logger.debug("Test create tenant")
110 tenant = test_config["client"].create_tenant(name=self.__class__.tenant_name,
111 description=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_010_list_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"].get_tenant(name=self.__class__.tenant_name)
120 logger.debug("{}".format(tenant))
121 self.assertEqual(tenant.get('tenant', {}).get('name', ''), self.__class__.tenant_name)
122
123 def test_020_delete_RO_tenant(self):
124 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
125 inspect.currentframe().f_code.co_name)
126 self.__class__.test_index += 1
127 tenant = test_config["client"].delete_tenant(name=self.__class__.tenant_name)
128 logger.debug("{}".format(tenant))
129 assert('deleted' in tenant.get('result',""))
130
131
132 class test_VIM_datacenter_operations(test_base):
133 datacenter_name = None
134
135 def test_000_create_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 self.__class__.datacenter_name = _get_random_string(20)
139 self.__class__.test_index += 1
140 self.datacenter = test_config["client"].create_datacenter(name=self.__class__.datacenter_name,
141 vim_url="http://fakeurl/fake")
142 logger.debug("{}".format(self.datacenter))
143 self.assertEqual (self.datacenter.get('datacenter', {}).get('name',''), self.__class__.datacenter_name)
144
145 def test_010_list_datacenter(self):
146 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
147 inspect.currentframe().f_code.co_name)
148
149 self.__class__.test_index += 1
150 self.datacenter = test_config["client"].get_datacenter(all_tenants=True, name=self.__class__.datacenter_name)
151 logger.debug("{}".format(self.datacenter))
152 self.assertEqual (self.datacenter.get('datacenter', {}).get('name', ''), self.__class__.datacenter_name)
153
154 def test_020_attach_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"].attach_datacenter(name=self.__class__.datacenter_name,
160 vim_tenant_name='fake')
161 logger.debug("{}".format(self.datacenter))
162 assert ('uuid' in self.datacenter.get('datacenter', {}))
163
164 def test_030_list_attached_datacenter(self):
165 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
166 inspect.currentframe().f_code.co_name)
167
168 self.__class__.test_index += 1
169 self.datacenter = test_config["client"].get_datacenter(all_tenants=False, name=self.__class__.datacenter_name)
170 logger.debug("{}".format(self.datacenter))
171 self.assertEqual (self.datacenter.get('datacenter', {}).get('name', ''), self.__class__.datacenter_name)
172
173 def test_040_detach_datacenter(self):
174 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
175 inspect.currentframe().f_code.co_name)
176
177 self.__class__.test_index += 1
178 self.datacenter = test_config["client"].detach_datacenter(name=self.__class__.datacenter_name)
179 logger.debug("{}".format(self.datacenter))
180 assert ('detached' in self.datacenter.get('result', ""))
181
182 def test_050_delete_datacenter(self):
183 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
184 inspect.currentframe().f_code.co_name)
185
186 self.__class__.test_index += 1
187 self.datacenter = test_config["client"].delete_datacenter(name=self.__class__.datacenter_name)
188 logger.debug("{}".format(self.datacenter))
189 assert('deleted' in self.datacenter.get('result',""))
190
191
192 class test_VIM_network_operations(test_base):
193 vim_network_name = None
194 vim_network_uuid = None
195
196 def test_000_create_VIM_network(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__.vim_network_name = _get_random_string(20)
200 self.__class__.test_index += 1
201 network = test_config["client"].vim_action("create", "networks", name=self.__class__.vim_network_name)
202 logger.debug("{}".format(network))
203 self.__class__.vim_network_uuid = network["network"]["id"]
204 self.assertEqual(network.get('network', {}).get('name', ''), self.__class__.vim_network_name)
205
206 def test_010_list_VIM_networks(self):
207 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
208 inspect.currentframe().f_code.co_name)
209 self.__class__.test_index += 1
210 networks = test_config["client"].vim_action("list", "networks")
211 logger.debug("{}".format(networks))
212
213 def test_020_get_VIM_network_by_uuid(self):
214 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
215 inspect.currentframe().f_code.co_name)
216
217 self.__class__.test_index += 1
218 network = test_config["client"].vim_action("show", "networks", uuid=self.__class__.vim_network_uuid)
219 logger.debug("{}".format(network))
220 self.assertEqual(network.get('network', {}).get('name', ''), self.__class__.vim_network_name)
221
222 def test_030_delete_VIM_network_by_uuid(self):
223 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
224 inspect.currentframe().f_code.co_name)
225
226 self.__class__.test_index += 1
227 network = test_config["client"].vim_action("delete", "networks", uuid=self.__class__.vim_network_uuid)
228 logger.debug("{}".format(network))
229 assert ('deleted' in network.get('result', ""))
230
231
232 class test_VIM_image_operations(test_base):
233
234 def test_000_list_VIM_images(self):
235 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
236 inspect.currentframe().f_code.co_name)
237 self.__class__.test_index += 1
238 images = test_config["client"].vim_action("list", "images")
239 logger.debug("{}".format(images))
240
241 '''
242 The following is a non critical test that will fail most of the times.
243 In case of OpenStack datacenter these tests will only success if RO has access to the admin endpoint
244 This test will only be executed in case it is specifically requested by the user
245 '''
246 class test_VIM_tenant_operations(test_base):
247 vim_tenant_name = None
248 vim_tenant_uuid = None
249
250 @classmethod
251 def setUpClass(cls):
252 test_base.setUpClass(cls)
253 logger.warning("In case of OpenStack datacenter these tests will only success "
254 "if RO has access to the admin endpoint")
255
256 def test_000_create_VIM_tenant(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__.vim_tenant_name = _get_random_string(20)
260 self.__class__.test_index += 1
261 tenant = test_config["client"].vim_action("create", "tenants", name=self.__class__.vim_tenant_name)
262 logger.debug("{}".format(tenant))
263 self.__class__.vim_tenant_uuid = tenant["tenant"]["id"]
264 self.assertEqual(tenant.get('tenant', {}).get('name', ''), self.__class__.vim_tenant_name)
265
266 def test_010_list_VIM_tenants(self):
267 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
268 inspect.currentframe().f_code.co_name)
269 self.__class__.test_index += 1
270 tenants = test_config["client"].vim_action("list", "tenants")
271 logger.debug("{}".format(tenants))
272
273 def test_020_get_VIM_tenant_by_uuid(self):
274 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
275 inspect.currentframe().f_code.co_name)
276
277 self.__class__.test_index += 1
278 tenant = test_config["client"].vim_action("show", "tenants", uuid=self.__class__.vim_tenant_uuid)
279 logger.debug("{}".format(tenant))
280 self.assertEqual(tenant.get('tenant', {}).get('name', ''), self.__class__.vim_tenant_name)
281
282 def test_030_delete_VIM_tenant_by_uuid(self):
283 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
284 inspect.currentframe().f_code.co_name)
285
286 self.__class__.test_index += 1
287 tenant = test_config["client"].vim_action("delete", "tenants", uuid=self.__class__.vim_tenant_uuid)
288 logger.debug("{}".format(tenant))
289 assert ('deleted' in tenant.get('result', ""))
290
291
292 class test_vimconn_connect(test_base):
293
294 def test_000_connect(self):
295 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
296 self.__class__.test_index,
297 inspect.currentframe().f_code.co_name)
298
299 self.__class__.test_index += 1
300 if test_config['vimtype'] == 'vmware':
301 vca_object = test_config["vim_conn"].connect()
302 logger.debug("{}".format(vca_object))
303 self.assertIsNotNone(vca_object)
304 elif test_config['vimtype'] in ('openstack', 'azure'):
305 test_config["vim_conn"]._reload_connection()
306 network_list = test_config["vim_conn"].get_network_list()
307 logger.debug("{}".format(network_list))
308 self.assertIsNotNone(network_list)
309
310
311 class test_vimconn_new_network(test_base):
312 network_name = None
313
314 def test_000_new_network(self):
315 self.__class__.network_name = _get_random_string(20)
316 network_type = 'bridge'
317
318 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
319 self.__class__.test_index, inspect.currentframe().f_code.co_name)
320 self.__class__.test_index += 1
321
322 network, _ = test_config["vim_conn"].new_network(net_name=self.__class__.network_name,
323 net_type=network_type)
324 self.__class__.network_id = network
325 logger.debug("Created network {}".format(network))
326
327 network_list = test_config["vim_conn"].get_network_list()
328 logger.debug("Network list {}".format(network_list))
329 for net in network_list:
330 if self.__class__.network_name in net.get('name'):
331 self.assertIn(self.__class__.network_name, net.get('name'))
332 self.assertEqual(net.get('type'), network_type)
333
334 # Deleting created network
335 result = test_config["vim_conn"].delete_network(self.__class__.network_id)
336 if result:
337 logger.info("Network id {} sucessfully deleted".format(self.__class__.network_id))
338 else:
339 logger.info("Failed to delete network id {}".format(self.__class__.network_id))
340
341 network_list = test_config["vim_conn"].get_network_list()
342 logger.debug("Network list after deletion {}".format(network_list))
343
344 def test_010_new_network_by_types(self):
345 delete_net_ids = []
346 network_types = ['data','bridge','mgmt']
347 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
348 self.__class__.test_index,
349 inspect.currentframe().f_code.co_name)
350 network_list = test_config["vim_conn"].get_network_list()
351 logger.debug("Network list at start {}".format(network_list))
352 self.__class__.test_index += 1
353 for net_type in network_types:
354 self.__class__.network_name = _get_random_string(20)
355 network_id, _ = test_config["vim_conn"].new_network(net_name=self.__class__.network_name,
356 net_type=net_type)
357
358 delete_net_ids.append(network_id)
359 logger.debug("{}".format(network_id))
360
361 network_list = test_config["vim_conn"].get_network_list()
362 for net in network_list:
363 if self.__class__.network_name in net.get('name'):
364 self.assertIn(self.__class__.network_name, net.get('name'))
365 if net_type in net.get('type'):
366 self.assertEqual(net.get('type'), net_type)
367 else:
368 self.assertNotEqual(net.get('type'), net_type)
369
370 # Deleting created network
371 for net_id in delete_net_ids:
372 result = test_config["vim_conn"].delete_network(net_id)
373 if result:
374 logger.info("Network id {} sucessfully deleted".format(net_id))
375 else:
376 logger.info("Failed to delete network id {}".format(net_id))
377 network_list = test_config["vim_conn"].get_network_list()
378 logger.debug("Network list after test {}".format(network_list))
379
380 def test_020_new_network_by_ipprofile(self):
381 test_directory_content = os.listdir(test_config["test_directory"])
382
383 for dir_name in test_directory_content:
384 if dir_name == 'simple_multi_vnfc':
385 self.__class__.scenario_test_path = test_config["test_directory"] + '/'+ dir_name
386 vnfd_files = glob.glob(self.__class__.scenario_test_path+'/vnfd_*.yaml')
387 break
388
389 for vnfd in vnfd_files:
390 with open(vnfd, 'r') as stream:
391 vnf_descriptor = yaml.load(stream, Loader=yaml.Loader)
392
393 #internal_connections_list = vnf_descriptor['vnf']['internal-connections']
394 internal_connections_list = vnf_descriptor['vnfd-catalog']['vnfd'][0]['ip-profiles']
395 for item in internal_connections_list:
396 version = item['ip-version']
397 dhcp_count = item['dhcp-params']['count']
398 dhcp_enabled = item['dhcp-params']['enabled']
399 dhcp_start_address = item['dhcp-params']['start-address']
400 subnet_address = item['subnet-address']
401
402 self.__class__.network_name = _get_random_string(20)
403 ip_profile = {'dhcp_count': dhcp_count,
404 'dhcp_enabled': dhcp_enabled,
405 'dhcp_start_address': dhcp_start_address,
406 'ip_version': version,
407 'subnet_address': subnet_address
408 }
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='mgmt',
415 ip_profile=ip_profile)
416 self.__class__.network_id = network
417 logger.debug("{}".format(network))
418
419 network_list = test_config["vim_conn"].get_network_list()
420 logger.debug("Created network by ip_profile {}".format(network_list))
421 for net in network_list:
422 if self.__class__.network_name in net.get('name'):
423 self.assertIn(self.__class__.network_name, net.get('name'))
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_030_new_network_by_isshared(self):
433 self.__class__.network_name = _get_random_string(20)
434 shared = True
435 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
436 self.__class__.test_index,
437 inspect.currentframe().f_code.co_name)
438 self.__class__.test_index += 1
439 network, _ = test_config["vim_conn"].new_network(net_name=self.__class__.network_name,
440 net_type='bridge',
441 shared=shared)
442 self.__class__.network_id = network
443 logger.debug("{}".format(network))
444
445 network_list = test_config["vim_conn"].get_network_list()
446 for net in network_list:
447 if self.__class__.network_name in net.get('name'):
448 self.assertIn(self.__class__.network_name, net.get('name'))
449 self.assertEqual(net.get('shared'), shared)
450
451 # Deleting created network
452 result = test_config["vim_conn"].delete_network(self.__class__.network_id)
453 if result:
454 logger.info("Network id {} sucessfully deleted".format(self.__class__.network_id))
455 else:
456 logger.info("Failed to delete network id {}".format(self.__class__.network_id))
457
458 def test_040_new_network_by_negative(self):
459 self.__class__.network_name = _get_random_string(20)
460 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
461 self.__class__.test_index,
462 inspect.currentframe().f_code.co_name)
463 self.__class__.test_index += 1
464 network, _ = test_config["vim_conn"].new_network(net_name=self.__class__.network_name,
465 net_type='unknowntype')
466 self.__class__.network_id = network
467 logger.debug("{}".format(network))
468 network_list = test_config["vim_conn"].get_network_list()
469 for net in network_list:
470 if self.__class__.network_name in net.get('name'):
471 self.assertIn(self.__class__.network_name, net.get('name'))
472
473 # Deleting created network
474 result = test_config["vim_conn"].delete_network(self.__class__.network_id)
475 if result:
476 logger.info("Network id {} sucessfully deleted".format(self.__class__.network_id))
477 else:
478 logger.info("Failed to delete network id {}".format(self.__class__.network_id))
479
480 def test_050_refresh_nets_status(self):
481 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
482 self.__class__.test_index,
483 inspect.currentframe().f_code.co_name)
484 self.__class__.test_index += 1
485 # creating new network
486 network_name = _get_random_string(20)
487 net_type = 'bridge'
488 network_id, _ = test_config["vim_conn"].new_network(net_name=network_name,
489 net_type=net_type)
490 # refresh net status
491 net_dict = test_config["vim_conn"].refresh_nets_status([network_id])
492 for attr in net_dict[network_id]:
493 if attr == 'status':
494 self.assertEqual(net_dict[network_id][attr], 'ACTIVE')
495
496 # Deleting created network
497 result = test_config["vim_conn"].delete_network(network_id)
498 if result:
499 logger.info("Network id {} sucessfully deleted".format(network_id))
500 else:
501 logger.info("Failed to delete network id {}".format(network_id))
502
503 def test_060_refresh_nets_status_negative(self):
504 unknown_net_id = str(uuid.uuid4())
505 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
506 self.__class__.test_index,
507 inspect.currentframe().f_code.co_name)
508 self.__class__.test_index += 1
509
510 # refresh net status
511 # if azure network name must have the following format
512 if test_config['vimtype'] == 'azure':
513 unknown_net_id = "/" + "/".join(["subscriptions", test_config["vim_conn"].subscription_id,
514 "resourceGroups", test_config["vim_conn"].resource_group,
515 "providers", "Microsoft.Network",
516 "virtualNetworks", test_config["vim_conn"].vnet_name,
517 "subnets", unknown_net_id])
518 #unknown_net_id = "/subscriptions/ca3d18ab-d373-4afb-a5d6-7c44f098d16a/resourceGroups/osmRG/providers/Microsoft.Network/virtualNetworks/osm_vnet/subnets/unnkown_net"
519
520 net_dict = test_config["vim_conn"].refresh_nets_status([unknown_net_id])
521 if test_config['vimtype'] in ('openstack', 'azure', 'vmware'):
522 self.assertEqual(net_dict[unknown_net_id]['status'], 'DELETED')
523 else:
524 # TODO : Fix vmware connector to return status DELETED as per vimconn.py
525 self.assertEqual(net_dict, {})
526
527 class test_vimconn_get_network_list(test_base):
528 network_name = None
529
530 def setUp(self):
531 # creating new network
532 self.__class__.network_name = _get_random_string(20)
533 self.__class__.net_type = 'bridge'
534 network, _ = test_config["vim_conn"].new_network(net_name=self.__class__.network_name,
535 net_type=self.__class__.net_type)
536 self.__class__.network_id = network
537 logger.debug("{}".format(network))
538
539 def tearDown(self):
540 test_base.tearDown(self)
541
542 # Deleting created network
543 result = test_config["vim_conn"].delete_network(self.__class__.network_id)
544 if result:
545 logger.info("Network id {} sucessfully deleted".format(self.__class__.network_id))
546 else:
547 logger.info("Failed to delete network id {}".format(self.__class__.network_id))
548
549 def test_000_get_network_list(self):
550 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
551 self.__class__.test_index,
552 inspect.currentframe().f_code.co_name)
553 self.__class__.test_index += 1
554
555 network_list = test_config["vim_conn"].get_network_list()
556 for net in network_list:
557 if self.__class__.network_name in net.get('name'):
558 self.assertIn(self.__class__.network_name, net.get('name'))
559 self.assertEqual(net.get('type'), self.__class__.net_type)
560 self.assertEqual(net.get('status'), 'ACTIVE')
561 if test_config['vimtype'] != 'azure':
562 self.assertEqual(net.get('shared'), False)
563
564 def test_010_get_network_list_by_name(self):
565 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
566 self.__class__.test_index,
567 inspect.currentframe().f_code.co_name)
568 self.__class__.test_index += 1
569
570 if test_config['vimtype'] in ('openstack', 'azure'):
571 network_name = test_config['vim_conn'].get_network(self.__class__.network_id)['name']
572 else:
573 network_name = test_config['vim_conn'].get_network_name_by_id(self.__class__.network_id)
574
575 # find network from list by it's name
576 new_network_list = test_config["vim_conn"].get_network_list({'name': network_name})
577 for list_item in new_network_list:
578 if self.__class__.network_name in list_item.get('name'):
579 self.assertEqual(network_name, list_item.get('name'))
580 self.assertEqual(list_item.get('type'), self.__class__.net_type)
581 self.assertEqual(list_item.get('status'), 'ACTIVE')
582
583 def test_020_get_network_list_by_id(self):
584 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
585 self.__class__.test_index,
586 inspect.currentframe().f_code.co_name)
587 self.__class__.test_index += 1
588
589 # find network from list by it's id
590 new_network_list = test_config["vim_conn"].get_network_list({'id':self.__class__.network_id})
591 for list_item in new_network_list:
592 if self.__class__.network_id in list_item.get('id'):
593 self.assertEqual(self.__class__.network_id, list_item.get('id'))
594 self.assertEqual(list_item.get('type'), self.__class__.net_type)
595 self.assertEqual(list_item.get('status'), 'ACTIVE')
596
597 def test_030_get_network_list_by_shared(self):
598 Shared = False
599 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
600 self.__class__.test_index,
601 inspect.currentframe().f_code.co_name)
602 self.__class__.test_index += 1
603
604 if test_config['vimtype'] in ('openstack', 'azure'):
605 network_name = test_config['vim_conn'].get_network(self.__class__.network_id)['name']
606 else:
607 network_name = test_config['vim_conn'].get_network_name_by_id(self.__class__.network_id)
608 # find network from list by it's shared value
609 new_network_list = test_config["vim_conn"].get_network_list({'shared':Shared,
610 'name':network_name})
611 for list_item in new_network_list:
612 if list_item.get('shared') == Shared:
613 self.assertEqual(list_item.get('shared'), Shared)
614 self.assertEqual(list_item.get('type'), self.__class__.net_type)
615 self.assertEqual(network_name, list_item.get('name'))
616
617 def test_040_get_network_list_by_tenant_id(self):
618 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
619 self.__class__.test_index,
620 inspect.currentframe().f_code.co_name)
621 self.__class__.test_index += 1
622
623 tenant_list = test_config["vim_conn"].get_tenant_list()
624 if test_config['vimtype'] in ('openstack', 'azure'):
625 network_name = test_config['vim_conn'].get_network(self.__class__.network_id)['name']
626 else:
627 network_name = test_config['vim_conn'].get_network_name_by_id(self.__class__.network_id)
628
629 for tenant_item in tenant_list:
630 if test_config['tenant'] == tenant_item.get('name'):
631 # find network from list by it's tenant id
632 tenant_id = tenant_item.get('id')
633 new_network_list = test_config["vim_conn"].get_network_list({'tenant_id':tenant_id,
634 'name':network_name})
635 for list_item in new_network_list:
636 self.assertEqual(tenant_id, list_item.get('tenant_id'))
637 self.assertEqual(network_name, list_item.get('name'))
638 self.assertEqual(list_item.get('type'), self.__class__.net_type)
639 self.assertEqual(list_item.get('status'), 'ACTIVE')
640
641 def test_050_get_network_list_by_status(self):
642 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
643 self.__class__.test_index,
644 inspect.currentframe().f_code.co_name)
645 self.__class__.test_index += 1
646 status = 'ACTIVE'
647
648 if test_config['vimtype'] in ('openstack', 'azure'):
649 network_name = test_config['vim_conn'].get_network(self.__class__.network_id)['name']
650 else:
651 network_name = test_config['vim_conn'].get_network_name_by_id(self.__class__.network_id)
652
653 # find network from list by it's status
654 new_network_list = test_config["vim_conn"].get_network_list({'status':status,
655 'name': network_name})
656 for list_item in new_network_list:
657 self.assertIn(self.__class__.network_name, list_item.get('name'))
658 self.assertEqual(list_item.get('type'), self.__class__.net_type)
659 self.assertEqual(list_item.get('status'), status)
660
661 def test_060_get_network_list_by_negative(self):
662 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
663 self.__class__.test_index,
664 inspect.currentframe().f_code.co_name)
665 self.__class__.test_index += 1
666
667 network_list = test_config["vim_conn"].get_network_list({'name': 'unknown_name'})
668 self.assertEqual(network_list, [])
669
670 class test_vimconn_get_network(test_base):
671 network_name = None
672
673 def setUp(self):
674 # creating new network
675 self.__class__.network_name = _get_random_string(20)
676 self.__class__.net_type = 'bridge'
677 network, _ = test_config["vim_conn"].new_network(net_name=self.__class__.network_name,
678 net_type=self.__class__.net_type)
679 self.__class__.network_id = network
680 logger.debug("{}".format(network))
681
682 def tearDown(self):
683 test_base.tearDown(self)
684
685 # Deleting created network
686 result = test_config["vim_conn"].delete_network(self.__class__.network_id)
687 if result:
688 logger.info("Network id {} sucessfully deleted".format(self.__class__.network_id))
689 else:
690 logger.info("Failed to delete network id {}".format(self.__class__.network_id))
691
692 def test_000_get_network(self):
693 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
694 self.__class__.test_index,
695 inspect.currentframe().f_code.co_name)
696 self.__class__.test_index += 1
697
698 network_info = test_config["vim_conn"].get_network(self.__class__.network_id)
699 self.assertEqual(network_info.get('status'), 'ACTIVE')
700 self.assertIn(self.__class__.network_name, network_info.get('name'))
701 self.assertEqual(network_info.get('type'), self.__class__.net_type)
702 self.assertEqual(network_info.get('id'), self.__class__.network_id)
703
704 def test_010_get_network_negative(self):
705 Non_exist_id = str(uuid.uuid4())
706 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
707 self.__class__.test_index,
708 inspect.currentframe().f_code.co_name)
709 self.__class__.test_index += 1
710 with self.assertRaises(Exception) as context:
711 test_config["vim_conn"].get_network(Non_exist_id)
712
713 self.assertEqual((context.exception).http_code, 404)
714
715 class test_vimconn_delete_network(test_base):
716 network_name = None
717
718 def test_000_delete_network(self):
719 # Creating network
720 self.__class__.network_name = _get_random_string(20)
721 self.__class__.net_type = 'bridge'
722 network, _ = test_config["vim_conn"].new_network(net_name=self.__class__.network_name,
723 net_type=self.__class__.net_type)
724 self.__class__.network_id = network
725 logger.debug("{}".format(network))
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
732 result = test_config["vim_conn"].delete_network(self.__class__.network_id)
733 if result:
734 logger.info("Network id {} sucessfully deleted".format(self.__class__.network_id))
735 else:
736 logger.info("Failed to delete network id {}".format(self.__class__.network_id))
737 time.sleep(5)
738 # after deleting network we check in network list
739 network_list = test_config["vim_conn"].get_network_list({ 'id':self.__class__.network_id })
740 self.assertEqual(network_list, [])
741
742 def test_010_delete_network_negative(self):
743 Non_exist_id = str(uuid.uuid4())
744
745 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
746 self.__class__.test_index,
747 inspect.currentframe().f_code.co_name)
748 self.__class__.test_index += 1
749
750 with self.assertRaises(Exception) as context:
751 test_config["vim_conn"].delete_network(Non_exist_id)
752
753 self.assertEqual((context.exception).http_code, 404)
754
755 class test_vimconn_get_flavor(test_base):
756
757 def test_000_get_flavor(self):
758 test_directory_content = os.listdir(test_config["test_directory"])
759
760 for dir_name in test_directory_content:
761 if dir_name == 'simple_linux':
762 self.__class__.scenario_test_path = test_config["test_directory"] + '/'+ dir_name
763 vnfd_files = glob.glob(self.__class__.scenario_test_path+'/vnfd_*.yaml')
764 break
765
766 for vnfd in vnfd_files:
767 with open(vnfd, 'r') as stream:
768 vnf_descriptor = yaml.load(stream, Loader=yaml.Loader)
769
770 vnfc_list = vnf_descriptor['vnf']['VNFC']
771 for item in vnfc_list:
772 if 'ram' in item and 'vcpus' in item and 'disk' in item:
773 ram = item['ram']
774 vcpus = item['vcpus']
775 disk = item['disk']
776
777 flavor_data = {
778 'name' : _get_random_string(20),
779 'ram': ram,
780 'vcpus': vcpus,
781 'disk': disk
782 }
783
784 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
785 self.__class__.test_index,
786 inspect.currentframe().f_code.co_name)
787 self.__class__.test_index += 1
788 # create new flavor
789 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
790 # get flavor by id
791 result = test_config["vim_conn"].get_flavor(flavor_id)
792 self.assertEqual(ram, result['ram'])
793 self.assertEqual(vcpus, result['vcpus'])
794 self.assertEqual(disk, result['disk'])
795
796 # delete flavor
797 result = test_config["vim_conn"].delete_flavor(flavor_id)
798 if result:
799 logger.info("Flavor id {} sucessfully deleted".format(result))
800 else:
801 logger.info("Failed to delete flavor id {}".format(result))
802
803 def test_010_get_flavor_negative(self):
804 Non_exist_flavor_id = str(uuid.uuid4())
805
806 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
807 self.__class__.test_index,
808 inspect.currentframe().f_code.co_name)
809 self.__class__.test_index += 1
810
811 with self.assertRaises(Exception) as context:
812 test_config["vim_conn"].get_flavor(Non_exist_flavor_id)
813
814 self.assertEqual((context.exception).http_code, 404)
815
816 class test_vimconn_new_flavor(test_base):
817 flavor_id = None
818
819 def test_000_new_flavor(self):
820 flavor_data = {'name': _get_random_string(20), 'ram': 1024, 'vcpus': 1, 'disk': 10}
821
822 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
823 self.__class__.test_index,
824 inspect.currentframe().f_code.co_name)
825 self.__class__.test_index += 1
826
827 if test_config['vimtype'] == 'azure':
828 with self.assertRaises(Exception) as context:
829 test_config["vim_conn"].new_flavor(flavor_data)
830
831 self.assertEqual((context.exception).http_code, 401)
832 else:
833 # create new flavor
834 self.__class__.flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
835 self.assertIsInstance(self.__class__.flavor_id, str)
836 self.assertIsInstance(uuid.UUID(self.__class__.flavor_id), uuid.UUID)
837
838 def test_010_delete_flavor(self):
839 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
840 self.__class__.test_index,
841 inspect.currentframe().f_code.co_name)
842 self.__class__.test_index += 1
843
844 # delete flavor
845 if test_config['vimtype'] == 'azure':
846 with self.assertRaises(Exception) as context:
847 test_config["vim_conn"].delete_flavor(self.__class__.flavor_id)
848
849 self.assertEqual((context.exception).http_code, 401)
850 else:
851 result = test_config["vim_conn"].delete_flavor(self.__class__.flavor_id)
852 if result:
853 logger.info("Flavor id {} sucessfully deleted".format(result))
854 else:
855 logger.error("Failed to delete flavor id {}".format(result))
856 raise Exception ("Failed to delete created flavor")
857
858 def test_020_new_flavor_negative(self):
859 Invalid_flavor_data = {'ram': '1024', 'vcpus': 2.0, 'disk': 2.0}
860
861 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
862 self.__class__.test_index,
863 inspect.currentframe().f_code.co_name)
864 self.__class__.test_index += 1
865
866 with self.assertRaises(Exception) as context:
867 test_config["vim_conn"].new_flavor(Invalid_flavor_data)
868 if test_config['vimtype'] != 'azure':
869 self.assertEqual((context.exception).http_code, 400)
870 else:
871 self.assertEqual((context.exception).http_code, 401)
872
873 def test_030_delete_flavor_negative(self):
874 Non_exist_flavor_id = str(uuid.uuid4())
875
876 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
877 self.__class__.test_index,
878 inspect.currentframe().f_code.co_name)
879 self.__class__.test_index += 1
880
881 with self.assertRaises(Exception) as context:
882 test_config["vim_conn"].delete_flavor(Non_exist_flavor_id)
883
884 if test_config['vimtype'] != 'azure':
885 self.assertEqual((context.exception).http_code, 404)
886 else:
887 self.assertEqual((context.exception).http_code, 401)
888
889 # class test_vimconn_new_image(test_base):
890 #
891 # def test_000_new_image(self):
892 # self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
893 # self.__class__.test_index,
894 # inspect.currentframe().f_code.co_name)
895 # self.__class__.test_index += 1
896 #
897 # image_path = test_config['image_path']
898 # if image_path:
899 # self.__class__.image_id = test_config["vim_conn"].new_image({ 'name': 'TestImage', 'location' : image_path, 'metadata': {'upload_location':None} })
900 # time.sleep(20)
901 #
902 # self.assertIsInstance(self.__class__.image_id, (str, unicode))
903 # self.assertIsInstance(uuid.UUID(self.__class__.image_id), uuid.UUID)
904 # else:
905 # self.skipTest("Skipping test as image file not present at RO container")
906 #
907 # def test_010_new_image_negative(self):
908 # Non_exist_image_path = '/temp1/cirros.ovf'
909 #
910 # self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
911 # self.__class__.test_index,
912 # inspect.currentframe().f_code.co_name)
913 # self.__class__.test_index += 1
914 #
915 # with self.assertRaises(Exception) as context:
916 # test_config["vim_conn"].new_image({ 'name': 'TestImage', 'location' : Non_exist_image_path})
917 #
918 # self.assertEqual((context.exception).http_code, 400)
919 #
920 # def test_020_delete_image(self):
921 # self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
922 # self.__class__.test_index,
923 # inspect.currentframe().f_code.co_name)
924 # self.__class__.test_index += 1
925 #
926 # image_id = test_config["vim_conn"].delete_image(self.__class__.image_id)
927 #
928 # self.assertIsInstance(image_id, (str, unicode))
929 #
930 # def test_030_delete_image_negative(self):
931 # Non_exist_image_id = str(uuid.uuid4())
932 #
933 # self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
934 # self.__class__.test_index,
935 # inspect.currentframe().f_code.co_name)
936 # self.__class__.test_index += 1
937 #
938 # with self.assertRaises(Exception) as context:
939 # test_config["vim_conn"].delete_image(Non_exist_image_id)
940 #
941 # self.assertEqual((context.exception).http_code, 404)
942
943 # class test_vimconn_get_image_id_from_path(test_base):
944 #
945 # def test_000_get_image_id_from_path(self):
946 # self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
947 # self.__class__.test_index,
948 # inspect.currentframe().f_code.co_name)
949 # self.__class__.test_index += 1
950 #
951 # image_path = test_config['image_path']
952 # if image_path:
953 # image_id = test_config["vim_conn"].get_image_id_from_path( image_path )
954 # self.assertEqual(type(image_id),str)
955 # else:
956 # self.skipTest("Skipping test as image file not present at RO container")
957 #
958 # def test_010_get_image_id_from_path_negative(self):
959 # Non_exist_image_path = '/temp1/cirros.ovf'
960 #
961 # self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
962 # self.__class__.test_index,
963 # inspect.currentframe().f_code.co_name)
964 # self.__class__.test_index += 1
965 #
966 # with self.assertRaises(Exception) as context:
967 # test_config["vim_conn"].new_image({ 'name': 'TestImage', 'location' : Non_exist_image_path })
968 #
969 # self.assertEqual((context.exception).http_code, 400)
970
971 class test_vimconn_get_image_list(test_base):
972 image_name = None
973 image_id = None
974
975 def test_000_get_image_list(self):
976 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
977 self.__class__.test_index,
978 inspect.currentframe().f_code.co_name)
979 self.__class__.test_index += 1
980
981 if test_config['vimtype'] != 'azure':
982 image_list = test_config["vim_conn"].get_image_list()
983 logger.debug("{}: Result image list: {}".format(self.__class__.test_text, image_list))
984
985 for item in image_list:
986 if 'name' in item:
987 self.__class__.image_name = item['name']
988 self.__class__.image_id = item['id']
989 self.assertIsInstance(self.__class__.image_name, str)
990 self.assertIsInstance(self.__class__.image_id, str)
991 else:
992 with self.assertRaises(Exception) as context:
993 image_list = test_config["vim_conn"].get_image_list()
994 self.assertEqual((context.exception).http_code, 401)
995 logger.debug(self.__class__.test_text + "Exception unauthorized: " + str(context.exception))
996
997 def test_010_get_image_list_by_name(self):
998 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
999 self.__class__.test_index,
1000 inspect.currentframe().f_code.co_name)
1001 self.__class__.test_index += 1
1002 self.__class__.image_name = test_config['image_name']
1003 logger.debug("{}: Image name: {}".format(self.__class__.test_text, self.__class__.image_name))
1004
1005 image_list = test_config["vim_conn"].get_image_list({'name': self.__class__.image_name})
1006 logger.debug("{}: Result image list: {}".format(self.__class__.test_text, image_list))
1007
1008 for item in image_list:
1009 self.assertIsInstance(item['id'], str)
1010 self.assertIsInstance(item['name'], str)
1011 #self.assertEqual(item['id'], self.__class__.image_id)
1012 self.assertEqual(item['name'], self.__class__.image_name)
1013
1014 def test_020_get_image_list_by_id(self):
1015 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1016 self.__class__.test_index,
1017 inspect.currentframe().f_code.co_name)
1018 self.__class__.test_index += 1
1019
1020 filter_image_list = test_config["vim_conn"].get_image_list({'id': self.__class__.image_id})
1021
1022 for item1 in filter_image_list:
1023 self.assertIsInstance(item1['id'], str)
1024 self.assertIsInstance(item1['name'], str)
1025 self.assertEqual(item1['id'], self.__class__.image_id)
1026 self.assertEqual(item1['name'], self.__class__.image_name)
1027
1028 def test_030_get_image_list_negative(self):
1029 Non_exist_image_id = uuid.uuid4()
1030 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1031 self.__class__.test_index,
1032 inspect.currentframe().f_code.co_name)
1033 self.__class__.test_index += 1
1034 image_list = test_config["vim_conn"].get_image_list({'name': 'Unknown_name', 'id': Non_exist_image_id})
1035
1036 self.assertIsNotNone(image_list, None)
1037 self.assertEqual(image_list, [])
1038
1039 class test_vimconn_new_vminstance(test_base):
1040 network_name = None
1041 net_type = None
1042 network_id = None
1043 image_id = None
1044 instance_id = None
1045
1046 def setUp(self):
1047 # create network
1048 self.__class__.network_name = _get_random_string(20)
1049 self.__class__.net_type = 'bridge'
1050
1051 self.__class__.network_id, _ = test_config["vim_conn"].new_network(net_name=self.__class__.network_name,
1052 net_type=self.__class__.net_type)
1053 # find image name and image id
1054 if test_config['image_name']:
1055 image_list = test_config['vim_conn'].get_image_list({'name': test_config['image_name']})
1056 if len(image_list) == 0:
1057 raise Exception("Image {} is not found at VIM".format(test_config['image_name']))
1058 else:
1059 self.__class__.image_id = image_list[0]['id']
1060 else:
1061 image_list = test_config['vim_conn'].get_image_list()
1062 if len(image_list) == 0:
1063 raise Exception("Not found any image at VIM")
1064 else:
1065 self.__class__.image_id = image_list[0]['id']
1066
1067 def tearDown(self):
1068 test_base.tearDown(self)
1069 # Deleting created network
1070 result = test_config["vim_conn"].delete_network(self.__class__.network_id)
1071 if result:
1072 logger.info("Network id {} sucessfully deleted".format(self.__class__.network_id))
1073 else:
1074 logger.info("Failed to delete network id {}".format(self.__class__.network_id))
1075
1076 def test_000_new_vminstance(self):
1077 vpci = "0000:00:11.0"
1078 name = "eth0"
1079
1080 flavor_data = {'name': _get_random_string(20), 'ram': 1024, 'vcpus': 1, 'disk': 10}
1081
1082 # create new flavor
1083 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
1084
1085
1086 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1087 self.__class__.test_index,
1088 inspect.currentframe().f_code.co_name)
1089 self.__class__.test_index += 1
1090
1091 net_list = [{'use': self.__class__.net_type, 'name': name, 'floating_ip': False, 'vpci': vpci,
1092 'port_security': True, 'type': 'virtual', 'net_id': self.__class__.network_id}]
1093
1094 self.__class__.instance_id, _ = test_config["vim_conn"].new_vminstance(name='Test1_vm', description='',
1095 start=False,
1096 image_id=self.__class__.image_id,
1097 flavor_id=flavor_id, net_list=net_list)
1098
1099 self.assertIsInstance(self.__class__.instance_id, str)
1100
1101 def test_010_new_vminstance_by_model(self):
1102 flavor_data = {'name': _get_random_string(20),'ram': 1024, 'vcpus': 2, 'disk': 10}
1103 model_name = 'e1000'
1104 name = 'eth0'
1105
1106 # create new flavor
1107 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
1108
1109 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1110 self.__class__.test_index,
1111 inspect.currentframe().f_code.co_name)
1112 self.__class__.test_index += 1
1113
1114 net_list = [{'use': self.__class__.net_type, 'name': name, 'floating_ip': False, 'port_security': True,
1115 'model': model_name, 'type': 'virtual', 'net_id': self.__class__.network_id}]
1116
1117 instance_id, _ = test_config["vim_conn"].new_vminstance(name='Test1_vm', description='', start=False,
1118 image_id=self.__class__.image_id,
1119 flavor_id=flavor_id,net_list=net_list)
1120
1121 self.assertIsInstance(instance_id, str)
1122
1123 # Deleting created vm instance
1124 logger.info("Deleting created vm intance")
1125 test_config["vim_conn"].delete_vminstance(instance_id)
1126 time.sleep(10)
1127
1128 def test_020_new_vminstance_by_net_use(self):
1129 flavor_data = {'name': _get_random_string(20),'ram': 1024, 'vcpus': 2, 'disk': 10}
1130 net_use = 'data'
1131 name = 'eth0'
1132
1133 # create new flavor
1134 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
1135
1136 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1137 self.__class__.test_index,
1138 inspect.currentframe().f_code.co_name)
1139 self.__class__.test_index += 1
1140
1141 net_list = [{'use': net_use, 'name': name, 'floating_ip': False, 'port_security': True, 'type': 'virtual',
1142 'net_id': self.__class__.network_id}]
1143
1144 instance_id, _ = test_config["vim_conn"].new_vminstance(name='Test1_vm', description='', start=False,
1145 image_id=self.__class__.image_id,disk_list=None,
1146 flavor_id=flavor_id, net_list=net_list)
1147 self.assertIsInstance(instance_id, str)
1148
1149 # Deleting created vm instance
1150 logger.info("Deleting created vm intance")
1151 test_config["vim_conn"].delete_vminstance(instance_id)
1152 time.sleep(10)
1153
1154 def test_030_new_vminstance_by_net_type(self):
1155 flavor_data = {'name':_get_random_string(20),'ram': 1024, 'vcpus': 2, 'disk': 10}
1156 _type = 'VF'
1157 name = 'eth0'
1158
1159 # create new flavor
1160 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
1161
1162 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1163 self.__class__.test_index,
1164 inspect.currentframe().f_code.co_name)
1165 self.__class__.test_index += 1
1166
1167 if test_config['vimtype'] == 'vmware':
1168 net_list = [{'use': self.__class__.net_type, 'name': name, 'floating_ip': False, 'port_security': True,
1169 'type': _type, 'net_id': self.__class__.network_id}]
1170
1171 instance_id, _ = test_config["vim_conn"].new_vminstance(name='Test1_vm', image_id=self.__class__.image_id,
1172 flavor_id=flavor_id,
1173 net_list=net_list)
1174 self.assertEqual(type(instance_id),str)
1175
1176 if test_config['vimtype'] in ('openstack', 'azure'):
1177 # create network of type data
1178 network_name = _get_random_string(20)
1179 net_type = 'data'
1180
1181 network_id, _ = test_config["vim_conn"].new_network(net_name=network_name,
1182 net_type=net_type)
1183 net_list = [{'use': net_type, 'name': name, 'floating_ip': False, 'port_security': True,
1184 'type': _type, 'net_id': network_id}]
1185
1186 instance_id, _ = test_config["vim_conn"].new_vminstance(name='Test1_vm', description='', start=False,
1187 image_id=self.__class__.image_id, disk_list=None,
1188 flavor_id=flavor_id,
1189 net_list=net_list)
1190
1191 self.assertEqual(type(instance_id), unicode)
1192
1193 # delete created network
1194 result = test_config["vim_conn"].delete_network(network_id)
1195 if result:
1196 logger.info("Network id {} sucessfully deleted".format(network_id))
1197 else:
1198 logger.info("Failed to delete network id {}".format(network_id))
1199
1200 # Deleting created vm instance
1201 logger.info("Deleting created vm intance")
1202 test_config["vim_conn"].delete_vminstance(instance_id)
1203 time.sleep(10)
1204
1205 def test_040_new_vminstance_by_cloud_config(self):
1206 flavor_data = {'name': _get_random_string(20),'ram': 1024, 'vcpus': 2, 'disk': 10}
1207 name = 'eth0'
1208 user_name = 'test_user'
1209
1210 key_pairs = ['ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDtAjl5R+GSKP3gFrdFxgizKEUzhXKQbyjaxJH9thsK 0/fDiYlaNEjvijgPgiVZkfwvqgWeLprPcpzL2j4jvmmSJ3+7C8ihCwObWP0VUiuewmbIINBPAR0RqusjMRyPsa+q0asFBPOoZLx3Cv3vzmC1AA3mKuCNeT EuA0rlWhDIOVwMcU5sP1grnmuexQB8HcR7BdKcA9y08pTwnCQR8vmtW77SRkaxEGXm4Gnw5qw8Z27mHdk2wWS2SnbVH7aFwWvDXc6jjf5TpEWypdr/EAPC +eJipeS2Oa4FsntEqAu3Fz6gp/9ub8uNqgCgHfMzs6FhYpZpipwS0hXYyF6eVsSx osm@osm']
1211
1212 users_data = [{'key-pairs': ['ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDtAjl5R+GSKP3gFrdFxgizKEUzhXKQbyjaxJH9thsK0/fDiYlaNEjvijgPgiVZkfwvqgWeLprPcpzL2j4jvmmSJ3+7C8ihCwObWP0VUiuewmbIINBPAR0RqusjMRyPsa+q0asFBPOoZLx3Cv3vzmC1AA3mKuCNeTEuA0rlWhDIOVwMcU5sP1grnmuexQB8HcR7BdKcA9y08pTwnCQR8vmtW77SRkaxEGXm4Gnw5qw8Z27mHdk2wWS2SnbVH7aFwWvDXc6jjf5TpEWypdr/EAPC+eJipeS2Oa4FsntEqAu3Fz6gp/9ub8uNqgCgHfMzs6FhYpZpipwS0hXYyF6eVsSx osm@osm'], 'name': 'cloudinit'}]
1213
1214 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 }
1215 #cloud_data = {'users': users_data }
1216
1217 # create new flavor
1218 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
1219
1220 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1221 self.__class__.test_index,
1222 inspect.currentframe().f_code.co_name)
1223 self.__class__.test_index += 1
1224
1225 net_list = [{'use': self.__class__.net_type, 'name': name, 'floating_ip': False, 'port_security': True,
1226 'type': 'virtual', 'net_id': self.__class__.network_id}]
1227
1228 instance_id, _ = test_config["vim_conn"].new_vminstance(name='Cloud_vm', description='', start=False,
1229 image_id=self.__class__.image_id,
1230 flavor_id=flavor_id,net_list=net_list,
1231 cloud_config=cloud_data)
1232
1233 self.assertIsInstance(instance_id, str)
1234
1235 # Deleting created vm instance
1236 logger.info("Deleting created vm intance")
1237 test_config["vim_conn"].delete_vminstance(instance_id)
1238 time.sleep(10)
1239
1240 def test_050_new_vminstance_by_disk_list(self):
1241 flavor_data = {'name':_get_random_string(20),'ram': 1024, 'vcpus': 2, 'disk': 10}
1242 name = 'eth0'
1243
1244 device_data = [{'image_id': self.__class__.image_id, 'size': '10'}]
1245
1246 # create new flavor
1247 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
1248
1249 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1250 self.__class__.test_index,
1251 inspect.currentframe().f_code.co_name)
1252 self.__class__.test_index += 1
1253
1254 net_list = [{'use': self.__class__.net_type, 'name': name, 'floating_ip': False, 'port_security': True,
1255 'type': 'virtual', 'net_id': self.__class__.network_id}]
1256
1257 instance_id, _ = test_config["vim_conn"].new_vminstance(name='VM_test1', description='', start=False,
1258 image_id=self.__class__.image_id,
1259 flavor_id=flavor_id, net_list=net_list,
1260 disk_list=device_data)
1261
1262 self.assertIsInstance(instance_id, str)
1263 # Deleting created vm instance
1264 logger.info("Deleting created vm intance")
1265 test_config["vim_conn"].delete_vminstance(instance_id)
1266 time.sleep(10)
1267
1268 def test_060_new_vminstance_negative(self):
1269 unknown_flavor_id = str(uuid.uuid4())
1270 unknown_image_id = str(uuid.uuid4())
1271 name = 'eth2'
1272
1273 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1274 self.__class__.test_index,
1275 inspect.currentframe().f_code.co_name)
1276 self.__class__.test_index += 1
1277
1278 net_list = [{'use': self.__class__.net_type, 'name': name, 'floating_ip': False, 'port_security': True,
1279 'type': 'virtual', 'net_id': self.__class__.network_id}]
1280
1281 with self.assertRaises(Exception) as context:
1282 test_config["vim_conn"].new_vminstance(name='Test1_vm', description='', start=False,
1283 image_id=unknown_image_id,
1284 flavor_id=unknown_flavor_id,
1285 net_list=net_list)
1286
1287 self.assertIn((context.exception).http_code, (400, 404))
1288
1289
1290 def test_070_get_vminstance(self):
1291 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1292 self.__class__.test_index,
1293 inspect.currentframe().f_code.co_name)
1294 self.__class__.test_index += 1
1295
1296 # Get instance by its id
1297 vm_info = test_config["vim_conn"].get_vminstance(self.__class__.instance_id)
1298
1299 if test_config['vimtype'] == 'vmware':
1300 for attr in vm_info:
1301 if attr == 'status':
1302 self.assertEqual(vm_info[attr], 'ACTIVE')
1303 if attr == 'hostId':
1304 self.assertEqual(type(vm_info[attr]), str)
1305 if attr == 'interfaces':
1306 self.assertEqual(type(vm_info[attr]), list)
1307 self.assertEqual(vm_info[attr][0]['IsConnected'], 'true')
1308 if attr == 'IsEnabled':
1309 self.assertEqual(vm_info[attr], 'true')
1310
1311 def test_080_get_vminstance_negative(self):
1312 unknown_instance_id = str(uuid.uuid4())
1313
1314 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1315 self.__class__.test_index,
1316 inspect.currentframe().f_code.co_name)
1317 self.__class__.test_index += 1
1318
1319 with self.assertRaises(Exception) as context:
1320 test_config["vim_conn"].get_vminstance(unknown_instance_id)
1321
1322 self.assertEqual((context.exception).http_code, 404)
1323
1324 def test_090_refresh_vms_status(self):
1325 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1326 self.__class__.test_index,
1327 inspect.currentframe().f_code.co_name)
1328 self.__class__.test_index += 1
1329
1330 if test_config['vimtype'] == 'vmware':
1331 vm_list = []
1332 vm_list.append(self.__class__.instance_id)
1333
1334 # refresh vm status
1335 vm_info = test_config["vim_conn"].refresh_vms_status(vm_list)
1336 for attr in vm_info[self.__class__.instance_id]:
1337 if attr == 'status':
1338 self.assertEqual(vm_info[self.__class__.instance_id][attr], 'ACTIVE')
1339 if attr == 'interfaces':
1340 self.assertEqual(type(vm_info[self.__class__.instance_id][attr]), list)
1341
1342 if test_config['vimtype'] in ('openstack', 'azure'):
1343 vpci = "0000:00:11.0"
1344 name = "eth0"
1345
1346 flavor_data = {'name': _get_random_string(20), 'ram': 1024, 'vcpus': 1, 'disk': 10}
1347
1348 # create new flavor
1349 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
1350 # create new vm instance
1351 net_list = [{'use': self.__class__.net_type, 'name': name, 'floating_ip': False, 'vpci': vpci,
1352 'port_security': True, 'type': 'virtual', 'net_id': self.__class__.network_id}]
1353
1354 instance_id, _ = test_config["vim_conn"].new_vminstance(name='Test1_vm', description='', start=False,
1355 image_id=self.__class__.image_id,
1356 flavor_id=flavor_id, net_list=net_list)
1357
1358 time.sleep(30)
1359 vm_list = []
1360 vm_list.append(instance_id)
1361
1362 # refresh vm status
1363 vm_info = test_config["vim_conn"].refresh_vms_status(vm_list)
1364 for attr in vm_info[instance_id]:
1365 if attr == 'status':
1366 self.assertEqual(vm_info[instance_id][attr], 'ACTIVE')
1367 if attr == 'interfaces':
1368 self.assertEqual(type(vm_info[instance_id][attr]), list)
1369
1370 #Deleting created vm instance
1371 logger.info("Deleting created vm intance")
1372 test_config["vim_conn"].delete_vminstance(instance_id)
1373 time.sleep(10)
1374
1375
1376 def test_100_refresh_vms_status_negative(self):
1377 unknown_id = str(uuid.uuid4())
1378
1379 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1380 self.__class__.test_index,
1381 inspect.currentframe().f_code.co_name)
1382 self.__class__.test_index += 1
1383
1384 vm_dict = test_config["vim_conn"].refresh_vms_status([unknown_id])
1385
1386 if test_config['vimtype'] == 'vmware':
1387 self.assertEqual(vm_dict,{})
1388
1389 if test_config['vimtype'] in ('openstack', 'azure'):
1390 self.assertEqual(vm_dict[unknown_id]['status'], 'DELETED')
1391
1392 def test_110_action_vminstance(self):
1393 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1394 self.__class__.test_index,
1395 inspect.currentframe().f_code.co_name)
1396 self.__class__.test_index += 1
1397
1398 if test_config['vimtype'] == 'vmware':
1399 action_list = ['shutdown', 'start', 'shutoff', 'rebuild', 'pause', 'resume']
1400 # various action on vminstace
1401 for action in action_list:
1402 instance_id = test_config["vim_conn"].action_vminstance(self.__class__.instance_id,
1403 {action: None})
1404 self.assertEqual(instance_id, self.__class__.instance_id)
1405
1406 if test_config['vimtype'] in ('openstack', 'azure'):
1407 # create new vm instance
1408 vpci = "0000:00:11.0"
1409 name = "eth0"
1410
1411 flavor_data = {'name': _get_random_string(20), 'ram': 1024, 'vcpus': 1, 'disk': 10}
1412
1413 # create new flavor
1414 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
1415
1416 net_list = [{'use': self.__class__.net_type, 'name': name, 'floating_ip': False, 'vpci': vpci,
1417 'port_security': True, 'type': 'virtual', 'net_id': self.__class__.network_id}]
1418
1419 new_instance_id, _ = test_config["vim_conn"].new_vminstance(name='Test1_vm', description='',
1420 start=False, image_id=self.__class__.image_id,
1421 flavor_id=flavor_id, net_list=net_list)
1422
1423 if test_config['vimtype'] == 'openstack':
1424 action_list = ['shutdown','start','shutoff','rebuild','start','pause','start']
1425 else:
1426 action_list = ['shutdown','start','stop','start','shutoff','start','reboot']
1427
1428 # various action on vminstace
1429 for action in action_list:
1430 # sleep for sometime till status is changed
1431 time.sleep(25)
1432 instance_id = test_config["vim_conn"].action_vminstance(new_instance_id,
1433 { action: None})
1434
1435 self.assertTrue(instance_id is None)
1436
1437 # Deleting created vm instance
1438 logger.info("Deleting created vm intance")
1439 test_config["vim_conn"].delete_vminstance(new_instance_id)
1440 time.sleep(10)
1441
1442 def test_120_action_vminstance_negative(self):
1443 non_exist_id = str(uuid.uuid4())
1444 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1445 self.__class__.test_index,
1446 inspect.currentframe().f_code.co_name)
1447 self.__class__.test_index += 1
1448
1449 action = 'start'
1450 with self.assertRaises(Exception) as context:
1451 test_config["vim_conn"].action_vminstance(non_exist_id, { action: None})
1452
1453 self.assertEqual((context.exception).http_code, 404)
1454
1455
1456 def test_130_delete_vminstance(self):
1457 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1458 self.__class__.test_index,
1459 inspect.currentframe().f_code.co_name)
1460 self.__class__.test_index += 1
1461
1462 # Deleting created vm instance
1463 logger.info("Deleting created vm instance")
1464 test_config["vim_conn"].delete_vminstance(self.__class__.instance_id)
1465 time.sleep(10)
1466
1467 def test_140_new_vminstance_sriov(self):
1468 logger.info("Testing creation of sriov vm instance using {}".format(test_config['sriov_net_name']))
1469 flavor_data = {'name': _get_random_string(20),'ram': 1024, 'vcpus': 2, 'disk': 10}
1470 name = 'eth0'
1471
1472 # create new flavor
1473 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
1474
1475 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1476 self.__class__.test_index,
1477 inspect.currentframe().f_code.co_name)
1478 self.__class__.test_index += 1
1479
1480 sriov_net_name = test_config['sriov_net_name']
1481 new_network_list = test_config["vim_conn"].get_network_list({'name': sriov_net_name})
1482 for list_item in new_network_list:
1483 self.assertEqual(sriov_net_name, list_item.get('name'))
1484 self.__class__.sriov_network_id = list_item.get('id')
1485
1486 net_list = [{'use': 'data', 'name': name, 'floating_ip': False, 'port_security': True, 'type': 'VF',
1487 'net_id': self.__class__.sriov_network_id}]
1488
1489 instance_id, _ = test_config["vim_conn"].new_vminstance(name='Test1_sriov_vm', description='', start=False,
1490 image_id=self.__class__.image_id, flavor_id=flavor_id,
1491 net_list=net_list)
1492
1493 self.assertIsInstance(instance_id, str)
1494
1495 logger.info("Waiting for created sriov-vm intance")
1496 time.sleep(10)
1497 # Deleting created vm instance
1498 logger.info("Deleting created sriov-vm intance")
1499 test_config["vim_conn"].delete_vminstance(instance_id)
1500 time.sleep(10)
1501
1502 class test_vimconn_get_tenant_list(test_base):
1503 tenant_id = None
1504
1505 def test_000_get_tenant_list(self):
1506 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1507 self.__class__.test_index,
1508 inspect.currentframe().f_code.co_name)
1509 self.__class__.test_index += 1
1510
1511 # Getting tenant list
1512 tenant_list = test_config["vim_conn"].get_tenant_list()
1513 logger.debug(self.__class__.test_text + "Tenant list: " + str(tenant_list))
1514
1515 for item in tenant_list:
1516 if test_config['tenant'] == item['name']:
1517 self.__class__.tenant_id = item['id']
1518 self.assertIsInstance(item['name'], str)
1519 self.assertIsInstance(item['id'], str)
1520
1521 def test_010_get_tenant_list_by_id(self):
1522 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1523 self.__class__.test_index,
1524 inspect.currentframe().f_code.co_name)
1525 self.__class__.test_index += 1
1526
1527 # Getting filter tenant list by its id
1528 filter_tenant_list = test_config["vim_conn"].get_tenant_list({'id': self.__class__.tenant_id})
1529 logger.debug(self.__class__.test_text + "Tenant list: " + str(filter_tenant_list))
1530
1531 for item in filter_tenant_list:
1532 self.assertIsInstance(item['id'], str)
1533 self.assertEqual(item['id'], self.__class__.tenant_id)
1534
1535 def test_020_get_tenant_list_by_name(self):
1536 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1537 self.__class__.test_index,
1538 inspect.currentframe().f_code.co_name)
1539 self.__class__.test_index += 1
1540
1541 # Getting filter tenant list by its name
1542 filter_tenant_list = test_config["vim_conn"].get_tenant_list({'name': test_config['tenant']})
1543 logger.debug(self.__class__.test_text + "Tenant list: " + str(filter_tenant_list))
1544
1545 for item in filter_tenant_list:
1546 self.assertIsInstance(item['name'], str)
1547 self.assertEqual(item['name'], test_config['tenant'])
1548
1549 def test_030_get_tenant_list_by_name_and_id(self):
1550 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1551 self.__class__.test_index,
1552 inspect.currentframe().f_code.co_name)
1553 self.__class__.test_index += 1
1554
1555 # Getting filter tenant list by its name and id
1556 filter_tenant_list = test_config["vim_conn"].get_tenant_list({'name': test_config['tenant'],
1557 'id': self.__class__.tenant_id})
1558 logger.debug(self.__class__.test_text + "Tenant list: " + str(filter_tenant_list))
1559
1560 for item in filter_tenant_list:
1561 self.assertIsInstance(item['name'], str)
1562 self.assertIsInstance(item['id'], str)
1563 self.assertEqual(item['name'], test_config['tenant'])
1564 self.assertEqual(item['id'], self.__class__.tenant_id)
1565
1566 def test_040_get_tenant_list_negative(self):
1567 non_exist_tenant_name = "Tenant_123"
1568 non_exist_tenant_id = "kjhgrt456-45345kjhdfgnbdk-34dsfjdfg"
1569 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1570 self.__class__.test_index,
1571 inspect.currentframe().f_code.co_name)
1572 self.__class__.test_index += 1
1573
1574 filter_tenant_list = test_config["vim_conn"].get_tenant_list({'name': non_exist_tenant_name,
1575 'id': non_exist_tenant_id})
1576 logger.debug(self.__class__.test_text + "Tenant list: " + str(filter_tenant_list))
1577
1578 self.assertEqual(filter_tenant_list, [])
1579
1580
1581 class test_vimconn_new_tenant(test_base):
1582 tenant_id = None
1583
1584 def test_000_new_tenant(self):
1585 tenant_name = _get_random_string(20)
1586 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1587 self.__class__.test_index,
1588 inspect.currentframe().f_code.co_name)
1589 self.__class__.test_index += 1
1590
1591 if test_config['vimtype'] != 'azure':
1592 self.__class__.tenant_id = test_config["vim_conn"].new_tenant(tenant_name, "")
1593 time.sleep(15)
1594
1595 self.assertIsInstance(self.__class__.tenant_id, str)
1596 else:
1597 with self.assertRaises(Exception) as context:
1598 test_config["vim_conn"].new_tenant(self.__class__.tenant_id, "")
1599 self.assertEqual((context.exception).http_code, 401)
1600 logger.debug(self.__class__.test_text + "Exception unauthorized: " + str(context.exception))
1601
1602
1603 def test_010_new_tenant_negative(self):
1604 Invalid_tenant_name = 10121
1605 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1606 self.__class__.test_index,
1607 inspect.currentframe().f_code.co_name)
1608 self.__class__.test_index += 1
1609
1610 with self.assertRaises(Exception) as context:
1611 test_config["vim_conn"].new_tenant(Invalid_tenant_name, "")
1612
1613 if test_config['vimtype'] != 'azure':
1614 self.assertEqual((context.exception).http_code, 400)
1615 else:
1616 self.assertEqual((context.exception).http_code, 401)
1617 logger.debug(self.__class__.test_text + "Exception unauthorized: " + str(context.exception))
1618
1619
1620 def test_020_delete_tenant(self):
1621 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1622 self.__class__.test_index,
1623 inspect.currentframe().f_code.co_name)
1624 self.__class__.test_index += 1
1625
1626 if test_config['vimtype'] != 'azure':
1627 tenant_id = test_config["vim_conn"].delete_tenant(self.__class__.tenant_id)
1628 self.assertIsInstance(tenant_id, str)
1629 else:
1630 with self.assertRaises(Exception) as context:
1631 test_config["vim_conn"].delete_tenant(self.__class__.tenant_id)
1632 self.assertEqual((context.exception).http_code, 401)
1633 logger.debug(self.__class__.test_text + "Exception unauthorized: " + str(context.exception))
1634
1635 def test_030_delete_tenant_negative(self):
1636 non_exist_tenant_name = 'Test_30_tenant'
1637 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1638 self.__class__.test_index,
1639 inspect.currentframe().f_code.co_name)
1640 self.__class__.test_index += 1
1641
1642 with self.assertRaises(Exception) as context:
1643 test_config["vim_conn"].delete_tenant(non_exist_tenant_name)
1644
1645 if test_config['vimtype'] != 'azure':
1646 self.assertEqual((context.exception).http_code, 404)
1647 else:
1648 self.assertEqual((context.exception).http_code, 401)
1649 logger.debug(self.__class__.test_text + "Exception unauthorized: " + str(context.exception))
1650
1651
1652 def get_image_id():
1653 if test_config['image_name']:
1654 image_list = test_config['vim_conn'].get_image_list({'name': test_config['image_name']})
1655 if len(image_list) == 0:
1656 raise Exception("Image {} is not found at VIM".format(test_config['image_name']))
1657 else:
1658 image_id = image_list[0]['id']
1659 else:
1660 image_list = test_config['vim_conn'].get_image_list()
1661 if len(image_list) == 0:
1662 raise Exception("Not found any image at VIM")
1663 else:
1664 image_id = image_list[0]['id']
1665 return image_id
1666
1667
1668 class test_vimconn_vminstance_by_ip_address(test_base):
1669 network_name = None
1670 network_id = None
1671
1672 def setUp(self):
1673 # create network
1674 self.network_name = _get_random_string(20)
1675
1676 self.network_id, _ = test_config["vim_conn"].new_network(net_name=self.network_name,
1677 net_type='bridge')
1678
1679 def tearDown(self):
1680 test_base.tearDown(self)
1681 # Deleting created network
1682 result = test_config["vim_conn"].delete_network(self.network_id)
1683 if result:
1684 logger.info("Network id {} sucessfully deleted".format(self.network_id))
1685 else:
1686 logger.info("Failed to delete network id {}".format(self.network_id))
1687
1688
1689 def test_000_vminstance_by_ip_address(self):
1690 """
1691 This test case will deploy VM with provided IP address
1692 Pre-requesite: provided IP address should be from IP pool range which has used for network creation
1693 """
1694 name = "eth0"
1695 # provide ip address
1696 ip_address = ''
1697
1698 flavor_data = {'ram': 1024, 'vcpus': 1, 'disk': 10}
1699
1700 # create new flavor
1701 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
1702
1703 # find image id
1704 image_id = get_image_id()
1705
1706 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1707 self.__class__.test_index,
1708 inspect.currentframe().f_code.co_name)
1709 self.__class__.test_index += 1
1710
1711 net_list = [{'use': 'bridge', 'name': name, 'floating_ip': False, 'port_security': True, 'type': 'virtual',
1712 'net_id': self.network_id, 'ip_address': ip_address}]
1713
1714 instance_id, _ = test_config["vim_conn"].new_vminstance(name='Test1_vm', image_id=image_id,
1715 flavor_id=flavor_id, net_list=net_list)
1716
1717 self.assertEqual(type(instance_id),str)
1718 logger.info("Deleting created vm instance")
1719 test_config["vim_conn"].delete_vminstance(instance_id)
1720 time.sleep(10)
1721
1722 def test_010_vminstance_by_ip_address_negative(self):
1723 name = "eth1"
1724 # IP address not from subnet range
1725 invalid_ip_address = '10.10.12.1'
1726
1727 flavor_data = {'ram': 1024, 'vcpus': 1, 'disk': 10}
1728
1729 # create new flavor
1730 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
1731
1732 # find image name and image id
1733 image_id = get_image_id()
1734
1735 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1736 self.__class__.test_index,
1737 inspect.currentframe().f_code.co_name)
1738 self.__class__.test_index += 1
1739
1740 net_list = [{'use': 'bridge', 'name': name, 'floating_ip': False, 'port_security': True, 'type': 'virtual',
1741 'net_id': self.network_id, 'ip_address': invalid_ip_address}]
1742
1743 with self.assertRaises(Exception) as context:
1744 test_config["vim_conn"].new_vminstance(name='Test1_vm', image_id=image_id,
1745 flavor_id=flavor_id,
1746 net_list=net_list)
1747 self.assertEqual((context.exception).http_code, 400)
1748
1749 def test_020_vminstance_by_floating_ip(self):
1750 name = "eth1"
1751 flavor_data = {'ram': 1024, 'vcpus': 1, 'disk': 10}
1752
1753 # create new flavor
1754 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
1755
1756 # find image name and image id
1757 image_id = get_image_id()
1758
1759 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1760 self.__class__.test_index,
1761 inspect.currentframe().f_code.co_name)
1762 self.__class__.test_index += 1
1763
1764 net_list = [{'use': 'bridge', 'name': name, 'floating_ip': True, 'port_security': True, 'type': 'virtual',
1765 'net_id': self.network_id}]
1766
1767 instance_id, _ = test_config["vim_conn"].new_vminstance(name='Test1_vm', image_id=image_id,
1768 flavor_id=flavor_id, net_list=net_list)
1769
1770 self.assertEqual(type(instance_id),str)
1771 logger.info("Deleting created vm instance")
1772 test_config["vim_conn"].delete_vminstance(instance_id)
1773 time.sleep(10)
1774
1775 def test_030_vminstance_by_mac_address(self):
1776 name = "eth1"
1777 mac_address = "74:54:2f:21:da:8c"
1778 flavor_data = {'ram': 1024, 'vcpus': 1, 'disk': 10}
1779
1780 # create new flavor
1781 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
1782
1783 # find image name and image id
1784 image_id = get_image_id()
1785
1786 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1787 self.__class__.test_index,
1788 inspect.currentframe().f_code.co_name)
1789 self.__class__.test_index += 1
1790
1791 net_list = [{'use': 'bridge', 'name': name, 'floating_ip': False, 'port_security': True, 'type': 'virtual',
1792 'net_id': self.network_id,'mac_address': mac_address}]
1793
1794 instance_id, _ = test_config["vim_conn"].new_vminstance(name='Test1_vm', image_id=image_id,
1795 flavor_id=flavor_id, net_list=net_list)
1796
1797 self.assertEqual(type(instance_id),str)
1798 logger.info("Deleting created vm instance")
1799 test_config["vim_conn"].delete_vminstance(instance_id)
1800 time.sleep(10)
1801
1802 class test_vimconn_vminstance_by_adding_10_nics(test_base):
1803 network_name = None
1804 net_ids = []
1805
1806 def setUp(self):
1807 # create network
1808 i = 0
1809 for i in range(10):
1810 self.network_name = _get_random_string(20)
1811 network_id, _ = test_config["vim_conn"].new_network(net_name=self.network_name,
1812 net_type='bridge')
1813 self.net_ids.append(network_id)
1814
1815 def tearDown(self):
1816 test_base.tearDown(self)
1817 # Deleting created network
1818 for net_id in self.net_ids:
1819 result = test_config["vim_conn"].delete_network(net_id)
1820 if result:
1821 logger.info("Network id {} sucessfully deleted".format(net_id))
1822 else:
1823 logger.info("Failed to delete network id {}".format(net_id))
1824
1825 def test_000_vminstance_by_adding_10_nics(self):
1826 flavor_data = {'ram': 1024, 'vcpus': 1, 'disk': 10}
1827
1828 # create new flavor
1829 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
1830
1831 # find image name and image id
1832 image_id = get_image_id()
1833
1834 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1835 self.__class__.test_index,
1836 inspect.currentframe().f_code.co_name)
1837 self.__class__.test_index += 1
1838
1839 net_list = []
1840 c = 1
1841 for net_id in self.net_ids:
1842 name = "eth{}".format(c)
1843 net_list.append({'use': 'bridge', 'name': name, 'floating_ip': False,
1844 'port_security': True, 'type': 'virtual', 'net_id': net_id})
1845 c = c+1
1846
1847 instance_id, _ = test_config["vim_conn"].new_vminstance(name='Test1_vm', image_id=image_id,
1848 flavor_id=flavor_id, net_list=net_list)
1849
1850 self.assertEqual(type(instance_id),str)
1851 logger.info("Deleting created vm instance")
1852 test_config["vim_conn"].delete_vminstance(instance_id)
1853 time.sleep(10)
1854
1855
1856 class test_vimconn_vminstance_by_existing_disk(test_base):
1857 network_name = None
1858 network_id = None
1859
1860 def setUp(self):
1861 # create network
1862 self.network_name = _get_random_string(20)
1863 self.network_id, _ = test_config["vim_conn"].new_network(net_name=self.network_name,
1864 net_type='bridge')
1865
1866 def tearDown(self):
1867 test_base.tearDown(self)
1868 # Deleting created network
1869 result = test_config["vim_conn"].delete_network(self.network_id)
1870 if result:
1871 logger.info("Network id {} sucessfully deleted".format(self.network_id))
1872 else:
1873 logger.info("Failed to delete network id {}".format(self.network_id))
1874
1875
1876 def test_000_vminstance_by_existing_disk(self):
1877 """ This testcase will add existing disk only if given catalog/image is free
1878 means not used by any other VM
1879 """
1880
1881 flavor_data = {'ram': 1024, 'vcpus': 1, 'disk': 10}
1882 name = "eth10"
1883
1884 # create new flavor
1885 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
1886
1887 # find image name and image id
1888 image_id = get_image_id()
1889 cirros_image = test_config["vim_conn"].get_image_list({'name': 'cirros'})
1890 disk_list = [{'image_id': cirros_image[0]['id'],'size': 5}]
1891
1892 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1893 self.__class__.test_index,
1894 inspect.currentframe().f_code.co_name)
1895 self.__class__.test_index += 1
1896
1897 net_list = [{'use': 'bridge', 'name': name, 'floating_ip': False, 'port_security': True,
1898 'type': 'virtual', 'net_id': self.network_id}]
1899
1900 instance_id, _ = test_config["vim_conn"].new_vminstance(name='Test1_vm', description='', start=False,
1901 image_id=image_id,
1902 flavor_id=flavor_id, net_list=net_list,disk_list=disk_list)
1903
1904 self.assertEqual(type(instance_id),str)
1905 logger.info("Deleting created vm instance")
1906 test_config["vim_conn"].delete_vminstance(instance_id)
1907 time.sleep(10)
1908
1909 def test_010_vminstance_by_new_disk(self):
1910 flavor_data = {'ram': 1024, 'vcpus': 1, 'disk': 10}
1911 name = "eth10"
1912
1913 # create new flavor
1914 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
1915
1916 # find image name and image id
1917 image_id = get_image_id()
1918 disk_list = [{'size': '5'}]
1919
1920 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1921 self.__class__.test_index,
1922 inspect.currentframe().f_code.co_name)
1923 self.__class__.test_index += 1
1924
1925 net_list = [{'use': 'bridge', 'name': name, 'floating_ip': False, 'port_security': True,
1926 'type': 'virtual', 'net_id': self.network_id}]
1927
1928 instance_id, _ = test_config["vim_conn"].new_vminstance(name='Test1_vm', description='', start=False,
1929 image_id=image_id,
1930 flavor_id=flavor_id, net_list=net_list,disk_list=disk_list)
1931
1932 self.assertEqual(type(instance_id),str)
1933 logger.info("Deleting created vm instance")
1934 test_config["vim_conn"].delete_vminstance(instance_id)
1935 time.sleep(10)
1936
1937 def test_020_vminstance_by_CDROM(self):
1938 """ This testcase will insert media file only if provided catalog
1939 has pre-created ISO media file into vCD
1940 """
1941 flavor_data ={'ram': 1024, 'vcpus': 1, 'disk': 10}
1942 name = "eth10"
1943 image_list = test_config["vim_conn"].get_image_list({'name':'Ubuntu'})
1944 disk_list = [{'image_id':image_list[0]['id'],'device_type':'cdrom'}]
1945
1946 # create new flavor
1947 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
1948
1949 # find image name and image id
1950 image_id = get_image_id()
1951
1952 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
1953 self.__class__.test_index,
1954 inspect.currentframe().f_code.co_name)
1955 self.__class__.test_index += 1
1956
1957 net_list = [{'use': 'bridge', 'name': name, 'floating_ip': False, 'port_security': True,
1958 'type': 'virtual', 'net_id': self.network_id}]
1959
1960 instance_id, _ = test_config["vim_conn"].new_vminstance(name='Test1_vm', description='', start=False,
1961 image_id=image_id,
1962 flavor_id=flavor_id, net_list=net_list,disk_list=disk_list )
1963
1964 self.assertEqual(type(instance_id),str)
1965 logger.info("Deleting created vm instance")
1966 test_config["vim_conn"].delete_vminstance(instance_id)
1967 time.sleep(10)
1968
1969
1970 class test_vimconn_vminstance_by_affinity_anti_affinity(test_base):
1971 network_name = None
1972 network_id = None
1973
1974 def setUp(self):
1975 # create network
1976 self.network_name = _get_random_string(20)
1977 self.network_id, _ = test_config["vim_conn"].new_network(net_name=self.network_name,
1978 net_type='bridge')
1979
1980 def tearDown(self):
1981 test_base.tearDown(self)
1982 # Deleting created network
1983 result = test_config["vim_conn"].delete_network(self.network_id)
1984 if result:
1985 logger.info("Network id {} sucessfully deleted".format(self.network_id))
1986 else:
1987 logger.info("Failed to delete network id {}".format(self.network_id))
1988
1989 def test_000_vminstance_by_affinity_anti_affinity(self):
1990 """ This testcase will deploy VM into provided HOSTGROUP in VIM config
1991 Pre-requisites: User has created Hosh Groups in vCenter with respective Hosts to be used
1992 While creating VIM account user has to pass the Host Group names in availability_zone list
1993 """
1994 flavor_data = {'ram': 1024, 'vcpus': 1, 'disk': 10}
1995 name = "eth10"
1996
1997 # create new flavor
1998 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
1999
2000 # find image name and image id
2001 image_id = get_image_id()
2002
2003 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
2004 self.__class__.test_index,
2005 inspect.currentframe().f_code.co_name)
2006 self.__class__.test_index += 1
2007
2008 net_list = [{'use': 'bridge', 'name': name, 'floating_ip': False, 'port_security': True,
2009 'type': 'virtual', 'net_id': self.network_id}]
2010
2011 instance_id, _ = test_config["vim_conn"].new_vminstance(name='Test1_vm', description='', start=False,
2012 image_id=image_id,
2013 flavor_id=flavor_id, net_list=net_list,availability_zone_index=1,
2014 availability_zone_list=['HG_174','HG_175'])
2015
2016 self.assertEqual(type(instance_id),str)
2017 time.sleep(10)
2018 logger.info("Deleting created vm instance")
2019 test_config["vim_conn"].delete_vminstance(instance_id)
2020
2021 class test_vimconn_vminstance_by_numa_affinity(test_base):
2022 network_name = None
2023 network_id = None
2024
2025 def setUp(self):
2026 # create network
2027 self.network_name = _get_random_string(20)
2028 self.network_id, _ = test_config["vim_conn"].new_network(net_name=self.network_name,
2029 net_type='bridge')
2030
2031 def tearDown(self):
2032 test_base.tearDown(self)
2033 # Deleting created network
2034 result = test_config["vim_conn"].delete_network(self.network_id)
2035 if result:
2036 logger.info("Network id {} sucessfully deleted".format(self.network_id))
2037 else:
2038 logger.info("Failed to delete network id {}".format(self.network_id))
2039
2040 def test_000_vminstance_by_numa_affinity(self):
2041 flavor_data = {'extended': {'numas': [{'paired-threads-id': [['1', '3'], ['2', '4']],
2042 ' paired-threads': 2, 'memory': 1}]},
2043 'ram': 1024, 'vcpus': 1, 'disk': 10}
2044 name = "eth10"
2045
2046 # create new flavor
2047 flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
2048
2049 # find image name and image id
2050 image_id = get_image_id()
2051
2052 self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
2053 self.__class__.test_index,
2054 inspect.currentframe().f_code.co_name)
2055 self.__class__.test_index += 1
2056
2057 net_list = [{'use': 'bridge', 'name': name, 'floating_ip': False, 'port_security': True,
2058 'type': 'virtual', 'net_id': self.network_id}]
2059
2060 instance_id, _ = test_config["vim_conn"].new_vminstance(name='Test1_vm', description='', start=False,
2061 image_id=image_id,
2062 flavor_id=flavor_id, net_list=net_list)
2063
2064 self.assertEqual(type(instance_id),str)
2065 logger.info("Deleting created vm instance")
2066 test_config["vim_conn"].delete_vminstance(instance_id)
2067 time.sleep(10)
2068
2069
2070 '''
2071 IMPORTANT NOTE
2072 The following unittest class does not have the 'test_' on purpose. This test is the one used for the
2073 scenario based tests.
2074 '''
2075 class descriptor_based_scenario_test(test_base):
2076 test_index = 0
2077 scenario_test_path = None
2078
2079 @classmethod
2080 def setUpClass(cls):
2081 cls.test_index = 1
2082 cls.to_delete_list = []
2083 cls.scenario_uuids = []
2084 cls.instance_scenario_uuids = []
2085 cls.scenario_test_path = test_config["test_directory"] + '/' + test_config["test_folder"]
2086 logger.info("{}. {} {}".format(test_config["test_number"], cls.__name__, test_config["test_folder"]))
2087
2088 @classmethod
2089 def tearDownClass(cls):
2090 test_config["test_number"] += 1
2091
2092 def test_000_load_scenario(self):
2093 self.__class__.test_text = "{}.{}. TEST {} {}".format(test_config["test_number"], self.__class__.test_index,
2094 inspect.currentframe().f_code.co_name,
2095 test_config["test_folder"])
2096 self.__class__.test_index += 1
2097 # load VNFD and NSD
2098 descriptor_files = glob.glob(self.__class__.scenario_test_path+'/*.yaml')
2099 vnf_descriptors = []
2100 scenario_descriptors = []
2101 for descriptor_file in descriptor_files:
2102 with open(descriptor_file, 'r') as stream:
2103 descriptor = yaml.load(stream, Loader=yaml.Loader)
2104 if "vnf" in descriptor or "vnfd:vnfd-catalog" in descriptor or "vnfd-catalog" in descriptor:
2105 vnf_descriptors.append(descriptor)
2106 else:
2107 scenario_descriptors.append(descriptor)
2108
2109 scenario_file = glob.glob(self.__class__.scenario_test_path + '/scenario_*.yaml')
2110 if not vnf_descriptors or not scenario_descriptors or len(scenario_descriptors) > 1:
2111 raise Exception("Test '{}' not valid. It must contain an scenario file and at least one vnfd file'".format(
2112 test_config["test_folder"]))
2113
2114 # load all vnfd
2115 for vnf_descriptor in vnf_descriptors:
2116 logger.debug("VNF descriptor: {}".format(vnf_descriptor))
2117 vnf = test_config["client"].create_vnf(descriptor=vnf_descriptor, image_name=test_config["image_name"])
2118 logger.debug(vnf)
2119 if 'vnf' in vnf:
2120 vnf_uuid = vnf['vnf']['uuid']
2121 else:
2122 vnf_uuid = vnf['vnfd'][0]['uuid']
2123 self.__class__.to_delete_list.insert(0, {"item": "vnf", "function": test_config["client"].delete_vnf,
2124 "params": {"uuid": vnf_uuid}})
2125
2126 # load the scenario definition
2127 for scenario_descriptor in scenario_descriptors:
2128 # networks = scenario_descriptor['scenario']['networks']
2129 # networks[test_config["mgmt_net"]] = networks.pop('mgmt')
2130 logger.debug("Scenario descriptor: {}".format(scenario_descriptor))
2131 scenario = test_config["client"].create_scenario(descriptor=scenario_descriptor)
2132 logger.debug(scenario)
2133 if 'scenario' in scenario:
2134 scenario_uuid = scenario['scenario']['uuid']
2135 else:
2136 scenario_uuid = scenario['nsd'][0]['uuid']
2137 self.__class__.to_delete_list.insert(0, {"item": "scenario",
2138 "function": test_config["client"].delete_scenario,
2139 "params": {"uuid": scenario_uuid}})
2140 self.__class__.scenario_uuids.append(scenario_uuid)
2141
2142 def test_010_instantiate_scenario(self):
2143 self.__class__.test_text = "{}.{}. TEST {} {}".format(test_config["test_number"], self.__class__.test_index,
2144 inspect.currentframe().f_code.co_name,
2145 test_config["test_folder"])
2146 self.__class__.test_index += 1
2147 for scenario_uuid in self.__class__.scenario_uuids:
2148 instance_descriptor = {
2149 "instance":{
2150 "name": self.__class__.test_text,
2151 "scenario": scenario_uuid,
2152 "networks":{
2153 "mgmt": {"sites": [ { "netmap-use": test_config["mgmt_net"]} ]}
2154 }
2155 }
2156 }
2157 instance = test_config["client"].create_instance(instance_descriptor)
2158 self.__class__.instance_scenario_uuids.append(instance['uuid'])
2159 logger.debug(instance)
2160 self.__class__.to_delete_list.insert(0, {"item": "instance",
2161 "function": test_config["client"].delete_instance,
2162 "params": {"uuid": instance['uuid']}})
2163
2164 def test_020_check_deployent(self):
2165 self.__class__.test_text = "{}.{}. TEST {} {}".format(test_config["test_number"], self.__class__.test_index,
2166 inspect.currentframe().f_code.co_name,
2167 test_config["test_folder"])
2168 self.__class__.test_index += 1
2169
2170 if test_config["manual"]:
2171 input('Scenario has been deployed. Perform manual check and press any key to resume')
2172 return
2173
2174 keep_waiting = test_config["timeout"]
2175 pending_instance_scenario_uuids = list(self.__class__.instance_scenario_uuids) # make a copy
2176 while pending_instance_scenario_uuids:
2177 index = 0
2178 while index < len(pending_instance_scenario_uuids):
2179 result = check_instance_scenario_active(pending_instance_scenario_uuids[index])
2180 if result[0]:
2181 del pending_instance_scenario_uuids[index]
2182 break
2183 elif 'ERROR' in result[1]:
2184 msg = 'Got error while waiting for the instance to get active: '+result[1]
2185 logging.error(msg)
2186 raise Exception(msg)
2187 index += 1
2188
2189 if keep_waiting >= 5:
2190 time.sleep(5)
2191 keep_waiting -= 5
2192 elif keep_waiting > 0:
2193 time.sleep(keep_waiting)
2194 keep_waiting = 0
2195 else:
2196 msg = 'Timeout reached while waiting instance scenario to get active'
2197 logging.error(msg)
2198 raise Exception(msg)
2199
2200 def test_030_clean_deployment(self):
2201 self.__class__.test_text = "{}.{}. TEST {} {}".format(test_config["test_number"], self.__class__.test_index,
2202 inspect.currentframe().f_code.co_name,
2203 test_config["test_folder"])
2204 self.__class__.test_index += 1
2205 #At the moment if you delete an scenario right after creating it, in openstack datacenters
2206 #sometimes scenario ports get orphaned. This sleep is just a dirty workaround
2207 time.sleep(5)
2208 for item in self.__class__.to_delete_list:
2209 response = item["function"](**item["params"])
2210 logger.debug(response)
2211
2212
2213 def _get_random_string(maxLength):
2214 '''generates a string with random characters string.letters and string.digits
2215 with a random length up to maxLength characters. If maxLength is <15 it will be changed automatically to 15
2216 '''
2217 prefix = 'testing_'
2218 min_string = 15
2219 minLength = min_string - len(prefix)
2220 if maxLength < min_string: maxLength = min_string
2221 maxLength -= len(prefix)
2222 length = random.randint(minLength,maxLength)
2223 return 'testing_'+"".join([random.choice(string.ascii_letters+string.digits) for i in range(length)])
2224
2225
2226 def test_vimconnector(args):
2227 global test_config
2228 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + "/osm_ro")
2229 test_config['vimtype'] = args.vimtype
2230 if args.vimtype == "vmware":
2231 from osm_rovim_vmware import vimconn_vmware as vim
2232
2233 test_config["test_directory"] = os.path.dirname(__file__) + "/RO_tests"
2234
2235 tenant_name = args.tenant_name
2236 test_config['tenant'] = tenant_name
2237 config_params = yaml.load(args.config_param, Loader=yaml.Loader)
2238 org_name = config_params.get('orgname')
2239 org_user = config_params.get('user')
2240 org_passwd = config_params.get('passwd')
2241 vim_url = args.endpoint_url
2242 test_config['image_path'] = args.image_path
2243 test_config['image_name'] = args.image_name
2244 test_config['sriov_net_name'] = args.sriov_net_name
2245
2246 # vmware connector obj
2247 test_config['vim_conn'] = vim.vimconnector(name=org_name, tenant_name=tenant_name, user=org_user,
2248 passwd=org_passwd, url=vim_url, config=config_params)
2249
2250 elif args.vimtype == "aws":
2251 from osm_rovim_aws import vimconn_aws as vim
2252 elif args.vimtype == "openstack":
2253 from osm_rovim_openstack import vimconn_openstack as vim
2254
2255 test_config["test_directory"] = os.path.dirname(__file__) + "/RO_tests"
2256
2257 tenant_name = args.tenant_name
2258 test_config['tenant'] = tenant_name
2259 config_params = yaml.load(args.config_param, Loader=yaml.Loader)
2260 os_user = config_params.get('user')
2261 os_passwd = config_params.get('passwd')
2262 vim_url = args.endpoint_url
2263 test_config['image_path'] = args.image_path
2264 test_config['image_name'] = args.image_name
2265 test_config['sriov_net_name'] = args.sriov_net_name
2266
2267 # openstack connector obj
2268 vim_persistent_info = {}
2269 test_config['vim_conn'] = vim.vimconnector(
2270 uuid="test-uuid-1", name="VIO-openstack",
2271 tenant_id=None, tenant_name=tenant_name,
2272 url=vim_url, url_admin=None,
2273 user=os_user, passwd=os_passwd,
2274 config=config_params, persistent_info=vim_persistent_info
2275 )
2276 test_config['vim_conn'].debug = "true"
2277
2278 elif args.vimtype == "openvim":
2279 from osm_rovim_openvim import vimconn_openvim as vim
2280 elif args.vimtype == "azure":
2281 from osm_rovim_azure import vimconn_azure as vim
2282
2283 test_config["test_directory"] = os.path.dirname(__file__) + "/RO_tests"
2284
2285 tenant_name = args.tenant_name
2286 test_config['tenant'] = tenant_name
2287 config_params = yaml.load(args.config_param)
2288 os_user = config_params.get('user')
2289 os_passwd = config_params.get('passwd')
2290 vim_url = args.endpoint_url
2291 test_config['image_path'] = args.image_path
2292 test_config['image_name'] = args.image_name
2293 #test_config['sriov_net_name'] = args.sriov_net_name
2294 args_log_level = "DEBUG" if args.debug else "INFO"
2295
2296 # azure connector obj
2297 vim_persistent_info = {}
2298 test_config['vim_conn'] = vim.vimconnector(
2299 uuid="test-uuid-1", name="VIO-azure",
2300 tenant_id=None, tenant_name=tenant_name,
2301 url=vim_url, url_admin=None,
2302 user=os_user, passwd=os_passwd, log_level= args_log_level,
2303 config=config_params, persistent_info=vim_persistent_info
2304 )
2305 test_config['vim_conn'].debug = "true"
2306
2307 else:
2308 logger.critical("vimtype '{}' not supported".format(args.vimtype))
2309 sys.exit(1)
2310 executed = 0
2311 failed = 0
2312 clsmembers = inspect.getmembers(sys.modules[__name__], inspect.isclass)
2313 # If only want to obtain a tests list print it and exit
2314 if args.list_tests:
2315 tests_names = []
2316 for cls in clsmembers:
2317 if cls[0].startswith('test_vimconn'):
2318 tests_names.append(cls[0])
2319
2320 msg = "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names))
2321 print(msg)
2322 logger.info(msg)
2323 sys.exit(0)
2324
2325 # Create the list of tests to be run
2326 code_based_tests = []
2327 if args.tests:
2328 for test in args.tests:
2329 for t in test.split(','):
2330 matches_code_based_tests = [item for item in clsmembers if item[0] == t]
2331 if len(matches_code_based_tests) > 0:
2332 code_based_tests.append(matches_code_based_tests[0][1])
2333 else:
2334 logger.critical("Test '{}' is not among the possible ones".format(t))
2335 sys.exit(1)
2336 if not code_based_tests:
2337 # include all tests
2338 for cls in clsmembers:
2339 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
2340 if cls[0].startswith('test_vimconn'):
2341 code_based_tests.append(cls[1])
2342
2343 logger.debug("tests to be executed: {}".format(code_based_tests))
2344
2345 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
2346 # This is handled in the tests using logging.
2347 stream = open('/dev/null', 'w')
2348
2349 # Run code based tests
2350 basic_tests_suite = unittest.TestSuite()
2351 for test in code_based_tests:
2352 basic_tests_suite.addTest(unittest.makeSuite(test))
2353 result = unittest.TextTestRunner(stream=stream, failfast=failfast).run(basic_tests_suite)
2354 executed += result.testsRun
2355 failed += len(result.failures) + len(result.errors)
2356 if failfast and failed:
2357 sys.exit(1)
2358 if len(result.failures) > 0:
2359 logger.debug("failures : {}".format(result.failures))
2360 if len(result.errors) > 0:
2361 logger.debug("errors : {}".format(result.errors))
2362 return executed, failed
2363
2364
2365 def test_vim(args):
2366 global test_config
2367 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + "/osm_ro")
2368 from osm_ro import openmanoclient
2369 executed = 0
2370 failed = 0
2371 test_config["client"] = openmanoclient.openmanoclient(
2372 endpoint_url=args.endpoint_url,
2373 tenant_name=args.tenant_name,
2374 datacenter_name=args.datacenter,
2375 debug=args.debug, logger=test_config["logger_name"])
2376 clsmembers = inspect.getmembers(sys.modules[__name__], inspect.isclass)
2377 # If only want to obtain a tests list print it and exit
2378 if args.list_tests:
2379 tests_names = []
2380 for cls in clsmembers:
2381 if cls[0].startswith('test_VIM'):
2382 tests_names.append(cls[0])
2383
2384 msg = "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names)) +\
2385 "\nNOTE: The test test_VIM_tenant_operations will fail in case the used datacenter is type OpenStack " \
2386 "unless RO has access to the admin endpoint. Therefore this test is excluded by default"
2387 print(msg)
2388 logger.info(msg)
2389 sys.exit(0)
2390
2391 # Create the list of tests to be run
2392 code_based_tests = []
2393 if args.tests:
2394 for test in args.tests:
2395 for t in test.split(','):
2396 matches_code_based_tests = [item for item in clsmembers if item[0] == t]
2397 if len(matches_code_based_tests) > 0:
2398 code_based_tests.append(matches_code_based_tests[0][1])
2399 else:
2400 logger.critical("Test '{}' is not among the possible ones".format(t))
2401 sys.exit(1)
2402 if not code_based_tests:
2403 # include all tests
2404 for cls in clsmembers:
2405 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
2406 if cls[0].startswith('test_VIM') and cls[0] != 'test_VIM_tenant_operations':
2407 code_based_tests.append(cls[1])
2408
2409 logger.debug("tests to be executed: {}".format(code_based_tests))
2410
2411 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
2412 # This is handled in the tests using logging.
2413 stream = open('/dev/null', 'w')
2414
2415 # Run code based tests
2416 basic_tests_suite = unittest.TestSuite()
2417 for test in code_based_tests:
2418 basic_tests_suite.addTest(unittest.makeSuite(test))
2419 result = unittest.TextTestRunner(stream=stream, failfast=failfast).run(basic_tests_suite)
2420 executed += result.testsRun
2421 failed += len(result.failures) + len(result.errors)
2422 if failfast and failed:
2423 sys.exit(1)
2424 if len(result.failures) > 0:
2425 logger.debug("failures : {}".format(result.failures))
2426 if len(result.errors) > 0:
2427 logger.debug("errors : {}".format(result.errors))
2428 return executed, failed
2429
2430
2431 def test_wim(args):
2432 global test_config
2433 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + "/osm_ro")
2434 from osm_ro import openmanoclient
2435 executed = 0
2436 failed = 0
2437 test_config["client"] = openmanoclient.openmanoclient(
2438 endpoint_url=args.endpoint_url,
2439 tenant_name=args.tenant_name,
2440 datacenter_name=args.datacenter,
2441 debug=args.debug, logger=test_config["logger_name"])
2442 clsmembers = inspect.getmembers(sys.modules[__name__], inspect.isclass)
2443 # If only want to obtain a tests list print it and exit
2444 if args.list_tests:
2445 tests_names = []
2446 for cls in clsmembers:
2447 if cls[0].startswith('test_WIM'):
2448 tests_names.append(cls[0])
2449
2450 msg = "The 'wim' set tests are:\n\t" + ', '.join(sorted(tests_names)) +\
2451 "\nNOTE: The test test_VIM_tenant_operations will fail in case the used datacenter is type OpenStack " \
2452 "unless RO has access to the admin endpoint. Therefore this test is excluded by default"
2453 print(msg)
2454 logger.info(msg)
2455 sys.exit(0)
2456
2457 # Create the list of tests to be run
2458 code_based_tests = []
2459 if args.tests:
2460 for test in args.tests:
2461 for t in test.split(','):
2462 matches_code_based_tests = [item for item in clsmembers if item[0] == t]
2463 if len(matches_code_based_tests) > 0:
2464 code_based_tests.append(matches_code_based_tests[0][1])
2465 else:
2466 logger.critical("Test '{}' is not among the possible ones".format(t))
2467 sys.exit(1)
2468 if not code_based_tests:
2469 # include all tests
2470 for cls in clsmembers:
2471 # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
2472 if cls[0].startswith('test_VIM') and cls[0] != 'test_VIM_tenant_operations':
2473 code_based_tests.append(cls[1])
2474
2475 logger.debug("tests to be executed: {}".format(code_based_tests))
2476
2477 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
2478 # This is handled in the tests using logging.
2479 stream = open('/dev/null', 'w')
2480
2481 # Run code based tests
2482 basic_tests_suite = unittest.TestSuite()
2483 for test in code_based_tests:
2484 basic_tests_suite.addTest(unittest.makeSuite(test))
2485 result = unittest.TextTestRunner(stream=stream, failfast=failfast).run(basic_tests_suite)
2486 executed += result.testsRun
2487 failed += len(result.failures) + len(result.errors)
2488 if failfast and failed:
2489 sys.exit(1)
2490 if len(result.failures) > 0:
2491 logger.debug("failures : {}".format(result.failures))
2492 if len(result.errors) > 0:
2493 logger.debug("errors : {}".format(result.errors))
2494 return executed, failed
2495
2496
2497 def test_deploy(args):
2498 global test_config
2499 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + "/osm_ro")
2500 from osm_ro import openmanoclient
2501 executed = 0
2502 failed = 0
2503 test_config["test_directory"] = os.path.dirname(__file__) + "/RO_tests"
2504 test_config["image_name"] = args.image_name
2505 test_config["mgmt_net"] = args.mgmt_net
2506 test_config["manual"] = args.manual
2507 test_directory_content = os.listdir(test_config["test_directory"])
2508 # If only want to obtain a tests list print it and exit
2509 if args.list_tests:
2510 msg = "the 'deploy' set tests are:\n\t" + ', '.join(sorted(test_directory_content))
2511 print(msg)
2512 # logger.info(msg)
2513 sys.exit(0)
2514
2515 descriptor_based_tests = []
2516 # Create the list of tests to be run
2517 code_based_tests = []
2518 if args.tests:
2519 for test in args.tests:
2520 for t in test.split(','):
2521 if t in test_directory_content:
2522 descriptor_based_tests.append(t)
2523 else:
2524 logger.critical("Test '{}' is not among the possible ones".format(t))
2525 sys.exit(1)
2526 if not descriptor_based_tests:
2527 # include all tests
2528 descriptor_based_tests = test_directory_content
2529
2530 logger.debug("tests to be executed: {}".format(code_based_tests))
2531
2532 # import openmanoclient from relative path
2533 test_config["client"] = openmanoclient.openmanoclient(
2534 endpoint_url=args.endpoint_url,
2535 tenant_name=args.tenant_name,
2536 datacenter_name=args.datacenter,
2537 debug=args.debug, logger=test_config["logger_name"])
2538
2539 # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
2540 # This is handled in the tests using logging.
2541 stream = open('/dev/null', 'w')
2542 # This scenario based tests are defined as directories inside the directory defined in 'test_directory'
2543 for test in descriptor_based_tests:
2544 test_config["test_folder"] = test
2545 test_suite = unittest.TestSuite()
2546 test_suite.addTest(unittest.makeSuite(descriptor_based_scenario_test))
2547 result = unittest.TextTestRunner(stream=stream, failfast=False).run(test_suite)
2548 executed += result.testsRun
2549 failed += len(result.failures) + len(result.errors)
2550 if failfast and failed:
2551 sys.exit(1)
2552 if len(result.failures) > 0:
2553 logger.debug("failures : {}".format(result.failures))
2554 if len(result.errors) > 0:
2555 logger.debug("errors : {}".format(result.errors))
2556
2557 return executed, failed
2558
2559 if __name__=="__main__":
2560
2561 parser = ArgumentParser(description='Test RO module')
2562 parser.add_argument('-v','--version', action='version', help="Show current version",
2563 version='%(prog)s version ' + __version__ + ' ' + version_date)
2564
2565 # Common parameters
2566 parent_parser = ArgumentParser(add_help=False)
2567 parent_parser.add_argument('--failfast', help='Stop when a test fails rather than execute all tests',
2568 dest='failfast', action="store_true", default=False)
2569 parent_parser.add_argument('--failed', help='Set logs to show only failed tests. --debug disables this option',
2570 dest='failed', action="store_true", default=False)
2571 default_logger_file = os.path.dirname(__file__)+'/'+os.path.splitext(os.path.basename(__file__))[0]+'.log'
2572 parent_parser.add_argument('--list-tests', help='List all available tests', dest='list_tests', action="store_true",
2573 default=False)
2574 parent_parser.add_argument('--logger_file', dest='logger_file', default=default_logger_file,
2575 help='Set the logger file. By default '+default_logger_file)
2576 parent_parser.add_argument("-t", '--tenant', dest='tenant_name', default="osm",
2577 help="Set the openmano tenant to use for the test. By default 'osm'")
2578 parent_parser.add_argument('--debug', help='Set logs to debug level', dest='debug', action="store_true")
2579 parent_parser.add_argument('--timeout', help='Specify the instantiation timeout in seconds. By default 300',
2580 dest='timeout', type=int, default=300)
2581 parent_parser.add_argument('--test', '--tests', help='Specify the tests to run', dest='tests', action="append")
2582
2583 subparsers = parser.add_subparsers(help='test sets')
2584
2585 # Deployment test set
2586 # -------------------
2587 deploy_parser = subparsers.add_parser('deploy', parents=[parent_parser],
2588 help="test deployment using descriptors at RO_test folder ")
2589 deploy_parser.set_defaults(func=test_deploy)
2590
2591 # Mandatory arguments
2592 mandatory_arguments = deploy_parser.add_argument_group('mandatory arguments')
2593 mandatory_arguments.add_argument('-d', '--datacenter', required=True, help='Set the datacenter to test')
2594 mandatory_arguments.add_argument("-i", '--image-name', required=True, dest="image_name",
2595 help='Image name available at datacenter used for the tests')
2596 mandatory_arguments.add_argument("-n", '--mgmt-net-name', required=True, dest='mgmt_net',
2597 help='Set the vim management network to use for tests')
2598
2599 # Optional arguments
2600 deploy_parser.add_argument('-m', '--manual-check', dest='manual', action="store_true", default=False,
2601 help='Pause execution once deployed to allow manual checking of the '
2602 'deployed instance scenario')
2603 deploy_parser.add_argument('-u', '--url', dest='endpoint_url', default='http://localhost:9090/openmano',
2604 help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
2605
2606 # Vimconn test set
2607 # -------------------
2608 vimconn_parser = subparsers.add_parser('vimconn', parents=[parent_parser], help="test vimconnector plugin")
2609 vimconn_parser.set_defaults(func=test_vimconnector)
2610 # Mandatory arguments
2611 mandatory_arguments = vimconn_parser.add_argument_group('mandatory arguments')
2612 mandatory_arguments.add_argument('--vimtype', choices=['vmware', 'aws', 'openstack', 'openvim','azure'], required=True,
2613 help='Set the vimconnector type to test')
2614 mandatory_arguments.add_argument('-c', '--config', dest='config_param', required=True,
2615 help='Set the vimconnector specific config parameters in dictionary format')
2616 mandatory_arguments.add_argument('-u', '--url', dest='endpoint_url',required=True, help="Set the vim connector url or Host IP")
2617 # Optional arguments
2618 vimconn_parser.add_argument('-i', '--image-path', dest='image_path', help="Provide image path present at RO container")
2619 vimconn_parser.add_argument('-n', '--image-name', dest='image_name', help="Provide image name for test")
2620 # TODO add optional arguments for vimconn tests
2621 # vimconn_parser.add_argument("-i", '--image-name', dest='image_name', help='<HELP>'))
2622 vimconn_parser.add_argument('-s', '--sriov-net-name', dest='sriov_net_name', help="Provide SRIOV network name for test")
2623
2624 # Datacenter test set
2625 # -------------------
2626 vimconn_parser = subparsers.add_parser('vim', parents=[parent_parser], help="test vim")
2627 vimconn_parser.set_defaults(func=test_vim)
2628
2629 # Mandatory arguments
2630 mandatory_arguments = vimconn_parser.add_argument_group('mandatory arguments')
2631 mandatory_arguments.add_argument('-d', '--datacenter', required=True, help='Set the datacenter to test')