docs/release/release-notes/functest-release.rst


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162

.. SPDX-License-Identifier: CC-BY-4.0

======================================
OPNFV master release note for Functest
======================================

Abstract
========

This document contains the release notes of the Functest project.

OPNFV master Release
====================

Functest deals with functional testing of the OPNFV solution.
It includes test cases developed within the project, test cases developed in
other OPNFV projects and it also integrates test cases from other upstream
communities.

OpenStack
---------

The internal test cases are:

 * connection_check
 * api_check
 * snaps_health_check
 * vping_ssh
 * vping_userdata
 * tempest_smoke_serial
 * rally_sanity
 * refstack_defcore
 * patrole
 * odl
 * snaps_smoke
 * neutron_trunk
 * tempest_full_parallel
 * rally_full
 * cloudify_ims
 * vyos_vrouter
 * juju_epc

The OPNFV projects integrated into Functest framework for automation are:

 * doctor
 * bgpvpn
 * odl-sfc
 * barometer
 * fds

Kubernetes
----------

The internal test cases are:

 * k8s_smoke
 * k8s_conformance

The OPNFV projects integrated into Functest framework for automation are:

 * stor4nfv
 * clover

Release Data
============

+--------------------------------------+--------------------------------------+
| **Project**                          | functest                             |
+--------------------------------------+--------------------------------------+
| **Repository branch**                | master                               |
+--------------------------------------+--------------------------------------+

Deliverables
============

Software
--------

 Functest Docker images (OpenStack):

 * https://hub.docker.com/r/opnfv/functest-healthcheck
 * https://hub.docker.com/r/opnfv/functest-smoke
 * https://hub.docker.com/r/opnfv/functest-features
 * https://hub.docker.com/r/opnfv/functest-components
 * https://hub.docker.com/r/opnfv/functest-vnf

 Functest Docker images (Kubernetes):

 * https://hub.docker.com/r/opnfv/functest-kubernetes-healthcheck
 * https://hub.docker.com/r/opnfv/functest-kubernetes-smoke
 * https://hub.docker.com/r/opnfv/functest-kubernetes-features

Docker tag for master: latest

Documents
---------

 * Config Guide: http://docs.opnfv.org/en/latest/submodules/functest/docs/testing/user/configguide/index.html
 * User Guide: http://docs.opnfv.org/en/latest/submodules/functest/docs/testing/user/userguide/index.html
 * Developer Guide: http://docs.opnfv.org/en/latest/submodules/functest/docs/testing/developer/devguide/index.html
 * API Docs: http://functest.readthedocs.io/en/latest/

Version change
==============

New test cases
--------------

 * neutron_trunk
 * patrole
 * juju_epc
 * k8s_smoke
 * k8s_conformance
 * stor4nfv
 * clover

Key changes
-----------

 * update test cases and containers to `OpenStack Pike`_
 * move the framework into a separate project: Xtesting_
 * ease running all containers thanks to default values
 * clean interfaces with OPNFV Installers and Features
 * rewrite all vnfs to allow multiple tests in parallel
 * fully support non-default region names and Keystone v3 domains
 * refactor all tempest-based test cases (e.g. refstack_defcore)
 * remove obsolete OpenStack and Functest utils
 * verify all changes via doc8 and yamllint too
 * generate reports for all tempest-based test cases

.. _`OpenStack Pike`: https://raw.githubusercontent.com/openstack/requirements/stable/pike/upper-constraints.txt
.. _Xtesting: http://xtesting.readthedocs.io/en/latest/

Key benefits
------------

 * the enduser can run all tests by setting only one input (EXTERNAL_NETWORK)
 * the developer can only work on the test suites without diving into CI/CD
   integration
 * both OpenStack and Kubernetes deployments can be verified
 * Functest test cases are trustable as they meet the best coding rules (unit
   tests, coverage, linters, etc.)
 * Functest can be reused in other projects (e.g.
   `ONS: Re-using OPNFV framework tests for LFN projects`_)

