-#!/usr/bin/python
-#
-# Copyright (c) 2015 Orange
+# copyrighi (c) 2015 Orange
+
# morgan.richomme@orange.com
#
# This program and the accompanying materials
def __init__(self):
self.project = ''
self.testcase = ''
- self.pod_id = -1
+ self.pod_name = 'all'
self.duration = 'all'
self.version = 'all'
self.installer = 'all'
- def setCriteria(self, project, testcase, pod_id,
+ def setCriteria(self, project, testcase, pod_name,
duration, version, installer):
self.project = project
self.testcase = testcase
- self.pod_id = pod_id
+ self.pod_name = pod_name
self.duration = duration
self.version = version
self.installer = installer
return ""
else:
if(type(name) == int):
- return "-" + str(name)
+ return "_" + str(name)
else:
- return "-" + name
+ return "_" + name
def format(self):
- pod_name = self.format_criteria(self.pod_id)
+ pod_name = self.format_criteria(self.pod_name)
version_name = self.format_criteria(self.version)
installer_name = self.format_criteria(self.installer)
duration_name = self.format_criteria(self.duration)
try:
- fileName = "result-" + self.project + "-" + self.testcase + \
+ fileName = "result_" + self.project + "_" + self.testcase + \
pod_name + version_name + installer_name + \
duration_name + ".json"
except:
for pod in pods:
# cast int becase otherwise API retrieve 1.0
# TODO check format with API
- pods_table.append(int(pod['_id']))
+ pods_table.append(pod['name'])
- pods_table.append(0) # 0 means all the pods here
+ pods_table.append('all')
return pods_table
except:
print "Error retrieving the list of PODs"
def get_versions(db_url):
# retrieve the list of versions
# TODO not supported in API yet
- url = db_url + "/versions"
+ url = db_url + "/results"
# Build headers
headers = {'Content-Type': 'application/json'}
# Get result as a json object
versions_data = json.loads(db_data.text)
# Get results
- versions = versions_data['versions']
-
+ versions = versions_data['test_results']
versions_table = []
for version in versions:
- versions_table.append(version['version'])
+ if (version['version'] is not None):
+ versions_table.append(version['version'])
versions_table.append('all')
-
+ versions_table = sorted(set(versions_table))
return versions_table
except:
print "Error retrieving the list of OPNFV versions"
def get_installers(db_url):
# retrieve the list of installers
# TODO not supported in API yet
- url = db_url + "/installers"
+ url = db_url + "/results"
# Build headers
headers = {'Content-Type': 'application/json'}
# Get result as a json object
installers_data = json.loads(db_data.text)
# Get results
- installers = installers_data['installers']
+ installers = installers_data['test_results']
installers_table = []
for installer in installers:
- installers_table.append(installer['installer'])
+ if (installer['installer'] is not None):
+ installers_table.append(installer['installer'])
installers_table.append('all')
-
- return installers
+ installers_table = sorted(set(installers_table))
+ return installers_table
except:
print "Error retrieving the list of OPNFV installers"
return None
# - yardstick tests on any POD since 30 days
# - Qtip tests on dell-test1 POD
#
- # params = {"pod_id":pod_id, "testcase":testcase}
+ # params = {"pod_name":pod, "testcase":testcase}
# filter_date = days # data from now - days
- # test_project = test_criteria.project
+ test_project = test_criteria.project
testcase = test_criteria.testcase
- # duration_frame = test_criteria.duration
- # version = test_criteria.version
- # installer_type = test_criteria.installer
- pod_id = test_criteria.pod_id
-
- pod_criteria = ""
- if (pod_id > 0):
- pod_criteria = "&pod=" + str(pod_id)
+ period = test_criteria.duration
+ version = test_criteria.version
+ installer = test_criteria.installer
+ pod = test_criteria.pod_name
# TODO complete params (installer type, testcase, version )
# need API to be up to date
# we assume that criteria could be used at the API level
# no need to processing on date for instance
- params = {"pod_id": pod_id}
+ # params = {"pod_name": pod_name}
# Build headers
headers = {'Content-Type': 'application/json'}
- url = db_url + "/results?case=" + testcase + pod_criteria
+ # build the request
+ # if criteria is all => remove criteria
+ url = db_url + "/results?project=" + test_project + "&case=" + testcase
+
+ if (pod != "all"):
+ url += "&pod=" + pod
+ if (installer != "all"):
+ url += "&installer=" + installer
+ if (version != "all"):
+ url += "&version=" + version
+ url += "&period=" + str(period)
# Send Request to Test DB
- myData = requests.get(url, data=json.dumps(params), headers=headers)
+ myData = requests.get(url, headers=headers)
+
# Get result as a json object
myNewData = json.loads(myData.text)
def generateJson(test_name, test_case, db_url):
- # pod_id = 1
+ # pod_id = "opnfv-jump-1'
# test_version = 'Arno master'
# test_installer = 'fuel'
# test_retention = 30
pods = get_pods(db_url)
- versions = ['ArnoR1', 'ArnoSR1', 'all'] # not available in the API yet
- installers = ['fuel', 'foreman', 'all'] # not available in the API yet
- test_durations = [90, 365, 'all'] # not available through the API yet
+ versions = get_versions(db_url)
+ installers = get_installers(db_url)
+
+ test_durations = [90, 365, 0] # 0 means since the beginning
# For all the PoDs
for pod in pods:
# http://www.apache.org/licenses/LICENSE-2.0
#
-import re, json, os, urllib2, shutil, subprocess, sys
-############# CREDENTIALS OPENSTACK #############
+import os
+import urllib2
+import subprocess
+import sys
+from git import Repo
+
+
+# ############ CREDENTIALS OPENSTACK #############
def check_credentials():
"""
Check if the OpenStack credentials (openrc) are sourced
"""
- #TODO: there must be a short way to do this, doing if os.environ["something"] == "" throws an error
+ # TODO: there must be a short way to do this
+ # doing if os.environ["something"] == "" throws an error
try:
- os.environ['OS_AUTH_URL']
+ os.environ['OS_AUTH_URL']
except KeyError:
return False
try:
- os.environ['OS_USERNAME']
+ os.environ['OS_USERNAME']
except KeyError:
return False
try:
- os.environ['OS_PASSWORD']
+ os.environ['OS_PASSWORD']
except KeyError:
return False
try:
- os.environ['OS_TENANT_NAME']
+ os.environ['OS_TENANT_NAME']
except KeyError:
return False
return True
:param service: a string indicating the name of the service
requesting the credentials.
"""
- #TODO: get credentials from the openrc file
creds = {}
# Unfortunately, each of the OpenStack client will request slightly
# different entries in their credentials dict.
# 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"), # add your cloud username details
- password: os.environ.get("OS_PASSWORD", 'admin'), # add password
- "auth_url": os.environ.get("OS_AUTH_URL","http://192.168.20.71:5000/v2.0"), # Auth URL
+ "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_instance_status(nova_client,instance):
+# ################ NOVA #################
+def get_instance_status(nova_client, instance):
try:
instance = nova_client.servers.get(instance.id)
return instance.status
return id
-################# NEUTRON #################
+# ################ NEUTRON #################
def create_neutron_net(neutron_client, name):
json_body = {'network': {'name': name,
- 'admin_state_up': True}}
+ 'admin_state_up': True}}
try:
network = neutron_client.create_network(body=json_body)
network_dict = network['network']
print "Error:", sys.exc_info()[0]
return False
+
def delete_neutron_net(neutron_client, network_id):
try:
neutron_client.delete_network(network_id)
print "Error:", sys.exc_info()[0]
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}]}
+ 'ip_version': 4, 'network_id': net_id}]}
try:
subnet = neutron_client.create_subnet(body=json_body)
return subnet['subnets'][0]['id']
print "Error:", sys.exc_info()[0]
return False
+
def delete_neutron_subnet(neutron_client, subnet_id):
try:
neutron_client.delete_subnet(subnet_id)
print "Error:", sys.exc_info()[0]
return False
+
def create_neutron_router(neutron_client, name):
json_body = {'router': {'name': name, 'admin_state_up': True}}
try:
print "Error:", sys.exc_info()[0]
return False
+
def delete_neutron_router(neutron_client, router_id):
json_body = {'router': {'id': router_id}}
try:
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)
+ neutron_client.remove_interface_router(router=router_id,
+ body=json_body)
return True
except:
print "Error:", sys.exc_info()[0]
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}]
- }}
+ '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']
def get_network_id(neutron_client, network_name):
networks = neutron_client.list_networks()['networks']
- id = ''
+ id = ''
for n in networks:
if n['name'] == network_name:
id = n['id']
break
return id
+
def check_neutron_net(neutron_client, net_name):
for network in neutron_client.list_networks()['networks']:
- if network['name'] == net_name :
+ if network['name'] == net_name:
for subnet in network['subnets']:
return True
return False
+
def get_network_list(neutron_client):
network_list = neutron_client.list_networks()['networks']
- if len(network_list) == 0 :
+ if len(network_list) == 0:
return None
- else :
+ else:
return network_list
+
def get_external_net(neutron_client):
for network in neutron_client.list_networks()['networks']:
if network['router:external']:
return network['name']
return False
+
def update_sg_quota(neutron_client, tenant_id, sg_quota, sg_rule_quota):
json_body = {"quota": {
"security_group": sg_quota,
}}
try:
- quota = neutron_client.update_quota(tenant_id=tenant_id, body=json_body)
+ quota = neutron_client.update_quota(tenant_id=tenant_id,
+ body=json_body)
return True
except:
print "Error:", sys.exc_info()[0]
return False
-################# GLANCE #################
+# ################ GLANCE #################
+
def get_image_id(glance_client, image_name):
images = glance_client.images.list()
break
return id
+
def create_glance_image(glance_client, image_name, file_path, is_public=True):
try:
with open(file_path) as fimage:
- image = glance_client.images.create(name=image_name, is_public=is_public, disk_format="qcow2",
- container_format="bare", data=fimage)
+ image = glance_client.images.create(name=image_name,
+ is_public=is_public,
+ disk_format="qcow2",
+ container_format="bare",
+ data=fimage)
return image.id
except:
return False
-################# KEYSTONE #################
+# ################ KEYSTONE #################
def get_tenant_id(keystone_client, tenant_name):
tenants = keystone_client.tenants.list()
id = ''
break
return id
+
def get_role_id(keystone_client, role_name):
roles = keystone_client.roles.list()
id = ''
break
return id
+
def get_user_id(keystone_client, user_name):
users = keystone_client.users.list()
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)
+ tenant = keystone_client.tenants.create(tenant_name,
+ tenant_description,
+ enabled=True)
return tenant.id
except:
print "Error:", sys.exc_info()[0]
return False
+
def delete_tenant(keystone_client, tenant_id):
try:
tenant = keystone_client.tenants.delete(tenant_id)
print "Error:", sys.exc_info()[0]
return False
-def create_user(keystone_client, user_name, user_password, user_email, tenant_id):
+
+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)
+ user = keystone_client.users.create(user_name, user_password,
+ user_email, tenant_id,
+ enabled=True)
return user.id
except:
print "Error:", sys.exc_info()[0]
return False
+
def delete_user(keystone_client, user_id):
try:
tenant = keystone_client.users.delete(user_id)
print "Error:", sys.exc_info()[0]
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 False
-################# UTILS #################
+# ################ UTILS #################
def check_internet_connectivity(url='http://www.opnfv.org/'):
"""
Check if there is access to the internet
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)
+ 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 :
+ if p == 0:
return True
else:
if logger:
- logger.error("Error when executing command %s" %cmd)
+ logger.error("Error when executing command %s" % cmd)
exit(-1)
+
+
+def get_git_branch(repo_path):
+ """
+ Get git branch name
+ """
+ repo = Repo(repo_path)
+ branch = repo.active_branch
+ print branch.name
+ return branch
+
+
+def get_installer_type(logger=None):
+ """
+ Get installer type (fuel, foreman, apex, joid, compass)
+ """
+ try:
+ installer = os.environ['INSTALLER_TYPE']
+ except KeyError:
+ if logger:
+ logger.error("Impossible to retrieve the installer type")
+ installer = "Unkown"
+
+ return installer
Code Review / functest-xtesting.git / commitdiff