test/test_RO.py

   1 #!/usr/bin/env python2
   2 # -*- coding: utf-8 -*-
   3
   4 ##
   5 # Copyright 2017
   6 # This file is part of openmano
   7 # All Rights Reserved.
   8 #
   9 # Licensed under the Apache License, Version 2.0 (the "License"); you may
  10 # not use this file except in compliance with the License. You may obtain
  11 # a copy of the License at
  12 #
  13 #         http://www.apache.org/licenses/LICENSE-2.0
  14 #
  15 # Unless required by applicable law or agreed to in writing, software
  16 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  17 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  18 # License for the specific language governing permissions and limitations
  19 # under the License.
  20 #
  21 ##
  22
  23 '''
  24 Module for testing openmano functionality. It uses openmanoclient.py for invoking openmano
  25 '''
  26 __author__="Pablo Montes, Alfonso Tierno"
  27 __date__ ="$16-Feb-2017 17:08:16$"
  28 __version__="0.0.3"
  29 version_date="May 2017"
  30
  31 import logging
  32 import os
  33 from argparse import ArgumentParser
  34 import argcomplete
  35 import unittest
  36 import string
  37 import inspect
  38 import random
  39 import traceback
  40 import glob
  41 import yaml
  42 import sys
  43 import time
  44 from pyvcloud.vcloudair import VCA
  45 import uuid
  46
  47 global test_config   #  used for global variables with the test configuration
  48 test_config = {}
  49
  50
  51 def check_instance_scenario_active(uuid):
  52     instance = test_config["client"].get_instance(uuid=uuid)
  53
  54     for net in instance['nets']:
  55         status = net['status']
  56         if status != 'ACTIVE':
  57             return (False, status)
  58
  59     for vnf in instance['vnfs']:
  60         for vm in vnf['vms']:
  61             status = vm['status']
  62             if status != 'ACTIVE':
  63                 return (False, status)
  64
  65     return (True, None)
  66
  67
  68 '''
  69 IMPORTANT NOTE
  70 All unittest classes for code based tests must have prefix 'test_' in order to be taken into account for tests
  71 '''
  72 class test_VIM_datacenter_tenant_operations(unittest.TestCase):
  73     test_index = 1
  74     tenant_name = None
  75     test_text = None
  76
  77     @classmethod
  78     def setUpClass(cls):
  79         logger.info("{}. {}".format(test_config["test_number"], cls.__name__))
  80
  81     @classmethod
  82     def tearDownClass(cls):
  83         test_config["test_number"] += 1
  84
  85     def tearDown(self):
  86         exec_info = sys.exc_info()
  87         if exec_info == (None, None, None):
  88             logger.info(self.__class__.test_text+" -> TEST OK")
  89         else:
  90             logger.warning(self.__class__.test_text+" -> TEST NOK")
  91             error_trace = traceback.format_exception(exec_info[0], exec_info[1], exec_info[2])
  92             msg = ""
  93             for line in error_trace:
  94                 msg = msg + line
  95             logger.critical("{}".format(msg))
  96
  97     def test_000_create_RO_tenant(self):
  98         self.__class__.tenant_name = _get_random_string(20)
  99         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
 100                                                            inspect.currentframe().f_code.co_name)
 101         self.__class__.test_index += 1
 102         tenant = test_config["client"].create_tenant(name=self.__class__.tenant_name,
 103                                                      description=self.__class__.tenant_name)
 104         logger.debug("{}".format(tenant))
 105         self.assertEqual(tenant.get('tenant', {}).get('name', ''), self.__class__.tenant_name)
 106
 107     def test_010_list_RO_tenant(self):
 108         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
 109                                                            inspect.currentframe().f_code.co_name)
 110         self.__class__.test_index += 1
 111         tenant = test_config["client"].get_tenant(name=self.__class__.tenant_name)
 112         logger.debug("{}".format(tenant))
 113         self.assertEqual(tenant.get('tenant', {}).get('name', ''), self.__class__.tenant_name)
 114
 115     def test_020_delete_RO_tenant(self):
 116         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
 117                                                            inspect.currentframe().f_code.co_name)
 118         self.__class__.test_index += 1
 119         tenant = test_config["client"].delete_tenant(name=self.__class__.tenant_name)
 120         logger.debug("{}".format(tenant))
 121         assert('deleted' in tenant.get('result',""))
 122
 123
 124 class test_VIM_datacenter_operations(unittest.TestCase):
 125     test_index = 1
 126     datacenter_name = None
 127     test_text = None
 128
 129     @classmethod
 130     def setUpClass(cls):
 131         logger.info("{}. {}".format(test_config["test_number"], cls.__name__))
 132
 133     @classmethod
 134     def tearDownClass(cls):
 135         test_config["test_number"] += 1
 136
 137     def tearDown(self):
 138         exec_info = sys.exc_info()
 139         if exec_info == (None, None, None):
 140             logger.info(self.__class__.test_text+" -> TEST OK")
 141         else:
 142             logger.warning(self.__class__.test_text+" -> TEST NOK")
 143             error_trace = traceback.format_exception(exec_info[0], exec_info[1], exec_info[2])
 144             msg = ""
 145             for line in error_trace:
 146                 msg = msg + line
 147             logger.critical("{}".format(msg))
 148
 149     def test_000_create_datacenter(self):
 150         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
 151                                                            inspect.currentframe().f_code.co_name)
 152         self.__class__.datacenter_name = _get_random_string(20)
 153         self.__class__.test_index += 1
 154         self.datacenter = test_config["client"].create_datacenter(name=self.__class__.datacenter_name,
 155                                                                   vim_url="http://fakeurl/fake")
 156         logger.debug("{}".format(self.datacenter))
 157         self.assertEqual (self.datacenter.get('datacenter', {}).get('name',''), self.__class__.datacenter_name)
 158
 159     def test_010_list_datacenter(self):
 160         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
 161                                                            inspect.currentframe().f_code.co_name)
 162
 163         self.__class__.test_index += 1
 164         self.datacenter = test_config["client"].get_datacenter(all_tenants=True, name=self.__class__.datacenter_name)
 165         logger.debug("{}".format(self.datacenter))
 166         self.assertEqual (self.datacenter.get('datacenter', {}).get('name', ''), self.__class__.datacenter_name)
 167
 168     def test_020_attach_datacenter(self):
 169         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
 170                                                            inspect.currentframe().f_code.co_name)
 171
 172         self.__class__.test_index += 1
 173         self.datacenter = test_config["client"].attach_datacenter(name=self.__class__.datacenter_name,
 174                                                                   vim_tenant_name='fake')
 175         logger.debug("{}".format(self.datacenter))
 176         assert ('vim_tenants' in self.datacenter.get('datacenter', {}))
 177
 178     def test_030_list_attached_datacenter(self):
 179         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
 180                                                            inspect.currentframe().f_code.co_name)
 181
 182         self.__class__.test_index += 1
 183         self.datacenter = test_config["client"].get_datacenter(all_tenants=False, name=self.__class__.datacenter_name)
 184         logger.debug("{}".format(self.datacenter))
 185         self.assertEqual (self.datacenter.get('datacenter', {}).get('name', ''), self.__class__.datacenter_name)
 186
 187     def test_040_detach_datacenter(self):
 188         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
 189                                                            inspect.currentframe().f_code.co_name)
 190
 191         self.__class__.test_index += 1
 192         self.datacenter = test_config["client"].detach_datacenter(name=self.__class__.datacenter_name)
 193         logger.debug("{}".format(self.datacenter))
 194         assert ('detached' in self.datacenter.get('result', ""))
 195
 196     def test_050_delete_datacenter(self):
 197         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
 198                                                            inspect.currentframe().f_code.co_name)
 199
 200         self.__class__.test_index += 1
 201         self.datacenter = test_config["client"].delete_datacenter(name=self.__class__.datacenter_name)
 202         logger.debug("{}".format(self.datacenter))
 203         assert('deleted' in self.datacenter.get('result',""))
 204
 205
 206 class test_VIM_network_operations(unittest.TestCase):
 207     test_index = 1
 208     vim_network_name = None
 209     test_text = None
 210     vim_network_uuid = None
 211
 212     @classmethod
 213     def setUpClass(cls):
 214         logger.info("{}. {}".format(test_config["test_number"], cls.__name__))
 215
 216     @classmethod
 217     def tearDownClass(cls):
 218         test_config["test_number"] += 1
 219
 220     def tearDown(self):
 221         exec_info = sys.exc_info()
 222         if exec_info == (None, None, None):
 223             logger.info(self.__class__.test_text + " -> TEST OK")
 224         else:
 225             logger.warning(self.__class__.test_text + " -> TEST NOK")
 226             error_trace = traceback.format_exception(exec_info[0], exec_info[1], exec_info[2])
 227             msg = ""
 228             for line in error_trace:
 229                 msg = msg + line
 230             logger.critical("{}".format(msg))
 231
 232     def test_000_create_VIM_network(self):
 233         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
 234                                                            inspect.currentframe().f_code.co_name)
 235         self.__class__.vim_network_name = _get_random_string(20)
 236         self.__class__.test_index += 1
 237         network = test_config["client"].vim_action("create", "networks", name=self.__class__.vim_network_name)
 238         logger.debug("{}".format(network))
 239         self.__class__.vim_network_uuid = network["network"]["id"]
 240         self.assertEqual(network.get('network', {}).get('name', ''), self.__class__.vim_network_name)
 241
 242     def test_010_list_VIM_networks(self):
 243         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
 244                                                            inspect.currentframe().f_code.co_name)
 245         self.__class__.test_index += 1
 246         networks = test_config["client"].vim_action("list", "networks")
 247         logger.debug("{}".format(networks))
 248
 249     def test_020_get_VIM_network_by_uuid(self):
 250         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
 251                                                            inspect.currentframe().f_code.co_name)
 252
 253         self.__class__.test_index += 1
 254         network = test_config["client"].vim_action("show", "networks", uuid=self.__class__.vim_network_uuid)
 255         logger.debug("{}".format(network))
 256         self.assertEqual(network.get('network', {}).get('name', ''), self.__class__.vim_network_name)
 257
 258     def test_030_delete_VIM_network_by_uuid(self):
 259         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
 260                                                            inspect.currentframe().f_code.co_name)
 261
 262         self.__class__.test_index += 1
 263         network = test_config["client"].vim_action("delete", "networks", uuid=self.__class__.vim_network_uuid)
 264         logger.debug("{}".format(network))
 265         assert ('deleted' in network.get('result', ""))
 266
 267
 268 class test_VIM_image_operations(unittest.TestCase):
 269     test_index = 1
 270     test_text = None
 271
 272     @classmethod
 273     def setUpClass(cls):
 274         logger.info("{}. {}".format(test_config["test_number"], cls.__name__))
 275
 276     @classmethod
 277     def tearDownClass(cls):
 278         test_config["test_number"] += 1
 279
 280     def tearDown(self):
 281         exec_info = sys.exc_info()
 282         if exec_info == (None, None, None):
 283             logger.info(self.__class__.test_text + " -> TEST OK")
 284         else:
 285             logger.warning(self.__class__.test_text + " -> TEST NOK")
 286             error_trace = traceback.format_exception(exec_info[0], exec_info[1], exec_info[2])
 287             msg = ""
 288             for line in error_trace:
 289                 msg = msg + line
 290             logger.critical("{}".format(msg))
 291
 292     def test_000_list_VIM_images(self):
 293         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
 294                                                            inspect.currentframe().f_code.co_name)
 295         self.__class__.test_index += 1
 296         images = test_config["client"].vim_action("list", "images")
 297         logger.debug("{}".format(images))
 298
 299 '''
 300 The following is a non critical test that will fail most of the times.
 301 In case of OpenStack datacenter these tests will only success if RO has access to the admin endpoint
 302 This test will only be executed in case it is specifically requested by the user
 303 '''
 304 class test_VIM_tenant_operations(unittest.TestCase):
 305     test_index = 1
 306     vim_tenant_name = None
 307     test_text = None
 308     vim_tenant_uuid = None
 309
 310     @classmethod
 311     def setUpClass(cls):
 312         logger.info("{}. {}".format(test_config["test_number"], cls.__name__))
 313         logger.warning("In case of OpenStack datacenter these tests will only success "
 314                        "if RO has access to the admin endpoint")
 315
 316     @classmethod
 317     def tearDownClass(cls):
 318         test_config["test_number"] += 1
 319
 320     def tearDown(self):
 321         exec_info = sys.exc_info()
 322         if exec_info == (None, None, None):
 323             logger.info(self.__class__.test_text + " -> TEST OK")
 324         else:
 325             logger.warning(self.__class__.test_text + " -> TEST NOK")
 326             error_trace = traceback.format_exception(exec_info[0], exec_info[1], exec_info[2])
 327             msg = ""
 328             for line in error_trace:
 329                 msg = msg + line
 330             logger.critical("{}".format(msg))
 331
 332     def test_000_create_VIM_tenant(self):
 333         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
 334                                                            inspect.currentframe().f_code.co_name)
 335         self.__class__.vim_tenant_name = _get_random_string(20)
 336         self.__class__.test_index += 1
 337         tenant = test_config["client"].vim_action("create", "tenants", name=self.__class__.vim_tenant_name)
 338         logger.debug("{}".format(tenant))
 339         self.__class__.vim_tenant_uuid = tenant["tenant"]["id"]
 340         self.assertEqual(tenant.get('tenant', {}).get('name', ''), self.__class__.vim_tenant_name)
 341
 342     def test_010_list_VIM_tenants(self):
 343         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
 344                                                            inspect.currentframe().f_code.co_name)
 345         self.__class__.test_index += 1
 346         tenants = test_config["client"].vim_action("list", "tenants")
 347         logger.debug("{}".format(tenants))
 348
 349     def test_020_get_VIM_tenant_by_uuid(self):
 350         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
 351                                                            inspect.currentframe().f_code.co_name)
 352
 353         self.__class__.test_index += 1
 354         tenant = test_config["client"].vim_action("show", "tenants", uuid=self.__class__.vim_tenant_uuid)
 355         logger.debug("{}".format(tenant))
 356         self.assertEqual(tenant.get('tenant', {}).get('name', ''), self.__class__.vim_tenant_name)
 357
 358     def test_030_delete_VIM_tenant_by_uuid(self):
 359         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"], self.__class__.test_index,
 360                                                            inspect.currentframe().f_code.co_name)
 361
 362         self.__class__.test_index += 1
 363         tenant = test_config["client"].vim_action("delete", "tenants", uuid=self.__class__.vim_tenant_uuid)
 364         logger.debug("{}".format(tenant))
 365         assert ('deleted' in tenant.get('result', ""))
 366
 367 class test_vimconn_connect(unittest.TestCase):
 368     test_index = 1
 369     test_text = None
 370
 371     @classmethod
 372     def setUpClass(cls):
 373         logger.info("{}. {}".format(test_config["test_number"], cls.__name__))
 374
 375     @classmethod
 376     def tearDownClass(cls):
 377         test_config["test_number"] += 1
 378
 379     def tearDown(self):
 380         exec_info = sys.exc_info()
 381         if exec_info == (None, None, None):
 382             logger.info(self.__class__.test_text+" -> TEST OK")
 383         else:
 384             logger.warning(self.__class__.test_text+" -> TEST NOK")
 385             logger.critical("Traceback error",exc_info=True)
 386
 387     def test_000_connect(self):
 388         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
 389                                                             self.__class__.test_index,
 390                                                 inspect.currentframe().f_code.co_name)
 391
 392         self.__class__.test_index += 1
 393         if test_config['vimtype'] == 'vmware':
 394             vca_object = test_config["vim_conn"].connect()
 395             logger.debug("{}".format(vca_object))
 396             self.assertIsInstance(vca_object, VCA)
 397
 398
 399 class test_vimconn_new_network(unittest.TestCase):
 400     test_index = 1
 401     network_name = None
 402     test_text = None
 403
 404     @classmethod
 405     def setUpClass(cls):
 406         logger.info("{}. {}".format(test_config["test_number"], cls.__name__))
 407
 408     @classmethod
 409     def tearDownClass(cls):
 410         test_config["test_number"] += 1
 411
 412     def tearDown(self):
 413         exec_info = sys.exc_info()
 414         if exec_info == (None, None, None):
 415             logger.info(self.__class__.test_text+" -> TEST OK")
 416         else:
 417             logger.warning(self.__class__.test_text+" -> TEST NOK")
 418             logger.critical("Traceback error",exc_info=True)
 419
 420     def test_000_new_network(self):
 421         self.__class__.network_name = _get_random_string(20)
 422         network_type = 'bridge'
 423
 424         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
 425                      self.__class__.test_index, inspect.currentframe().f_code.co_name)
 426         self.__class__.test_index += 1
 427
 428         network = test_config["vim_conn"].new_network(net_name=self.__class__.network_name,
 429                                                           net_type=network_type)
 430         self.__class__.network_id = network
 431         logger.debug("{}".format(network))
 432
 433         network_list = test_config["vim_conn"].get_vcd_network_list()
 434         for net in network_list:
 435             if self.__class__.network_name in net.get('name'):
 436                 self.assertIn(self.__class__.network_name, net.get('name'))
 437                 self.assertEqual(net.get('type'), network_type)
 438
 439         # Deleting created network
 440         result = test_config["vim_conn"].delete_network(self.__class__.network_id)
 441         if result:
 442             logger.info("Network id {} sucessfully deleted".format(self.__class__.network_id))
 443         else:
 444             logger.info("Failed to delete network id {}".format(self.__class__.network_id))
 445
 446     def test_010_new_network_by_types(self):
 447         delete_net_ids = []
 448         network_types = ['data','bridge','mgmt']
 449         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
 450                                                             self.__class__.test_index,
 451                                                 inspect.currentframe().f_code.co_name)
 452         self.__class__.test_index += 1
 453         for net_type in network_types:
 454             self.__class__.network_name = _get_random_string(20)
 455             network_id = test_config["vim_conn"].new_network(net_name=self.__class__.network_name,
 456                                                                                 net_type=net_type)
 457
 458             delete_net_ids.append(network_id)
 459             logger.debug("{}".format(network_id))
 460
 461             network_list = test_config["vim_conn"].get_vcd_network_list()
 462             for net in network_list:
 463                 if self.__class__.network_name in net.get('name'):
 464                     self.assertIn(self.__class__.network_name, net.get('name'))
 465                 if net_type in net.get('type'):
 466                     self.assertEqual(net.get('type'), net_type)
 467                 else:
 468                     self.assertNotEqual(net.get('type'), net_type)
 469
 470         # Deleting created network
 471         for net_id in delete_net_ids:
 472             result = test_config["vim_conn"].delete_network(net_id)
 473             if result:
 474                 logger.info("Network id {} sucessfully deleted".format(net_id))
 475             else:
 476                 logger.info("Failed to delete network id {}".format(net_id))
 477
 478     def test_020_new_network_by_ipprofile(self):
 479         test_directory_content = os.listdir(test_config["test_directory"])
 480
 481         for dir_name in test_directory_content:
 482             if dir_name == 'simple_multi_vnfc':
 483                 self.__class__.scenario_test_path = test_config["test_directory"] + '/'+ dir_name
 484                 vnfd_files = glob.glob(self.__class__.scenario_test_path+'/vnfd_*.yaml')
 485                 break
 486
 487         for vnfd in vnfd_files:
 488             with open(vnfd, 'r') as stream:
 489                 vnf_descriptor = yaml.load(stream)
 490
 491             internal_connections_list = vnf_descriptor['vnf']['internal-connections']
 492             for item in internal_connections_list:
 493                 if 'ip-profile' in item:
 494                     version = item['ip-profile']['ip-version']
 495                     dhcp_count = item['ip-profile']['dhcp']['count']
 496                     dhcp_enabled = item['ip-profile']['dhcp']['enabled']
 497
 498         self.__class__.network_name = _get_random_string(20)
 499         ip_profile = {'dhcp_count': dhcp_count,
 500                       'dhcp_enabled': dhcp_enabled,
 501                       'ip_version': version
 502                      }
 503         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
 504                                                             self.__class__.test_index,
 505                                                 inspect.currentframe().f_code.co_name)
 506         self.__class__.test_index += 1
 507         network = test_config["vim_conn"].new_network(net_name=self.__class__.network_name,
 508                                                                            net_type='mgmt',
 509                                                                      ip_profile=ip_profile)
 510         self.__class__.network_id = network
 511         logger.debug("{}".format(network))
 512
 513         network_list = test_config["vim_conn"].get_vcd_network_list()
 514         for net in network_list:
 515             if self.__class__.network_name in net.get('name'):
 516                 self.assertIn(self.__class__.network_name, net.get('name'))
 517
 518         # Deleting created network
 519         result = test_config["vim_conn"].delete_network(self.__class__.network_id)
 520         if result:
 521             logger.info("Network id {} sucessfully deleted".format(self.__class__.network_id))
 522         else:
 523             logger.info("Failed to delete network id {}".format(self.__class__.network_id))
 524
 525     def test_030_new_network_by_isshared(self):
 526         self.__class__.network_name = _get_random_string(20)
 527         shared = True
 528         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
 529                                                             self.__class__.test_index,
 530                                                 inspect.currentframe().f_code.co_name)
 531         self.__class__.test_index += 1
 532         network = test_config["vim_conn"].new_network(net_name=self.__class__.network_name,
 533                                                                          net_type='bridge',
 534                                                                              shared=shared)
 535         self.__class__.network_id = network
 536         logger.debug("{}".format(network))
 537
 538         network_list = test_config["vim_conn"].get_vcd_network_list()
 539         for net in network_list:
 540             if self.__class__.network_name in net.get('name'):
 541                 self.assertIn(self.__class__.network_name, net.get('name'))
 542                 self.assertEqual(net.get('shared'), shared)
 543
 544         # Deleting created network
 545         result = test_config["vim_conn"].delete_network(self.__class__.network_id)
 546         if result:
 547             logger.info("Network id {} sucessfully deleted".format(self.__class__.network_id))
 548         else:
 549             logger.info("Failed to delete network id {}".format(self.__class__.network_id))
 550
 551     def test_040_new_network_by_negative(self):
 552         self.__class__.network_name = _get_random_string(20)
 553         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
 554                                                             self.__class__.test_index,
 555                                                 inspect.currentframe().f_code.co_name)
 556         self.__class__.test_index += 1
 557         network = test_config["vim_conn"].new_network(net_name=self.__class__.network_name,
 558                                                                     net_type='unknowntype')
 559         self.__class__.network_id = network
 560         logger.debug("{}".format(network))
 561         network_list = test_config["vim_conn"].get_vcd_network_list()
 562         for net in network_list:
 563             if self.__class__.network_name in net.get('name'):
 564                 self.assertIn(self.__class__.network_name, net.get('name'))
 565
 566         # Deleting created network
 567         result = test_config["vim_conn"].delete_network(self.__class__.network_id)
 568         if result:
 569             logger.info("Network id {} sucessfully deleted".format(self.__class__.network_id))
 570         else:
 571             logger.info("Failed to delete network id {}".format(self.__class__.network_id))
 572
 573 class test_vimconn_get_network_list(unittest.TestCase):
 574     test_index = 1
 575     network_name = None
 576     test_text = None
 577
 578     @classmethod
 579     def setUpClass(cls):
 580         logger.info("{}. {}".format(test_config["test_number"], cls.__name__))
 581
 582     @classmethod
 583     def tearDownClass(cls):
 584         test_config["test_number"] += 1
 585
 586     def setUp(self):
 587         # creating new network
 588         self.__class__.network_name = _get_random_string(20)
 589         self.__class__.net_type = 'bridge'
 590         network = test_config["vim_conn"].new_network(net_name=self.__class__.network_name,
 591                                                           net_type=self.__class__.net_type)
 592         self.__class__.network_id = network
 593         logger.debug("{}".format(network))
 594
 595     def tearDown(self):
 596         exec_info = sys.exc_info()
 597         if exec_info == (None, None, None):
 598             logger.info(self.__class__.test_text+" -> TEST OK")
 599         else:
 600             logger.warning(self.__class__.test_text+" -> TEST NOK")
 601             logger.critical("Traceback error",exc_info=True)
 602
 603         # Deleting created network
 604         result = test_config["vim_conn"].delete_network(self.__class__.network_id)
 605         if result:
 606             logger.info("Network id {} sucessfully deleted".format(self.__class__.network_id))
 607         else:
 608             logger.info("Failed to delete network id {}".format(self.__class__.network_id))
 609
 610     def test_000_get_network_list(self):
 611         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
 612                                                             self.__class__.test_index,
 613                                                 inspect.currentframe().f_code.co_name)
 614         self.__class__.test_index += 1
 615
 616         network_list = test_config["vim_conn"].get_network_list()
 617         for net in network_list:
 618             if self.__class__.network_name in net.get('name'):
 619                 self.assertIn(self.__class__.network_name, net.get('name'))
 620                 self.assertEqual(net.get('type'), self.__class__.net_type)
 621                 self.assertEqual(net.get('status'), 'ACTIVE')
 622                 self.assertEqual(net.get('shared'), False)
 623
 624     def test_010_get_network_list_by_name(self):
 625         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
 626                                                             self.__class__.test_index,
 627                                                 inspect.currentframe().f_code.co_name)
 628         self.__class__.test_index += 1
 629
 630         network_name = test_config['vim_conn'].get_network_name_by_id(self.__class__.network_id)
 631
 632         # find network from list by it's name
 633         new_network_list = test_config["vim_conn"].get_network_list({'name': network_name})
 634         for list_item in new_network_list:
 635             if self.__class__.network_name in list_item.get('name'):
 636                 self.assertEqual(network_name, list_item.get('name'))
 637                 self.assertEqual(list_item.get('type'), self.__class__.net_type)
 638                 self.assertEqual(list_item.get('status'), 'ACTIVE')
 639
 640     def test_020_get_network_list_by_id(self):
 641         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
 642                                                             self.__class__.test_index,
 643                                                 inspect.currentframe().f_code.co_name)
 644         self.__class__.test_index += 1
 645
 646         # find network from list by it's id
 647         new_network_list = test_config["vim_conn"].get_network_list({'id':self.__class__.network_id})
 648         for list_item in new_network_list:
 649             if self.__class__.network_id in list_item.get('id'):
 650                 self.assertEqual(self.__class__.network_id, list_item.get('id'))
 651                 self.assertEqual(list_item.get('type'), self.__class__.net_type)
 652                 self.assertEqual(list_item.get('status'), 'ACTIVE')
 653
 654     def test_030_get_network_list_by_shared(self):
 655         Shared = False
 656         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
 657                                                             self.__class__.test_index,
 658                                                 inspect.currentframe().f_code.co_name)
 659         self.__class__.test_index += 1
 660
 661         network_name = test_config['vim_conn'].get_network_name_by_id(self.__class__.network_id)
 662         # find network from list by it's shared value
 663         new_network_list = test_config["vim_conn"].get_network_list({'shared':Shared,
 664                                                                 'name':network_name})
 665         for list_item in new_network_list:
 666             if list_item.get('shared') == Shared:
 667                 self.assertEqual(list_item.get('shared'), Shared)
 668                 self.assertEqual(list_item.get('type'), self.__class__.net_type)
 669                 self.assertEqual(network_name, list_item.get('name'))
 670
 671     def test_040_get_network_list_by_tenant_id(self):
 672         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
 673                                                             self.__class__.test_index,
 674                                                 inspect.currentframe().f_code.co_name)
 675         self.__class__.test_index += 1
 676
 677         tenant_list = test_config["vim_conn"].get_tenant_list()
 678         network_name = test_config['vim_conn'].get_network_name_by_id(self.__class__.network_id)
 679
 680         for tenant_item in tenant_list:
 681             if test_config['tenant'] == tenant_item.get('name'):
 682                 # find network from list by it's tenant id
 683                 tenant_id = tenant_item.get('id')
 684                 new_network_list = test_config["vim_conn"].get_network_list({'tenant_id':tenant_id,
 685                                                                               'name':network_name})
 686                 for list_item in new_network_list:
 687                     self.assertEqual(tenant_id, list_item.get('tenant_id'))
 688                     self.assertEqual(network_name, list_item.get('name'))
 689                     self.assertEqual(list_item.get('type'), self.__class__.net_type)
 690                     self.assertEqual(list_item.get('status'), 'ACTIVE')
 691
 692     def test_050_get_network_list_by_status(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         status = 'ACTIVE'
 698
 699         network_name = test_config['vim_conn'].get_network_name_by_id(self.__class__.network_id)
 700
 701         # find network from list by it's status
 702         new_network_list = test_config["vim_conn"].get_network_list({'status':status,
 703                                                                'name': network_name})
 704         for list_item in new_network_list:
 705             self.assertIn(self.__class__.network_name, list_item.get('name'))
 706             self.assertEqual(list_item.get('type'), self.__class__.net_type)
 707             self.assertEqual(list_item.get('status'), status)
 708
 709     def test_060_get_network_list_by_negative(self):
 710         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
 711                                                             self.__class__.test_index,
 712                                                 inspect.currentframe().f_code.co_name)
 713         self.__class__.test_index += 1
 714
 715         network_list = test_config["vim_conn"].get_network_list({'name': 'unknown_name'})
 716         self.assertEqual(network_list, [])
 717
 718 class test_vimconn_get_network(unittest.TestCase):
 719     test_index = 1
 720     network_name = None
 721     test_text = None
 722
 723     @classmethod
 724     def setUpClass(cls):
 725         logger.info("{}. {}".format(test_config["test_number"], cls.__name__))
 726
 727     @classmethod
 728     def tearDownClass(cls):
 729         test_config["test_number"] += 1
 730
 731     def setUp(self):
 732         # creating new network
 733         self.__class__.network_name = _get_random_string(20)
 734         self.__class__.net_type = 'bridge'
 735         network = test_config["vim_conn"].new_network(net_name=self.__class__.network_name,
 736                                                           net_type=self.__class__.net_type)
 737         self.__class__.network_id = network
 738         logger.debug("{}".format(network))
 739
 740     def tearDown(self):
 741         exec_info = sys.exc_info()
 742         if exec_info == (None, None, None):
 743             logger.info(self.__class__.test_text+" -> TEST OK")
 744         else:
 745             logger.warning(self.__class__.test_text+" -> TEST NOK")
 746             logger.critical("Traceback error",exc_info=True)
 747
 748         # Deleting created network
 749         result = test_config["vim_conn"].delete_network(self.__class__.network_id)
 750         if result:
 751             logger.info("Network id {} sucessfully deleted".format(self.__class__.network_id))
 752         else:
 753             logger.info("Failed to delete network id {}".format(self.__class__.network_id))
 754
 755     def test_000_get_network(self):
 756         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
 757                                                             self.__class__.test_index,
 758                                                 inspect.currentframe().f_code.co_name)
 759         self.__class__.test_index += 1
 760
 761         network_info = test_config["vim_conn"].get_network(self.__class__.network_id)
 762         self.assertEqual(network_info.get('status'), 'ACTIVE')
 763         self.assertIn(self.__class__.network_name, network_info.get('name'))
 764         self.assertEqual(network_info.get('type'), self.__class__.net_type)
 765         self.assertEqual(network_info.get('id'), self.__class__.network_id)
 766
 767     def test_010_get_network_negative(self):
 768         Non_exist_id = str(uuid.uuid4())
 769         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
 770                                                             self.__class__.test_index,
 771                                                 inspect.currentframe().f_code.co_name)
 772         self.__class__.test_index += 1
 773
 774         network_info = test_config["vim_conn"].get_network(Non_exist_id)
 775         self.assertEqual(network_info, {})
 776
 777 class test_vimconn_delete_network(unittest.TestCase):
 778     test_index = 1
 779     network_name = None
 780     test_text = None
 781
 782     @classmethod
 783     def setUpClass(cls):
 784         logger.info("{}. {}".format(test_config["test_number"], cls.__name__))
 785
 786     @classmethod
 787     def tearDownClass(cls):
 788         test_config["test_number"] += 1
 789
 790     def tearDown(self):
 791         exec_info = sys.exc_info()
 792         if exec_info == (None, None, None):
 793             logger.info(self.__class__.test_text+" -> TEST OK")
 794         else:
 795             logger.warning(self.__class__.test_text+" -> TEST NOK")
 796             logger.critical("Traceback error",exc_info=True)
 797
 798     def test_000_delete_network(self):
 799         # Creating network
 800         self.__class__.network_name = _get_random_string(20)
 801         self.__class__.net_type = 'bridge'
 802         network = test_config["vim_conn"].new_network(net_name=self.__class__.network_name,
 803                                                           net_type=self.__class__.net_type)
 804         self.__class__.network_id = network
 805         logger.debug("{}".format(network))
 806
 807         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
 808                                                             self.__class__.test_index,
 809                                                 inspect.currentframe().f_code.co_name)
 810         self.__class__.test_index += 1
 811
 812         result = test_config["vim_conn"].delete_network(self.__class__.network_id)
 813         if result:
 814             logger.info("Network id {} sucessfully deleted".format(self.__class__.network_id))
 815         else:
 816             logger.info("Failed to delete network id {}".format(self.__class__.network_id))
 817         time.sleep(5)
 818         # after deleting network we check in network list
 819         network_list = test_config["vim_conn"].get_network_list({ 'id':self.__class__.network_id })
 820         self.assertEqual(network_list, [])
 821
 822     def test_010_delete_network_negative(self):
 823         Non_exist_id = str(uuid.uuid4())
 824
 825         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
 826                                                             self.__class__.test_index,
 827                                                 inspect.currentframe().f_code.co_name)
 828         self.__class__.test_index += 1
 829
 830         with self.assertRaises(Exception) as context:
 831             test_config["vim_conn"].delete_network(Non_exist_id)
 832
 833         self.assertEqual((context.exception).http_code, 400)
 834
 835 class test_vimconn_get_flavor(unittest.TestCase):
 836     test_index = 1
 837     test_text = None
 838
 839     @classmethod
 840     def setUpClass(cls):
 841         logger.info("{}. {}".format(test_config["test_number"], cls.__name__))
 842
 843     @classmethod
 844     def tearDownClass(cls):
 845         test_config["test_number"] += 1
 846
 847     def tearDown(self):
 848         exec_info = sys.exc_info()
 849         if exec_info == (None, None, None):
 850             logger.info(self.__class__.test_text+" -> TEST OK")
 851         else:
 852             logger.warning(self.__class__.test_text+" -> TEST NOK")
 853             logger.critical("Traceback error",exc_info=True)
 854
 855     def test_000_get_flavor(self):
 856         test_directory_content = os.listdir(test_config["test_directory"])
 857
 858         for dir_name in test_directory_content:
 859             if dir_name == 'simple_linux':
 860                 self.__class__.scenario_test_path = test_config["test_directory"] + '/'+ dir_name
 861                 vnfd_files = glob.glob(self.__class__.scenario_test_path+'/vnfd_*.yaml')
 862                 break
 863
 864         for vnfd in vnfd_files:
 865             with open(vnfd, 'r') as stream:
 866                 vnf_descriptor = yaml.load(stream)
 867
 868             vnfc_list = vnf_descriptor['vnf']['VNFC']
 869             for item in vnfc_list:
 870                 if 'ram' in item and 'vcpus' in item and 'disk' in item:
 871                     ram = item['ram']
 872                     vcpus = item['vcpus']
 873                     disk = item['disk']
 874
 875         flavor_data = {'ram': ram,
 876                       'vcpus': vcpus,
 877                       'disk': disk
 878                       }
 879
 880         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
 881                                                             self.__class__.test_index,
 882                                                 inspect.currentframe().f_code.co_name)
 883         self.__class__.test_index += 1
 884         # create new flavor
 885         flavor_id = test_config["vim_conn"].new_flavor(flavor_data)
 886         # get flavor by id
 887         result = test_config["vim_conn"].get_flavor(flavor_id)
 888         self.assertEqual(ram, result['ram'])
 889         self.assertEqual(vcpus, result['vcpus'])
 890         self.assertEqual(disk, result['disk'])
 891
 892         # delete flavor
 893         result = test_config["vim_conn"].delete_flavor(flavor_id)
 894         if result:
 895             logger.info("Flavor id {} sucessfully deleted".format(result))
 896         else:
 897             logger.info("Failed to delete flavor id {}".format(result))
 898
 899     def test_010_get_flavor_negative(self):
 900         Non_exist_flavor_id = str(uuid.uuid4())
 901
 902         self.__class__.test_text = "{}.{}. TEST {}".format(test_config["test_number"],
 903                                                             self.__class__.test_index,
 904                                                 inspect.currentframe().f_code.co_name)
 905         self.__class__.test_index += 1
 906
 907         with self.assertRaises(Exception) as context:
 908             test_config["vim_conn"].get_flavor(Non_exist_flavor_id)
 909
 910         self.assertEqual((context.exception).http_code, 404)
 911
 912
 913 '''
 914 IMPORTANT NOTE
 915 The following unittest class does not have the 'test_' on purpose. This test is the one used for the
 916 scenario based tests.
 917 '''
 918 class descriptor_based_scenario_test(unittest.TestCase):
 919     test_index = 0
 920     test_text = None
 921     scenario_test_path = None
 922     scenario_uuid = None
 923     instance_scenario_uuid = None
 924     to_delete_list = []
 925
 926     @classmethod
 927     def setUpClass(cls):
 928         cls.test_index = 1
 929         cls.to_delete_list = []
 930         cls.scenario_test_path = test_config["test_directory"] + '/' + test_config["test_folder"]
 931         logger.info("{}. {} {}".format(test_config["test_number"], cls.__name__, test_config["test_folder"]))
 932
 933     @classmethod
 934     def tearDownClass(cls):
 935          test_config["test_number"] += 1
 936
 937     def tearDown(self):
 938         exec_info = sys.exc_info()
 939         if exec_info == (None, None, None):
 940             logger.info(self.__class__.test_text + " -> TEST OK")
 941         else:
 942             logger.warning(self.__class__.test_text + " -> TEST NOK")
 943             error_trace = traceback.format_exception(exec_info[0], exec_info[1], exec_info[2])
 944             msg = ""
 945             for line in error_trace:
 946                 msg = msg + line
 947             logger.critical("{}".format(msg))
 948
 949
 950     def test_000_load_scenario(self):
 951         self.__class__.test_text = "{}.{}. TEST {} {}".format(test_config["test_number"], self.__class__.test_index,
 952                                                            inspect.currentframe().f_code.co_name,
 953                                                            test_config["test_folder"])
 954         self.__class__.test_index += 1
 955         vnfd_files = glob.glob(self.__class__.scenario_test_path+'/vnfd_*.yaml')
 956         scenario_file = glob.glob(self.__class__.scenario_test_path + '/scenario_*.yaml')
 957         if len(vnfd_files) == 0 or len(scenario_file) > 1:
 958             raise Exception("Test '{}' not valid. It must contain an scenario file and at least one vnfd file'".format(
 959                 test_config["test_folder"]))
 960
 961         #load all vnfd
 962         for vnfd in vnfd_files:
 963             with open(vnfd, 'r') as stream:
 964                 vnf_descriptor = yaml.load(stream)
 965
 966             vnfc_list = vnf_descriptor['vnf']['VNFC']
 967             for vnfc in vnfc_list:
 968                 vnfc['image name'] = test_config["image_name"]
 969                 devices = vnfc.get('devices',[])
 970                 for device in devices:
 971                     if device['type'] == 'disk' and 'image name' in device:
 972                         device['image name'] = test_config["image_name"]
 973
 974             logger.debug("VNF descriptor: {}".format(vnf_descriptor))
 975             vnf = test_config["client"].create_vnf(descriptor=vnf_descriptor)
 976             logger.debug(vnf)
 977             self.__class__.to_delete_list.insert(0, {"item": "vnf", "function": test_config["client"].delete_vnf,
 978                                                      "params": {"uuid": vnf['vnf']['uuid']}})
 979
 980         #load the scenario definition
 981         with open(scenario_file[0], 'r') as stream:
 982             scenario_descriptor = yaml.load(stream)
 983         networks = scenario_descriptor['scenario']['networks']
 984         networks[test_config["mgmt_net"]] = networks.pop('mgmt')
 985         logger.debug("Scenario descriptor: {}".format(scenario_descriptor))
 986         scenario = test_config["client"].create_scenario(descriptor=scenario_descriptor)
 987         logger.debug(scenario)
 988         self.__class__.to_delete_list.insert(0,{"item": "scenario", "function": test_config["client"].delete_scenario,
 989                                  "params":{"uuid": scenario['scenario']['uuid']} })
 990         self.__class__.scenario_uuid = scenario['scenario']['uuid']
 991
 992     def test_010_instantiate_scenario(self):
 993         self.__class__.test_text = "{}.{}. TEST {} {}".format(test_config["test_number"], self.__class__.test_index,
 994                                                            inspect.currentframe().f_code.co_name,
 995                                                            test_config["test_folder"])
 996         self.__class__.test_index += 1
 997
 998         instance = test_config["client"].create_instance(scenario_id=self.__class__.scenario_uuid,
 999                                                          name=self.__class__.test_text)