.. _`ONS: Re-using OPNFV framework tests for LFN projects`: https://wiki.lfnetworking.org/display/LN/LFN+Developer+Forum+Schedule?preview=/328197/328329/ONS-OPNFV%20framework%20tests%20for%20LFN%20projects.pdf

Code quality
------------

 * pylint: ~9.5/10
 * code coverage: ~70%

Useful links
============

 * wiki project page: https://wiki.opnfv.org/opnfv_functional_testing
 * Functest git repository: https://git.opnfv.org/cgit/functest
 * Functest CI dashboard: https://build.opnfv.org/ci/view/functest/
 * JIRA dashboard: https://jira.opnfv.org/secure/Dashboard.jspa?selectPageId=10611
 * Functest IRC channel: #opnfv-functest
 * Reporting page: http://testresults.opnfv.org/reporting/master/functest/functest.html
#!/usr/bin/env python
#
# jose.lausuch@ericsson.com
# valentin.boucher@orange.com
# 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
#

import json
import os
import os.path
import re
import requests
import shutil
import socket
import subprocess
import sys
import urllib2
from git import Repo


# ----------------------------------------------------------
#
#               OPENSTACK UTILS
#
# -----------------------------------------------------------


#*********************************************
#   CREDENTIALS
#*********************************************
def check_credentials():
    """
    Check if the OpenStack credentials (openrc) are sourced
    """
    env_vars = ['OS_AUTH_URL', 'OS_USERNAME', 'OS_PASSWORD', 'OS_TENANT_NAME']
    return all(map(lambda v: v in os.environ and os.environ[v], env_vars))


def get_credentials(service):
    """Returns a creds dictionary filled with the following keys:
    * username
    * password/api_key (depending on the service)
    * tenant_name/project_id (depending on the service)
    * auth_url
    :param service: a string indicating the name of the service
                    requesting the credentials.
    """
    creds = {}
    # Unfortunately, each of the OpenStack client will request slightly
    # different entries in their credentials dict.
    if service.lower() in ("nova", "cinder"):
        password = "api_key"
        tenant = "project_id"
    else:
        password = "password"
        tenant = "tenant_name"

    # The most common way to pass these info to the script is to do it through
    # environment variables.
    creds.update({
        "username": os.environ.get('OS_USERNAME', "admin"),
        password: os.environ.get("OS_PASSWORD", 'admin'),
        "auth_url": os.environ.get("OS_AUTH_URL",
                                   "http://192.168.20.71:5000/v2.0"),
        tenant: os.environ.get("OS_TENANT_NAME", "admin"),
    })
    return creds


#*********************************************
#   NOVA
#*********************************************
def get_instances(nova_client):
    try:
        instances = nova_client.servers.list(search_opts={'all_tenants': 1})
        return instances
    except Exception, e:
        print "Error [get_instances(nova_client)]:", e
        return None


def get_instance_status(nova_client, instance):
    try:
        instance = nova_client.servers.get(instance.id)
        return instance.status
    except Exception, e:
        #print "Error [get_instance_status(nova_client, '%s')]:" % \
        #    str(instance), e
        return None


def get_instance_by_name(nova_client, instance_name):
    try:
        instance = nova_client.servers.find(name=instance_name)
        return instance
    except Exception, e:
        print "Error [get_instance_by_name(nova_client, '%s')]:" % \
            instance_name, e
        return None


def get_flavor_id(nova_client, flavor_name):
    flavors = nova_client.flavors.list(detailed=True)
    id = ''
    for f in flavors:
        if f.name == flavor_name:
            id = f.id
            break
    return id


def get_flavor_id_by_ram_range(nova_client, min_ram, max_ram):
    flavors = nova_client.flavors.list(detailed=True)
    id = ''
    for f in flavors:
        if min_ram <= f.ram and f.ram <= max_ram:
            id = f.id
            break
    return id


def get_floating_ips(nova_client):
    try:
        floating_ips = nova_client.floating_ips.list()
        return floating_ips
    except Exception, e:
        print "Error [get_floating_ips(nova_client)]:", e
        return None


