#!/usr/bin/python # # Copyright (c) 2015 All rights reserved # This program and the accompanying materials # are made available under the terms of the Apache License, Version 2.0 # which accompanies this distribution, and is available at # # http://www.apache.org/licenses/LICENSE-2.0 # # 0.1: This script boots the VM1 and allocates IP address from Nova # Later, the VM2 boots then execute cloud-init to ping VM1. # After successful ping, both the VMs are deleted. # 0.2: measure test duration and publish results under json format # # import os import time import argparse import pprint import sys import logging import yaml import datetime import novaclient.v2.client as novaclient from neutronclient.v2_0 import client as neutronclient from keystoneclient.v2_0 import client as keystoneclient from glanceclient import client as glanceclient pp = pprint.PrettyPrinter(indent=4) parser = argparse.ArgumentParser() parser.add_argument("repo_path", help="Path to the repository") parser.add_argument("-d", "--debug", help="Debug mode", action="store_true") parser.add_argument("-r", "--report", help="Create json result file", action="store_true") args = parser.parse_args() sys.path.append(args.repo_path + "testcases/") import functest_utils """ logging configuration """ logger = logging.getLogger('vPing') logger.setLevel(logging.DEBUG) ch = logging.StreamHandler() if args.debug: ch.setLevel(logging.DEBUG) else: ch.setLevel(logging.INFO) formatter = logging.Formatter('%(asctime)s - %(name)s' '- %(levelname)s - %(message)s') ch.setFormatter(formatter) logger.addHandler(ch) HOME = os.environ['HOME'] + "/" with open(args.repo_path + "testcases/config_functest.yaml") as f: functest_yaml = yaml.safe_load(f) f.close() # vPing parameters VM_BOOT_TIMEOUT = 180 VM_DELETE_TIMEOUT = 100 PING_TIMEOUT = functest_yaml.get("vping").get("ping_timeout") TEST_DB = functest_yaml.get("results").get("test_db_url") NAME_VM_1 = functest_yaml.get("vping").get("vm_name_1") NAME_VM_2 = functest_yaml.get("vping").get("vm_name_2") IP_1 = functest_yaml.get("vping").get("ip_1") IP_2 = functest_yaml.get("vping").get("ip_2") # GLANCE_IMAGE_NAME = functest_yaml.get("general"). \ # get("openstack").get("image_name") GLANCE_IMAGE_NAME = "functest-vping" GLANCE_IMAGE_FILENAME = functest_yaml.get("general"). \ get("openstack").get("image_file_name") GLANCE_IMAGE_FORMAT = functest_yaml.get("general"). \ get("openstack").get("image_disk_format") GLANCE_IMAGE_PATH = functest_yaml.get("general"). \ get("directories").get("dir_functest_data") + "/" + GLANCE_IMAGE_FILENAME FLAVOR = functest_yaml.get("vping").get("vm_flavor") # NEUTRON Private Network parameters NEUTRON_PRIVATE_NET_NAME = functest_yaml.get("vping"). \ get("vping_private_net_name") NEUTRON_PRIVATE_SUBNET_NAME = functest_yaml.get("vping"). \ get("vping_private_subnet_name") NEUTRON_PRIVATE_SUBNET_CIDR = functest_yaml.get("vping"). \ get("vping_private_subnet_cidr") NEUTRON_ROUTER_NAME = functest_yaml.get("vping"). \ get("vping_router_name") def pMsg(value): """pretty printing""" pp.pprint(value) def waitVmActive(nova, vm): # sleep and wait for VM status change sleep_time = 3 count = VM_BOOT_TIMEOUT / sleep_time while True: status = functest_utils.get_instance_status(nova, vm) logger.debug("Status: %s" % status) if status == "ACTIVE": return True if status == "ERROR" or count == 0: return False count -= 1 time.sleep(sleep_time) return False def waitVmDeleted(nova, vm): # sleep and wait for VM status change sleep_time = 3 count = VM_DELETE_TIMEOUT / sleep_time while True: status = functest_utils.get_instance_status(nova, vm) if not status: return True elif count == 0: logger.debug("Timeout") return False else: # return False count -= 1 time.sleep(sleep_time) return False def create_private_neutron_net(neutron): neutron.format = 'json' logger.info('Creating neutron network %s...' % NEUTRON_PRIVATE_NET_NAME) network_id = functest_utils. \ create_neutron_net(neutron, NEUTRON_PRIVATE_NET_NAME) if not network_id: return False logger.debug("Network '%s' created successfully" % network_id) logger.debug('Creating Subnet....') subnet_id = functest_utils. \ create_neutron_subnet(neutron, NEUTRON_PRIVATE_SUBNET_NAME, NEUTRON_PRIVATE_SUBNET_CIDR, network_id) if not subnet_id: return False logger.debug("Subnet '%s' created successfully" % subnet_id) logger.debug('Creating Router...') router_id = functest_utils. \ create_neutron_router(neutron, NEUTRON_ROUTER_NAME) if not router_id: return False logger.debug("Router '%s' created successfully" % router_id) logger.debug('Adding router to subnet...') result = functest_utils.add_interface_router(neutron, router_id, subnet_id) if not result: return False logger.debug("Interfac
heat_template_version: pike

description: >
  OpenStack Neutron Compute Midonet plugin

parameters:
  ServiceNetMap:
    default: {}
    description: Mapping of service_name -> network name. Typically set
                 via parameter_defaults in the resource registry.  This
                 mapping overrides those in ServiceNetMapDefaults.
    type: json
  DefaultPasswords:
    default: {}
    type: json
  RoleName:
    default: ''
    description: Role name on which the service is applied
    type: string
  RoleParameters:
    default: {}
    description: Parameters specific to the role
    type: json
  EndpointMap:
    default: {}