1000         self.__class__.instance_scenario_uuid = instance['uuid']
1001         logger.debug(instance)
1002         self.__class__.to_delete_list.insert(0, {"item": "instance", "function": test_config["client"].delete_instance,
1003                                   "params": {"uuid": instance['uuid']}})
1004
1005     def test_020_check_deployent(self):
1006         self.__class__.test_text = "{}.{}. TEST {} {}".format(test_config["test_number"], self.__class__.test_index,
1007                                                            inspect.currentframe().f_code.co_name,
1008                                                            test_config["test_folder"])
1009         self.__class__.test_index += 1
1010
1011         if test_config["manual"]:
1012             raw_input('Scenario has been deployed. Perform manual check and press any key to resume')
1013             return
1014
1015         keep_waiting = test_config["timeout"]
1016         instance_active = False
1017         while True:
1018             result = check_instance_scenario_active(self.__class__.instance_scenario_uuid)
1019             if result[0]:
1020                 break
1021             elif 'ERROR' in result[1]:
1022                 msg = 'Got error while waiting for the instance to get active: '+result[1]
1023                 logging.error(msg)
1024                 raise Exception(msg)
1025
1026             if keep_waiting >= 5:
1027                 time.sleep(5)
1028                 keep_waiting -= 5
1029             elif keep_waiting > 0:
1030                 time.sleep(keep_waiting)
1031                 keep_waiting = 0
1032             else:
1033                 msg = 'Timeout reached while waiting instance scenario to get active'
1034                 logging.error(msg)
1035                 raise Exception(msg)
1036
1037     def test_030_clean_deployment(self):
1038         self.__class__.test_text = "{}.{}. TEST {} {}".format(test_config["test_number"], self.__class__.test_index,
1039                                                               inspect.currentframe().f_code.co_name,
1040                                                               test_config["test_folder"])
1041         self.__class__.test_index += 1
1042         #At the moment if you delete an scenario right after creating it, in openstack datacenters
1043         #sometimes scenario ports get orphaned. This sleep is just a dirty workaround
1044         time.sleep(5)
1045         for item in self.__class__.to_delete_list:
1046             response = item["function"](**item["params"])
1047             logger.debug(response)
1048
1049
1050 def _get_random_string(maxLength):
1051     '''generates a string with random characters string.letters and string.digits
1052     with a random length up to maxLength characters. If maxLength is <15 it will be changed automatically to 15
1053     '''
1054     prefix = 'testing_'
1055     min_string = 15
1056     minLength = min_string - len(prefix)
1057     if maxLength < min_string: maxLength = min_string
1058     maxLength -= len(prefix)
1059     length = random.randint(minLength,maxLength)
1060     return 'testing_'+"".join([random.choice(string.letters+string.digits) for i in xrange(length)])
1061
1062
1063 def test_vimconnector(args):
1064     global test_config
1065     sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + "/osm_ro")
1066     test_config['vimtype'] = args.vimtype
1067     if args.vimtype == "vmware":
1068         import vimconn_vmware as vim
1069
1070         test_config["test_directory"] = os.path.dirname(__file__) + "/RO_tests"
1071
1072         tenant_name = args.tenant_name
1073         test_config['tenant'] = tenant_name
1074         config_params = json.loads(args.config_param)
1075         org_name = config_params.get('orgname')
1076         org_user = config_params.get('user')
1077         org_passwd = config_params.get('passwd')
1078         vim_url = args.endpoint_url
1079
1080         # vmware connector obj
1081         test_config['vim_conn'] = vim.vimconnector(name=org_name, tenant_name=tenant_name, user=org_user,passwd=org_passwd, url=vim_url, config=config_params)
1082
1083     elif args.vimtype == "aws":
1084         import vimconn_aws as vim
1085     elif args.vimtype == "openstack":
1086         import vimconn_openstack as vim
1087     elif args.vimtype == "openvim":
1088         import vimconn_openvim as vim
1089     else:
1090         logger.critical("vimtype '{}' not supported".format(args.vimtype))
1091         sys.exit(1)
1092     executed = 0
1093     failed = 0
1094     clsmembers = inspect.getmembers(sys.modules[__name__], inspect.isclass)
1095     # If only want to obtain a tests list print it and exit
1096     if args.list_tests:
1097         tests_names = []
1098         for cls in clsmembers:
1099             if cls[0].startswith('test_vimconnector'):
1100                 tests_names.append(cls[0])
1101
1102         msg = "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names))
1103         print(msg)
1104         logger.info(msg)
1105         sys.exit(0)
1106
1107     # Create the list of tests to be run
1108     code_based_tests = []
1109     if args.tests:
1110         for test in args.tests:
1111             for t in test.split(','):
1112                 matches_code_based_tests = [item for item in clsmembers if item[0] == t]
1113                 if len(matches_code_based_tests) > 0:
1114                     code_based_tests.append(matches_code_based_tests[0][1])
1115                 else:
1116                     logger.critical("Test '{}' is not among the possible ones".format(t))
1117                     sys.exit(1)
1118     if not code_based_tests:
1119         # include all tests
1120         for cls in clsmembers:
1121             # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
1122             if cls[0].startswith('test_vimconnector'):
1123                 code_based_tests.append(cls[1])
1124
1125     logger.debug("tests to be executed: {}".format(code_based_tests))
1126
1127     # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1128     # This is handled in the tests using logging.
1129     stream = open('/dev/null', 'w')
1130
1131     # Run code based tests
1132     basic_tests_suite = unittest.TestSuite()
1133     for test in code_based_tests:
1134         basic_tests_suite.addTest(unittest.makeSuite(test))
1135     result = unittest.TextTestRunner(stream=stream, failfast=failfast).run(basic_tests_suite)
1136     executed += result.testsRun
1137     failed += len(result.failures) + len(result.errors)
1138     if failfast and failed:
1139         sys.exit(1)
1140     if len(result.failures) > 0:
1141         logger.debug("failures : {}".format(result.failures))
1142     if len(result.errors) > 0:
1143         logger.debug("errors : {}".format(result.errors))
1144     return executed, failed
1145
1146
1147 def test_vim(args):
1148     global test_config
1149     sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + "/osm_ro")
1150     import openmanoclient
1151     executed = 0
1152     failed = 0
1153     test_config["client"] = openmanoclient.openmanoclient(
1154         endpoint_url=args.endpoint_url,
1155         tenant_name=args.tenant_name,
1156         datacenter_name=args.datacenter,
1157         debug=args.debug, logger=test_config["logger_name"])
1158     clsmembers = inspect.getmembers(sys.modules[__name__], inspect.isclass)
1159     # If only want to obtain a tests list print it and exit
1160     if args.list_tests:
1161         tests_names = []
1162         for cls in clsmembers:
1163             if cls[0].startswith('test_VIM'):
1164                 tests_names.append(cls[0])
1165
1166         msg = "The 'vim' set tests are:\n\t" + ', '.join(sorted(tests_names)) +\
1167               "\nNOTE: The test test_VIM_tenant_operations will fail in case the used datacenter is type OpenStack " \
1168               "unless RO has access to the admin endpoint. Therefore this test is excluded by default"
1169         print(msg)
1170         logger.info(msg)
1171         sys.exit(0)
1172
1173     # Create the list of tests to be run
1174     code_based_tests = []
1175     if args.tests:
1176         for test in args.tests:
1177             for t in test.split(','):
1178                 matches_code_based_tests = [item for item in clsmembers if item[0] == t]
1179                 if len(matches_code_based_tests) > 0:
1180                     code_based_tests.append(matches_code_based_tests[0][1])
1181                 else:
1182                     logger.critical("Test '{}' is not among the possible ones".format(t))
1183                     sys.exit(1)
1184     if not code_based_tests:
1185         # include all tests
1186         for cls in clsmembers:
1187             # We exclude 'test_VIM_tenant_operations' unless it is specifically requested by the user
1188             if cls[0].startswith('test_VIM') and cls[0] != 'test_VIM_tenant_operations':
1189                 code_based_tests.append(cls[1])
1190
1191     logger.debug("tests to be executed: {}".format(code_based_tests))
1192
1193     # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1194     # This is handled in the tests using logging.
1195     stream = open('/dev/null', 'w')
1196
1197     # Run code based tests
1198     basic_tests_suite = unittest.TestSuite()
1199     for test in code_based_tests:
1200         basic_tests_suite.addTest(unittest.makeSuite(test))
1201     result = unittest.TextTestRunner(stream=stream, failfast=failfast).run(basic_tests_suite)
1202     executed += result.testsRun
1203     failed += len(result.failures) + len(result.errors)
1204     if failfast and failed:
1205         sys.exit(1)
1206     if len(result.failures) > 0:
1207         logger.debug("failures : {}".format(result.failures))
1208     if len(result.errors) > 0:
1209         logger.debug("errors : {}".format(result.errors))
1210     return executed, failed
1211
1212
1213 def test_deploy(args):
1214     global test_config
1215     sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + "/osm_ro")
1216     import openmanoclient
1217     executed = 0
1218     failed = 0
1219     test_config["test_directory"] = os.path.dirname(__file__) + "/RO_tests"
1220     test_config["image_name"] = args.image_name
1221     test_config["mgmt_net"] = args.mgmt_net
1222     test_config["manual"] = args.manual
1223     test_directory_content = os.listdir(test_config["test_directory"])
1224     # If only want to obtain a tests list print it and exit
1225     if args.list_tests:
1226         msg = "he 'deploy' set tests are:\n\t" + ', '.join(sorted(test_directory_content))
1227         print(msg)
1228         logger.info(msg)
1229         sys.exit(0)
1230
1231     descriptor_based_tests = []
1232     # Create the list of tests to be run
1233     code_based_tests = []
1234     if args.tests:
1235         for test in args.tests:
1236             for t in test.split(','):
1237                 if t in test_directory_content:
1238                     descriptor_based_tests.append(t)
1239                 else:
1240                     logger.critical("Test '{}' is not among the possible ones".format(t))
1241                     sys.exit(1)
1242     if not descriptor_based_tests:
1243         # include all tests
1244         descriptor_based_tests = test_directory_content
1245
1246     logger.debug("tests to be executed: {}".format(code_based_tests))
1247
1248     # import openmanoclient from relative path
1249     test_config["client"] = openmanoclient.openmanoclient(
1250         endpoint_url=args.endpoint_url,
1251         tenant_name=args.tenant_name,
1252         datacenter_name=args.datacenter,
1253         debug=args.debug, logger=test_config["logger_name"])
1254
1255     # TextTestRunner stream is set to /dev/null in order to avoid the method to directly print the result of tests.
1256     # This is handled in the tests using logging.
1257     stream = open('/dev/null', 'w')
1258     # This scenario based tests are defined as directories inside the directory defined in 'test_directory'
1259     for test in descriptor_based_tests:
1260         test_config["test_folder"] = test
1261         test_suite = unittest.TestSuite()
1262         test_suite.addTest(unittest.makeSuite(descriptor_based_scenario_test))
1263         result = unittest.TextTestRunner(stream=stream, failfast=False).run(test_suite)
1264         executed += result.testsRun
1265         failed += len(result.failures) + len(result.errors)
1266         if failfast and failed:
1267             sys.exit(1)
1268         if len(result.failures) > 0:
1269             logger.debug("failures : {}".format(result.failures))
1270         if len(result.errors) > 0:
1271             logger.debug("errors : {}".format(result.errors))
1272
1273     return executed, failed
1274
1275 if __name__=="__main__":
1276
1277     parser = ArgumentParser(description='Test RO module')
1278     parser.add_argument('-v','--version', action='version', help="Show current version",
1279                              version='%(prog)s version ' + __version__  + ' ' + version_date)
1280
1281     # Common parameters
1282     parent_parser = ArgumentParser(add_help=False)
1283     parent_parser.add_argument('--failfast', help='Stop when a test fails rather than execute all tests',
1284                       dest='failfast', action="store_true", default=False)
1285     parent_parser.add_argument('--failed', help='Set logs to show only failed tests. --debug disables this option',
1286                       dest='failed', action="store_true", default=False)
1287     default_logger_file = os.path.dirname(__file__)+'/'+os.path.splitext(os.path.basename(__file__))[0]+'.log'
1288     parent_parser.add_argument('--list-tests', help='List all available tests', dest='list_tests', action="store_true",
1289                       default=False)
1290     parent_parser.add_argument('--logger_file', dest='logger_file', default=default_logger_file,
1291                                help='Set the logger file. By default '+default_logger_file)
1292     parent_parser.add_argument("-t", '--tenant', dest='tenant_name', default="osm",
1293                                help="Set the openmano tenant to use for the test. By default 'osm'")
1294     parent_parser.add_argument('--debug', help='Set logs to debug level', dest='debug', action="store_true")
1295     parent_parser.add_argument('--timeout', help='Specify the instantiation timeout in seconds. By default 300',
1296                           dest='timeout', type=int, default=300)
1297     parent_parser.add_argument('--test', '--tests', help='Specify the tests to run', dest='tests', action="append")
1298
1299     subparsers = parser.add_subparsers(help='test sets')
1300
1301     # Deployment test set
1302     # -------------------
1303     deploy_parser = subparsers.add_parser('deploy', parents=[parent_parser],
1304                                           help="test deployment using descriptors at RO_test folder ")
1305     deploy_parser.set_defaults(func=test_deploy)
1306
1307     # Mandatory arguments
1308     mandatory_arguments = deploy_parser.add_argument_group('mandatory arguments')
1309     mandatory_arguments.add_argument('-d', '--datacenter', required=True, help='Set the datacenter to test')
1310     mandatory_arguments.add_argument("-i", '--image-name', required=True, dest="image_name",
1311                                      help='Image name available at datacenter used for the tests')
1312     mandatory_arguments.add_argument("-n", '--mgmt-net-name', required=True, dest='mgmt_net',
1313                                      help='Set the vim management network to use for tests')
1314
1315     # Optional arguments
1316     deploy_parser.add_argument('-m', '--manual-check', dest='manual', action="store_true", default=False,
1317                                help='Pause execution once deployed to allow manual checking of the '
1318                                     'deployed instance scenario')
1319     deploy_parser.add_argument('-u', '--url', dest='endpoint_url', default='http://localhost:9090/openmano',
1320                                help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
1321
1322     # Vimconn test set
1323     # -------------------
1324     vimconn_parser = subparsers.add_parser('vimconn', parents=[parent_parser], help="test vimconnector plugin")
1325     vimconn_parser.set_defaults(func=test_vimconnector)
1326     # Mandatory arguments
1327     mandatory_arguments = vimconn_parser.add_argument_group('mandatory arguments')
1328     mandatory_arguments.add_argument('--vimtype', choices=['vmware', 'aws', 'openstack', 'openvim'], required=True,
1329                                      help='Set the vimconnector type to test')
1330     mandatory_arguments.add_argument('-c', '--config', dest='config_param', required=True,
1331                                     help='Set the vimconnector specific config parameters in dictionary format')
1332     mandatory_arguments.add_argument('-u', '--url', dest='endpoint_url',required=True, help="Set the vim connector url or Host IP")
1333     # Optional arguments
1334     # TODO add optional arguments for vimconn tests
1335     # vimconn_parser.add_argument("-i", '--image-name', dest='image_name', help='<HELP>'))
1336
1337     # Datacenter test set
1338     # -------------------
1339     vimconn_parser = subparsers.add_parser('vim', parents=[parent_parser], help="test vim")
1340     vimconn_parser.set_defaults(func=test_vim)
1341
1342     # Mandatory arguments
1343     mandatory_arguments = vimconn_parser.add_argument_group('mandatory arguments')
1344     mandatory_arguments.add_argument('-d', '--datacenter', required=True, help='Set the datacenter to test')
1345
1346     # Optional arguments
1347     vimconn_parser.add_argument('-u', '--url', dest='endpoint_url', default='http://localhost:9090/openmano',
1348                                help="Set the openmano server url. By default 'http://localhost:9090/openmano'")
1349
1350     argcomplete.autocomplete(parser)
1351     args = parser.parse_args()
1352     # print str(args)
1353     test_config = {}
1354
1355     # default logger level is INFO. Options --debug and --failed override this, being --debug prioritary
1356     logger_level = 'INFO'
1357     if args.debug:
1358         logger_level = 'DEBUG'
1359     elif args.failed:
1360         logger_level = 'WARNING'
1361     logger_name = os.path.basename(__file__)
1362     test_config["logger_name"] = logger_name
1363     logger = logging.getLogger(logger_name)
1364     logger.setLevel(logger_level)
1365     failfast = args.failfast
1366
1367     # Configure a logging handler to store in a logging file
1368     if args.logger_file:
1369         fileHandler = logging.FileHandler(args.logger_file)
1370         formatter_fileHandler = logging.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s')
1371         fileHandler.setFormatter(formatter_fileHandler)
1372         logger.addHandler(fileHandler)
1373
1374     # Configure a handler to print to stdout
1375     consoleHandler = logging.StreamHandler(sys.stdout)
1376     formatter_consoleHandler = logging.Formatter('%(asctime)s %(name)s %(levelname)s: %(message)s')
1377     consoleHandler.setFormatter(formatter_consoleHandler)
1378     logger.addHandler(consoleHandler)
1379
1380     logger.debug('Program started with the following arguments: ' + str(args))
1381
1382     # set test config parameters
1383     test_config["timeout"] = args.timeout
1384     test_config["test_number"] = 1
1385
1386     executed, failed = args.func(args)
1387
1388     # Log summary
1389     logger.warning("Total number of tests: {}; Total number of failures/errors: {}".format(executed, failed))
1390     sys.exit(1 if failed else 0)