def create_flavor(nova_client, flavor_name, ram, disk, vcpus):
    try:
        flavor = nova_client.flavors.create(flavor_name, ram, vcpus, disk)
    except Exception, e:
        print "Error [create_flavor(nova_client, '%s', '%s', '%s', "\
            "'%s')]:" % (flavor_name, ram, disk, vcpus), e
        return None
    return flavor.id


def create_floating_ip(neutron_client):
    extnet_id = get_external_net_id(neutron_client)
    props = {'floating_network_id': extnet_id}
    try:
        ip_json = neutron_client.create_floatingip({'floatingip': props})
        fip_addr = ip_json['floatingip']['floating_ip_address']
        fip_id = ip_json['floatingip']['id']
    except Exception, e:
        print "Error [create_floating_ip(neutron_client)]:", e
        return None
    return {'fip_addr': fip_addr, 'fip_id': fip_id}


def add_floating_ip(nova_client, server_id, floatingip_id):
    try:
        nova_client.servers.add_floating_ip(server_id, floatingip_id)
        return True
    except Exception, e:
        print "Error [add_floating_ip(nova_client, '%s', '%s')]:" % \
            (server_id, floatingip_id), e
        return False


def delete_instance(nova_client, instance_id):
    try:
        nova_client.servers.force_delete(instance_id)
        return True
    except Exception, e:
        print "Error [delete_instance(nova_client, '%s')]:" % instance_id, e
        return False


def delete_floating_ip(nova_client, floatingip_id):
    try:
        nova_client.floating_ips.delete(floatingip_id)
        return True
    except Exception, e:
        print "Error [delete_floating_ip(nova_client, '%s')]:" % floatingip_id, e
        return False


#*********************************************
#   NEUTRON
#*********************************************
def get_network_list(neutron_client):
    network_list = neutron_client.list_networks()['networks']
    if len(network_list) == 0:
        return None
    else:
        return network_list


def get_router_list(neutron_client):
    router_list = neutron_client.list_routers()['routers']
    if len(router_list) == 0:
        return None
    else:
        return router_list


def get_port_list(neutron_client):
    port_list = neutron_client.list_ports()['ports']
    if len(port_list) == 0:
        return None
    else:
        return port_list


def get_network_id(neutron_client, network_name):
    networks = neutron_client.list_networks()['networks']
    id = ''
    for n in networks:
        if n['name'] == network_name:
            id = n['id']
            break
    return id


def get_subnet_id(neutron_client, subnet_name):
    subnets = neutron_client.list_subnets()['subnets']
    id = ''
    for s in subnets:
        if s['name'] == subnet_name:
            id = s['id']
            break
    return id


def get_router_id(neutron_client, router_name):
    routers = neutron_client.list_routers()['routers']
    id = ''
    for r in routers:
        if r['name'] == router_name:
            id = r['id']
            break
    return id


def get_private_net(neutron_client):
    # Checks if there is an existing shared private network
    networks = neutron_client.list_networks()['networks']
    if len(networks) == 0:
        return None
    for net in networks:
        if (net['router:external'] is False) and (net['shared'] is True):
            return net
    return None


def get_external_net(neutron_client):
    for network in neutron_client.list_networks()['networks']:
        if network['router:external']:
            return network['name']
    return False


def get_external_net_id(neutron_client):
    for network in neutron_client.list_networks()['networks']:
        if network['router:external']:
            return network['id']
    return False


def check_neutron_net(neutron_client, net_name):
    for network in neutron_client.list_networks()['networks']:
        if network['name'] == net_name:
            for subnet in network['subnets']:
                return True
    return False


def create_neutron_net(neutron_client, name):
    json_body = {'network': {'name': name,
                             'admin_state_up': True}}
    try:
        network = neutron_client.create_network(body=json_body)
        network_dict = network['network']
        return network_dict['id']
    except Exception, e:
        print "Error [create_neutron_net(neutron_client, '%s')]:" % name, e
        return False


def create_neutron_subnet(neutron_client, name, cidr, net_id):
    json_body = {'subnets': [{'name': name, 'cidr': cidr,
                              'ip_version': 4, 'network_id': net_id}]}
    try:
        subnet = neutron_client.create_subnet(body=json_body)
        return subnet['subnets'][0]['id']
    except Exception, e:
        print "Error [create_neutron_subnet(neutron_client, '%s', '%s', "\
            "'%s')]:" % (name, cidr, net_id), e
        return False


