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