def create_neutron_router(neutron_client, name):
    json_body = {'router': {'name': name, 'admin_state_up': True}}
    try:
        router = neutron_client.create_router(json_body)
        return router['router']['id']
    except Exception, e:
        print "Error [create_neutron_router(neutron_client, '%s')]:" % name, e
        return False


def create_neutron_port(neutron_client, name, network_id, ip):
    json_body = {'port': {
                 'admin_state_up': True,
                 'name': name,
                 'network_id': network_id,
                 'fixed_ips': [{"ip_address": ip}]
                 }}
    try:
        port = neutron_client.create_port(body=json_body)
        return port['port']['id']
    except Exception, e:
        print "Error [create_neutron_port(neutron_client, '%s', '%s', "\
            "'%s')]:" % (name, network_id, ip), e
        return False


def update_neutron_net(neutron_client, network_id, shared=False):
    json_body = {'network': {'shared': shared}}
    try:
        neutron_client.update_network(network_id, body=json_body)
        return True
    except Exception, e:
        print "Error [update_neutron_net(neutron_client, '%s', '%s')]:" % \
            (network_id, str(shared)), e
        return False


def update_neutron_port(neutron_client, port_id, device_owner):
    json_body = {'port': {
                 'device_owner': device_owner,
                 }}
    try:
        port = neutron_client.update_port(port=port_id,
                                          body=json_body)
        return port['port']['id']
    except Exception, e:
        print "Error [update_neutron_port(neutron_client, '%s', '%s')]:" % \
            (port_id, device_owner), e
        return False


def add_interface_router(neutron_client, router_id, subnet_id):
    json_body = {"subnet_id": subnet_id}
    try:
        neutron_client.add_interface_router(router=router_id, body=json_body)
        return True
    except Exception, e:
        print "Error [add_interface_router(neutron_client, '%s', '%s')]:" % \
            (router_id, subnet_id), e
        return False


def add_gateway_router(neutron_client, router_id):
    ext_net_id = get_external_net_id(neutron_client)
    router_dict = {'network_id': ext_net_id}
    try:
        neutron_client.add_gateway_router(router_id, router_dict)
        return True
    except Exception, e:
        print "Error [add_gateway_router(neutron_client, '%s')]:" % router_id, e
        return False


def delete_neutron_net(neutron_client, network_id):
    try:
        neutron_client.delete_network(network_id)
        return True
    except Exception, e:
        print "Error [delete_neutron_net(neutron_client, '%s')]:" % network_id, e
        return False


def delete_neutron_subnet(neutron_client, subnet_id):
    try:
        neutron_client.delete_subnet(subnet_id)
        return True
    except Exception, e:
        print "Error [delete_neutron_subnet(neutron_client, '%s')]:" % subnet_id, e
        return False


def delete_neutron_router(neutron_client, router_id):
    json_body = {'router': {'id': router_id}}
    try:
        neutron_client.delete_router(router=router_id)
        return True
    except Exception, e:
        print "Error [delete_neutron_router(neutron_client, '%s')]:" % \
            router_id, e
        return False


def delete_neutron_port(neutron_client, port_id):
    try:
        neutron_client.delete_port(port_id)
        return True
    except Exception, e:
        print "Error [delete_neutron_port(neutron_client, '%s')]:" % port_id, e
        return False


def remove_interface_router(neutron_client, router_id, subnet_id):
    json_body = {"subnet_id": subnet_id}
    try:
        neutron_client.remove_interface_router(router=router_id,
                                               body=json_body)
        return True
    except Exception, e:
        print "Error [remove_interface_router(neutron_client, '%s', '%s')]:" % \
            (router_id, subnet_id), e
        return False


def remove_gateway_router(neutron_client, router_id):
    try:
        neutron_client.remove_gateway_router(router_id)
        return True
    except Exception, e:
        print "Error [remove_gateway_router(neutron_client, '%s')]:" % router_id, e
        return False


#*********************************************
#   SEC GROUPS
#*********************************************
def get_security_groups(neutron_client):
    try:
        security_groups = neutron_client.list_security_groups()[
            'security_groups']
        return security_groups
    except Exception, e:
        print "Error [get_security_groups(neutron_client)]:", e
        return None

def get_security_group_id(neutron_client, sg_name):
    security_groups = get_security_groups(neutron_client)
    id = ''
    for sg in security_groups:
        if sg['name'] == sg_name:
            id = sg['id']
            break
    return id


def create_security_group(neutron_client, sg_name, sg_description):
    json_body = {'security_group': {'name': sg_name,
                                    'description': sg_description}}
    try:
        secgroup = neutron_client.create_security_group(json_body)
        return secgroup['security_group']
    except Exception, e:
        print "Error [create_security_group(neutron_client, '%s', '%s')]:" % \
            (sg_name, sg_description), e
        return False


def create_secgroup_rule(neutron_client, sg_id, direction, protocol,
                         port_range_min=None, port_range_max=None):
    if port_range_min is None and port_range_max is None:
        json_body = {'security_group_rule': {'direction': direction,
                                             'security_group_id': sg_id,
                                             'protocol': protocol}}
    elif port_range_min is not None and port_range_max is not None:
        json_body = {'security_group_rule': {'direction': direction,
                                             'security_group_id': sg_id,
                                             'port_range_min': port_range_min,
                                             'port_range_max': port_range_max,
                                             'protocol': protocol}}
    else:
        print "Error [create_secgroup_rule(neutron_client, '%s', '%s', "\
              "'%s', '%s', '%s', '%s')]:" % (neutron_client, sg_id, direction, \
                                             port_range_min, port_range_max, protocol),\
              " Invalid values for port_range_min, port_range_max"
        return False
    try:
        neutron_client.create_security_group_rule(json_body)
        return True
    except Exception, e:
        print "Error [create_secgroup_rule(neutron_client, '%s', '%s', "\
            "'%s', '%s', '%s', '%s')]:" % (neutron_client, sg_id, direction,
                                           port_range_min, port_range_max,
                                           protocol), e
        return False


def add_secgroup_to_instance(nova_client, instance_id, secgroup_id):
    try:
        nova_client.servers.add_security_group(instance_id, secgroup_id)
        return True
    except Exception, e:
        print "Error [add_secgroup_to_instance(nova_client, '%s', '%s')]: " % \
            (instance_id, secgroup_id), e
        return False


def update_sg_quota(neutron_client, tenant_id, sg_quota, sg_rule_quota):
    json_body = {"quota": {
        "security_group": sg_quota,
        "security_group_rule": sg_rule_quota
    }}

    try:
        quota = neutron_client.update_quota(tenant_id=tenant_id,
                                            body=json_body)
        return True
    except Exception, e:
        print "Error [update_sg_quota(neutron_client, '%s', '%s', "\
            "'%s')]:" % (tenant_id, sg_quota, sg_rule_quota), e
        return False


def delete_security_group(neutron_client, secgroup_id):
    try:
        neutron_client.delete_security_group(secgroup_id)
        return True
    except Exception, e:
        print "Error [delete_security_group(neutron_client, '%s')]:" % secgroup_id, e
        return False


#*********************************************
#   GLANCE
#*********************************************
def get_images(nova_client):
    try:
        images = nova_client.images.list()
        return images
    except Exception, e:
        print "Error [get_images]:", e
        return None


def get_image_id(glance_client, image_name):
    images = glance_client.images.list()
    id = ''
    for i in images:
        if i.name == image_name:
            id = i.id
            break
    return id


def create_glance_image(glance_client, image_name, file_path, public=True):
    if not os.path.isfile(file_path):
        print "Error: file " + file_path + " does not exist."
        return False
    try:
        with open(file_path) as fimage:
            image = glance_client.images.create(name=image_name,
                                                is_public=public,
                                                disk_format="qcow2",
                                                container_format="bare",
                                                data=fimage)
        return image.id
    except Exception, e:
        print "Error [create_glance_image(glance_client, '%s', '%s', "\
            "'%s')]:" % (image_name, file_path, str(public)), e
        return False


def delete_glance_image(nova_client, image_id):
    try:
        nova_client.images.delete(image_id)
        return True
    except Exception, e:
        print "Error [delete_glance_image(nova_client, '%s')]:" % image_id, e
        return False


#*********************************************
#   CINDER
#*********************************************
def get_volumes(cinder_client):
    try:
        volumes = cinder_client.volumes.list(search_opts={'all_tenants': 1})
        return volumes
    except Exception, e:
        print "Error [get_volumes(cinder_client)]:", e
        return None


def list_volume_types(cinder_client, public=True, private=True):
    try:
        volume_types = cinder_client.volume_types.list()
        if not public:
            volume_types = [vt for vt in volume_types if not vt.is_public]
        if not private:
            volume_types = [vt for vt in volume_types if vt.is_public]
        return volume_types
    except Exception, e:
        print "Error [list_volume_types(cinder_client)]:", e
        return None


def create_volume_type(cinder_client, name):
    try:
        volume_type = cinder_client.volume_types.create(name)
        return volume_type
    except Exception, e:
        print "Error [create_volume_type(cinder_client, '%s')]:" % name, e
        return None


def update_cinder_quota(cinder_client, tenant_id, vols_quota,
                        snapshots_quota, gigabytes_quota):
    quotas_values = {"volumes": vols_quota,
                     "snapshots": snapshots_quota,
                     "gigabytes": gigabytes_quota}

    try:
        quotas_default = cinder_client.quotas.update(tenant_id,
                                                     **quotas_values)
        return True
    except Exception, e:
        print "Error [update_cinder_quota(cinder_client, '%s', '%s', '%s'" \
            "'%s')]:" % (tenant_id, vols_quota, snapshots_quota, gigabytes_quota), e
        return False


def delete_volume(cinder_client, volume_id, forced=False):
    try:
        if forced:
            try:
                cinder_client.volumes.detach(volume_id)
            except:
                print "Error:", sys.exc_info()[0]
            cinder_client.volumes.force_delete(volume_id)
        else:
            cinder_client.volumes.delete(volume_id)
        return True
    except Exception, e:
        print "Error [delete_volume(cinder_client, '%s', '%s')]:" % \
            (volume_id, str(forced)), e
        return False


def delete_volume_type(cinder_client, volume_type):
    try:
        cinder_client.volume_types.delete(volume_type)
        return True
    except Exception, e:
        print "Error [delete_volume_type(cinder_client, '%s')]:" % volume_type, e
        return False


#*********************************************
#   KEYSTONE
#*********************************************
def get_tenants(keystone_client):
    try:
        tenants = keystone_client.tenants.list()
        return tenants
    except Exception, e:
        print "Error [get_tenants(keystone_client)]:", e
        return None


def get_users(keystone_client):
    try:
        users = keystone_client.users.list()
        return users
    except Exception, e:
        print "Error [get_users(keystone_client)]:", e
        return None


def get_tenant_id(keystone_client, tenant_name):
    tenants = keystone_client.tenants.list()
    id = ''
    for t in tenants:
        if t.name == tenant_name:
            id = t.id
            break
    return id


def get_user_id(keystone_client, user_name):
    users = keystone_client.users.list()
    id = ''
    for u in users:
        if u.name == user_name:
            id = u.id
            break
    return id


def get_role_id(keystone_client, role_name):
    roles = keystone_client.roles.list()
    id = ''
    for r in roles:
        if r.name == role_name:
            id = r.id
            break
    return id


def create_tenant(keystone_client, tenant_name, tenant_description):
    try:
        tenant = keystone_client.tenants.create(tenant_name,
                                                tenant_description,
                                                enabled=True)
        return tenant.id
    except Exception, e:
        print "Error [create_tenant(cinder_client, '%s', '%s')]:" % \
            (tenant_name, tenant_description), e
        return False


def create_user(keystone_client, user_name, user_password,
                user_email, tenant_id):
    try:
        user = keystone_client.users.create(user_name, user_password,
                                            user_email, tenant_id,
                                            enabled=True)
        return user.id
    except Exception, e:
        print "Error [create_user(keystone_client, '%s', '%s', '%s'" \
            "'%s')]:" % (user_name, user_password, user_email, tenant_id), e
        return False


def add_role_user(keystone_client, user_id, role_id, tenant_id):
    try:
        keystone_client.roles.add_user_role(user_id, role_id, tenant_id)
        return True
    except Exception, e:
        print "Error [add_role_user(keystone_client, '%s', '%s'" \
            "'%s')]:" % (user_id, role_id, tenant_id), e
        return False


def delete_tenant(keystone_client, tenant_id):
    try:
        tenant = keystone_client.tenants.delete(tenant_id)
        return True
    except Exception, e:
        print "Error [delete_tenant(keystone_client, '%s')]:" % tenant_id, e
        return False


def delete_user(keystone_client, user_id):
    try:
        tenant = keystone_client.users.delete(user_id)
        return True
    except Exception, e:
        print "Error [delete_user(keystone_client, '%s')]:" % user_id, e
        return False


# ----------------------------------------------------------
#
#               INTERNET UTILS
#
# -----------------------------------------------------------
def check_internet_connectivity(url='http://www.opnfv.org/'):
    """
    Check if there is access to the internet
    """
    try:
        urllib2.urlopen(url, timeout=5)
        return True
    except urllib2.URLError:
        return False


def download_url(url, dest_path):
    """
    Download a file to a destination path given a URL
    """
    name = url.rsplit('/')[-1]
    dest = dest_path + "/" + name
    try:
        response = urllib2.urlopen(url)
    except (urllib2.HTTPError, urllib2.URLError):
        return False

    with open(dest, 'wb') as f:
        shutil.copyfileobj(response, f)
    return True


# ----------------------------------------------------------
#
#               CI UTILS
#
# -----------------------------------------------------------
def get_git_branch(repo_path):
    """
    Get git branch name
    """
    repo = Repo(repo_path)
    branch = repo.active_branch
    return branch.name


def get_installer_type(logger=None):
    """
    Get installer type (fuel, apex, joid, compass)
    """
    try:
        installer = os.environ['INSTALLER_TYPE']
    except KeyError:
        if logger:
            logger.error("Impossible to retrieve the installer type")
        installer = "Unknown_installer"

    return installer


def get_scenario(logger=None):
    """
    Get scenario
    """
    try:
        scenario = os.environ['DEPLOY_SCENARIO']
    except KeyError:
        if logger:
            logger.error("Impossible to retrieve the scenario")
        scenario = "Unknown_scenario"

    return scenario


def get_pod_name(logger=None):
    """
    Get PoD Name from env variable NODE_NAME
    """
    try:
        return os.environ['NODE_NAME']
    except KeyError:
        if logger:
            logger.error(
                "Unable to retrieve the POD name from environment.Using pod name 'unknown-pod'")
        return "unknown-pod"


def push_results_to_db(db_url, project, case_name, logger, pod_name,
                       version, payload):
    """
    POST results to the Result target DB
    """
    url = db_url + "/results"
    installer = get_installer_type(logger)
    params = {"project_name": project, "case_name": case_name,
              "pod_name": pod_name, "installer": installer,
              "version": version, "details": payload}

    headers = {'Content-Type': 'application/json'}
    try:
        r = requests.post(url, data=json.dumps(params), headers=headers)
        if logger:
            logger.debug(r)
        return True
    except Exception, e:
        print "Error [push_results_to_db('%s', '%s', '%s', '%s', '%s', '%s')]:" \
            % (db_url, project, case_name, pod_name, version, payload), e
        return False


def get_resolvconf_ns():
    """
    Get nameservers from current resolv.conf
    """
    nameservers = []
    rconf = open("/etc/resolv.conf", "r")
    line = rconf.readline()
    while line:
        ip = re.search(r"\b(?:[0-9]{1,3}\.){3}[0-9]{1,3}\b", line)
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        if ip:
            result = sock.connect_ex((ip.group(), 53))
            if result == 0:
                nameservers.append(ip.group())
        line = rconf.readline()
    return nameservers


def getTestEnv(test, functest_yaml):
    """
    Get the config of the testcase based on functest_config.yaml
      2 options
        - test = test project e.g; ovno
        - test = testcase e.g. functest/odl
       look for the / to see if it is a test project or a testcase
    """
    try:
        TEST_ENV = functest_yaml.get("test-dependencies")

        if test.find("/") < 0:
            config_test = TEST_ENV[test]
        else:
            test_split = test.split("/")
            testproject = test_split[0]
            testcase = test_split[1]
            config_test = TEST_ENV[testproject][testcase]
    except KeyError:
        # if not defined in dependencies => no dependencies
        config_test = ""
    except Exception, e:
        print "Error [getTestEnv]:", e

    return config_test


def get_ci_envvars():
    """
    Get the CI env variables
    """
    ci_env_var = {
        "installer": os.environ.get('INSTALLER_TYPE'),
        "scenario": os.environ.get('DEPLOY_SCENARIO')}
    return ci_env_var


def isTestRunnable(test, functest_yaml):
    """
    Return True if the test is runnable in the current scenario
    """
    # By default we assume that all the tests are always runnable...
    is_runnable = True
    # Retrieve CI environment
    ci_env = get_ci_envvars()
    # Retrieve test environement from config file
    test_env = getTestEnv(test, functest_yaml)

    # if test_env not empty => dependencies to be checked
    if test_env is not None and len(test_env) > 0:
        # possible criteria = ["installer", "scenario"]
        # consider test criteria from config file
        # compare towards CI env through CI en variable
        for criteria in test_env:
            if re.search(test_env[criteria], ci_env[criteria]) is None:
                # print "Test "+ test + " cannot be run on the environment"
                is_runnable = False
    return is_runnable


def generateTestcaseList(functest_yaml):
    """
    Generate a test file with the runnable test according to
    the current scenario
    """
    test_list = ""
    # get testcases
    testcase_list = functest_yaml.get("test-dependencies")
    projects = testcase_list.keys()

    for project in projects:
        testcases = testcase_list[project]
        # 1 or 2 levels for testcases project[/case]l
        # if only project name without controller or scenario
        # => shall be runnable on any controller/scenario
        if testcases is None:
            test_list += project + " "
        else:
            for testcase in testcases:
                if testcase == "installer" or testcase == "scenario":
                    # project (1 level)
                    if isTestRunnable(project, functest_yaml):
                        test_list += project + " "
                else:
                    # project/testcase (2 levels)
                    thetest = project + "/" + testcase
                    if isTestRunnable(thetest, functest_yaml):
                        test_list += testcase + " "

    # sort the list to execute the test in the right order
    test_order_list = functest_yaml.get("test_exec_priority")
    test_sorted_list = ""
    for test in test_order_list:
        if test_order_list[test] in test_list:
            test_sorted_list += test_order_list[test] + " "

    # create a file that could be consumed by run-test.sh
    # this method is used only for CI
    # so it can be run only in container
    # reuse default conf directory to store the list of runnable tests
    file = open("/home/opnfv/functest/conf/testcase-list.txt", 'w')
    file.write(test_sorted_list)
    file.close()

    return test_sorted_list


def execute_command(cmd, logger=None, exit_on_error=True):
    """
    Execute Linux command
        prints stdout to a file and depending on if there
        is a logger defined, it will print it or not.
    """
    if logger:
        logger.debug('Executing command : {}'.format(cmd))
    output_file = "output.txt"
    f = open(output_file, 'w+')
    p = subprocess.call(cmd, shell=True, stdout=f, stderr=subprocess.STDOUT)
    f.close()
    f = open(output_file, 'r')
    result = f.read()
    if result != "" and logger:
        logger.debug(result)
    if p == 0:
        return True
    else:
        if logger:
            logger.error("Error when executing command %s" % cmd)
        if exit_on_error:
            exit(-1)
        return False