--- /dev/null
+#!/usr/bin/env python
+
+# Copyright (c) 2017 Huawei Technologies Co.,Ltd and others.
+#
+# 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
+
+"""
+The base class to dispatch request
+
+"""
+
+import logging
+
+from flask import request
+from flask_restful import Resource
+
+from functest.api.common import api_utils, error
+
+
+LOGGER = logging.getLogger(__name__)
+
+
+class ApiResource(Resource):
+ """ API Resource class"""
+
+ def __init__(self):
+ super(ApiResource, self).__init__()
+
+ def _post_args(self): # pylint: disable=no-self-use
+ # pylint: disable=maybe-no-member
+ """ Return action and args after parsing request """
+
+ data = request.json if request.json else {}
+ params = api_utils.change_to_str_in_dict(data)
+ action = params.get('action', request.form.get('action', ''))
+ args = params.get('args', {})
+ try:
+ args['file'] = request.files['file']
+ except KeyError:
+ pass
+ LOGGER.debug('Input args are: action: %s, args: %s', action, args)
+
+ return action, args
+
+ def _dispatch_post(self):
+ """ Dispatch request """
+ action, args = self._post_args()
+ return self._dispatch(args, action)
+
+ def _dispatch(self, args, action):
+ """
+ Dynamically load the classes with reflection and
+ obtain corresponding methods
+ """
+ try:
+ return getattr(self, action)(args)
+ except AttributeError:
+ error.result_handler(status=1, data='No such action')
+
+
+# Import modules from package "functest.api.resources"
+# and append them into sys.modules
+api_utils.import_modules_from_package("functest.api.resources")
--- /dev/null
+#!/usr/bin/env python
+
+# Copyright (c) 2017 Huawei Technologies Co.,Ltd and others.
+#
+# 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
+
+"""
+Utils for functest restapi
+
+"""
+
+import collections
+import logging
+import os
+import sys
+from oslo_utils import importutils
+
+import six
+
+import functest
+
+LOGGER = logging.getLogger(__name__)
+
+
+def change_to_str_in_dict(obj):
+ """
+ Return a dict with key and value both in string if they are in Unicode
+ """
+ if isinstance(obj, collections.Mapping):
+ return {str(k): change_to_str_in_dict(v) for k, v in obj.items()}
+ elif isinstance(obj, list):
+ return [change_to_str_in_dict(ele) for ele in obj]
+ elif isinstance(obj, six.text_type):
+ return str(obj)
+ return obj
+
+
+def itersubclasses(cls, _seen=None):
+ """ Generator over all subclasses of a given class in depth first order """
+
+ if not isinstance(cls, type):
+ raise TypeError("itersubclasses must be called with "
+ "new-style classes, not %.100r" % cls)
+ _seen = _seen or set()
+ try:
+ subs = cls.__subclasses__()
+ except TypeError: # fails only when cls is type
+ subs = cls.__subclasses__(cls)
+ for sub in subs:
+ if sub not in _seen:
+ _seen.add(sub)
+ yield sub
+ for itersub in itersubclasses(sub, _seen):
+ yield itersub
+
+
+def import_modules_from_package(package):
+ """
+ Import modules from package and append into sys.modules
+ :param: package - Full package name. For example: functest.api.resources
+ """
+ path = [os.path.dirname(functest.__file__), ".."] + package.split(".")
+ path = os.path.join(*path)
+ for root, _, files in os.walk(path):
+ for filename in files:
+ if filename.startswith("__") or not filename.endswith(".py"):
+ continue
+ new_package = ".".join(root.split(os.sep)).split("....")[1]
+ module_name = "%s.%s" % (new_package, filename[:-3])
+ try:
+ try_append_module(module_name, sys.modules)
+ except ImportError:
+ LOGGER.exception("unable to import %s", module_name)
+
+
+def try_append_module(name, modules):
+ """ Append the module into specified module system """
+
+ if name not in modules:
+ modules[name] = importutils.import_module(name)
+
+
+def change_obj_to_dict(obj):
+ """ Transfer the object into dict """
+ dic = {}
+ for key, value in vars(obj).items():
+ dic.update({key: value})
+ return dic
--- /dev/null
+#!/usr/bin/env python
+
+# Copyright (c) 2017 Huawei Technologies Co.,Ltd and others.
+#
+# 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
+
+"""
+Used to handle results
+
+"""
+
+from flask import jsonify
+
+
+def result_handler(status, data):
+ """ Return the json format of result in dict """
+ result = {
+ 'status': status,
+ 'result': data
+ }
+ return jsonify(result)
--- /dev/null
+#!/usr/bin/env python
+
+# Copyright (c) 2017 Huawei Technologies Co.,Ltd and others.
+#
+# 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
+
+"""
+Resources to handle openstack related requests
+"""
+
+from flask import jsonify
+
+from functest.api.base import ApiResource
+from functest.cli.commands.cli_os import OpenStack
+from functest.utils import openstack_utils as os_utils
+from functest.utils.constants import CONST
+
+
+class V1Creds(ApiResource):
+ """ V1Creds Resource class"""
+
+ def get(self): # pylint: disable=no-self-use
+ """ Get credentials """
+ os_utils.source_credentials(CONST.__getattribute__('openstack_creds'))
+ credentials_show = OpenStack.show_credentials()
+ return jsonify(credentials_show)
--- /dev/null
+#!/usr/bin/env python
+#
+# Copyright (c) 2017 Huawei Technologies Co.,Ltd and others.
+# 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
+
+"""
+Resources to handle environment related requests
+"""
+
+from flask import jsonify
+
+from functest.api.base import ApiResource
+from functest.cli.commands.cli_env import Env
+import functest.utils.functest_utils as ft_utils
+
+
+class V1Envs(ApiResource):
+ """ V1Envs Resource class"""
+
+ def get(self): # pylint: disable=no-self-use
+ """ Get environment """
+ environment_show = Env().show()
+ return jsonify(environment_show)
+
+ def post(self):
+ """ Used to handle post request """
+ return self._dispatch_post()
+
+ def prepare(self, args): # pylint: disable=no-self-use, unused-argument
+ """ Prepare environment """
+ ft_utils.execute_command("prepare_env start")
--- /dev/null
+#!/usr/bin/env python
+
+# Copyright (c) 2017 Huawei Technologies Co.,Ltd and others.
+#
+# 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
+
+"""
+Resources to handle testcase related requests
+"""
+
+from flask import abort, jsonify
+
+from functest.api.base import ApiResource
+from functest.api.common import api_utils
+from functest.cli.commands.cli_testcase import Testcase
+
+
+class V1Testcases(ApiResource):
+ """ V1Testcases Resource class"""
+
+ def get(self): # pylint: disable=no-self-use
+ """ GET all testcases """
+ testcases_list = Testcase().list()
+ result = {'testcases': testcases_list.split('\n')[:-1]}
+ return jsonify(result)
+
+
+class V1Testcase(ApiResource):
+ """ V1Testcase Resource class"""
+
+ def get(self, testcase_name): # pylint: disable=no-self-use
+ """ GET the info of one testcase"""
+ testcase = Testcase().show(testcase_name)
+ if not testcase:
+ abort(404, "The test case '%s' does not exist or is not supported"
+ % testcase_name)
+ testcase_info = api_utils.change_obj_to_dict(testcase)
+ dependency_dict = api_utils.change_obj_to_dict(
+ testcase_info.get('dependency'))
+ testcase_info.pop('name')
+ testcase_info.pop('dependency')
+ result = {'testcase': testcase_name}
+ result.update(testcase_info)
+ result.update({'dependency': dependency_dict})
+ return jsonify(result)
--- /dev/null
+#!/usr/bin/env python
+
+# Copyright (c) 2017 Huawei Technologies Co.,Ltd and others.
+#
+# 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
+
+"""
+Resources to handle tier related requests
+"""
+
+import re
+
+from flask import abort, jsonify
+
+from functest.api.base import ApiResource
+from functest.cli.commands.cli_tier import Tier
+
+
+class V1Tiers(ApiResource):
+ """ V1Tiers Resource class """
+
+ def get(self):
+ # pylint: disable=no-self-use
+ """ GET all tiers """
+ tiers_list = Tier().list()
+ data = re.split("[\n\t]", tiers_list)
+ data = [i.strip() for i in data if i != '']
+ data_dict = dict()
+ for i in range(len(data) / 2):
+ one_data = {data[i * 2]: data[i * 2 + 1]}
+ if i == 0:
+ data_dict = one_data
+ else:
+ data_dict.update(one_data)
+ result = {'tiers': data_dict}
+ return jsonify(result)
+
+
+class V1Tier(ApiResource):
+ """ V1Tier Resource class """
+
+ def get(self, tier_name): # pylint: disable=no-self-use
+ """ GET the info of one tier """
+ testcases = Tier().gettests(tier_name)
+ if not testcases:
+ abort(404, "The tier with name '%s' does not exist." % tier_name)
+ tier_info = Tier().show(tier_name)
+ tier_info.__dict__.pop('name')
+ tier_info.__dict__.pop('tests_array')
+ result = {'tier': tier_name, 'testcases': testcases}
+ result.update(tier_info.__dict__)
+ return jsonify(result)
+
+
+class V1TestcasesinTier(ApiResource):
+ """ V1TestcasesinTier Resource class """
+
+ def get(self, tier_name): # pylint: disable=no-self-use
+ """ GET all testcases within given tier """
+ testcases = Tier().gettests(tier_name)
+ if not testcases:
+ abort(404, "The tier with name '%s' does not exist." % tier_name)
+ result = {'tier': tier_name, 'testcases': testcases}
+ return jsonify(result)
--- /dev/null
+#!/usr/bin/env python
+
+# Copyright (c) 2017 Huawei Technologies Co.,Ltd and others.
+#
+# 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
+
+"""
+Used to launch Functest RestApi
+
+"""
+
+import logging
+import socket
+from urlparse import urljoin
+import pkg_resources
+
+from flask import Flask
+from flask_restful import Api
+
+from functest.api.base import ApiResource
+from functest.api.urls import URLPATTERNS
+from functest.api.common import api_utils
+
+
+LOGGER = logging.getLogger(__name__)
+
+
+def get_resource(resource_name):
+ """ Obtain the required resource according to resource name """
+ name = ''.join(resource_name.split('_'))
+ return next((r for r in api_utils.itersubclasses(ApiResource)
+ if r.__name__.lower() == name))
+
+
+def get_endpoint(url):
+ """ Obtain the endpoint of url """
+ address = socket.gethostbyname(socket.gethostname())
+ return urljoin('http://{}:5000'.format(address), url)
+
+
+def api_add_resource(api):
+ """
+ The resource has multiple URLs and you can pass multiple URLs to the
+ add_resource() method on the Api object. Each one will be routed to
+ your Resource
+ """
+ for url_pattern in URLPATTERNS:
+ try:
+ api.add_resource(
+ get_resource(url_pattern.target), url_pattern.url,
+ endpoint=get_endpoint(url_pattern.url))
+ except StopIteration:
+ LOGGER.error('url resource not found: %s', url_pattern.url)
+
+
+def main():
+ """Entry point"""
+ logging.config.fileConfig(pkg_resources.resource_filename(
+ 'functest', 'ci/logging.ini'))
+ LOGGER.info('Starting Functest server')
+ app = Flask(__name__)
+ api = Api(app)
+ api_add_resource(api)
+ app.run(host='0.0.0.0')
--- /dev/null
+#!/usr/bin/env python
+
+# Copyright (c) 2017 Huawei Technologies Co.,Ltd and others.
+#
+# 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
+
+"""
+Define multiple URLs
+"""
+
+
+class Url(object): # pylint: disable=too-few-public-methods
+ """ Url Class """
+
+ def __init__(self, url, target):
+ super(Url, self).__init__()
+ self.url = url
+ self.target = target
+
+
+URLPATTERNS = [
+ # GET /api/v1/functest/envs => GET environment
+ Url('/api/v1/functest/envs', 'v1_envs'),
+
+ # POST /api/v1/functest/envs/action , {"action":"prepare"}
+ # => Prepare environment
+ Url('/api/v1/functest/envs/action', 'v1_envs'),
+
+ # GET /api/v1/functest/openstack/credentials => GET credentials
+ Url('/api/v1/functest/openstack/credentials', 'v1_creds'),
+
+ # GET /api/v1/functest/testcases => GET all testcases
+ Url('/api/v1/functest/testcases', 'v1_test_cases'),
+
+ # GET /api/v1/functest/testcases/<testcase_name>
+ # => GET the info of one testcase
+ Url('/api/v1/functest/testcases/<testcase_name>', 'v1_testcase'),
+
+ # GET /api/v1/functest/testcases => GET all tiers
+ Url('/api/v1/functest/tiers', 'v1_tiers'),
+
+ # GET /api/v1/functest/tiers/<tier_name>
+ # => GET the info of one tier
+ Url('/api/v1/functest/tiers/<tier_name>', 'v1_tier'),
+
+ # GET /api/v1/functest/tiers/<tier_name>/testcases
+ # => GET all testcases within given tier
+ Url('/api/v1/functest/tiers/<tier_name>/testcases', 'v1_testcases_in_tier')
+]
api_url: http://energy.opnfv.fr/resources
api_user: ""
api_password: ""
+
[loggers]
-keys=root,functest,ci,cli,core,energy,opnfv_tests,utils
+keys=root,functest,api,ci,cli,core,energy,opnfv_tests,utils
[handlers]
keys=console,wconsole,file,null
handlers=file
qualname=functest
+[logger_api]
+level=NOTSET
+handlers=wconsole
+qualname=functest.api
+
[logger_ci]
level=NOTSET
handlers=console
import functest.utils.functest_utils as ft_utils
-class CliEnv(object):
+class Env(object):
def __init__(self):
pass
if self.status(verbose=False) == 0:
STATUS = "ready"
- msg = prettytable.PrettyTable(
- header_style='upper', padding_width=5,
- field_names=['Functest Environment', 'value'])
- msg.add_row(['INSTALLER', installer_info])
- msg.add_row(['SCENARIO', scenario])
- msg.add_row(['POD', node])
- if build_tag:
- msg.add_row(['BUILD TAG', build_tag])
- msg.add_row(['DEBUG FLAG', is_debug])
- msg.add_row(['STATUS', STATUS])
- click.echo(msg.get_string())
+ env_info = {'INSTALLER': installer_info,
+ 'SCENARIO': scenario,
+ 'POD': node,
+ 'DEBUG FLAG': is_debug,
+ 'BUILD_TAG': build_tag,
+ 'STATUS': STATUS}
+
+ return env_info
def status(self, verbose=True):
ret_val = 0
click.echo("Functest environment ready to run tests.\n")
return ret_val
+
+
+class CliEnv(Env):
+
+ def __init__(self):
+ super(CliEnv, self).__init__()
+
+ def show(self):
+ env_info = super(CliEnv, self).show()
+ msg = prettytable.PrettyTable(
+ header_style='upper', padding_width=5,
+ field_names=['Functest Environment', 'value'])
+ for key, value in env_info.iteritems():
+ if key is not None:
+ msg.add_row([key, value])
+ click.echo(msg.get_string())
import functest.utils.openstack_snapshot as os_snapshot
-class CliOpenStack(object):
+class OpenStack(object):
def __init__(self):
self.os_auth_url = CONST.__getattribute__('OS_AUTH_URL')
@staticmethod
def show_credentials():
+ dic_credentials = {}
for key, value in os.environ.items():
if key.startswith('OS_'):
- click.echo("{}={}".format(key, value))
+ dic_credentials.update({key: value})
+ return dic_credentials
def check(self):
self.ping_endpoint()
"'functest openstack snapshot-create'")
return
os_clean.main()
+
+
+class CliOpenStack(OpenStack):
+
+ def __init__(self):
+ super(CliOpenStack, self).__init__()
+
+ @staticmethod
+ def show_credentials():
+ dic_credentials = OpenStack.show_credentials()
+ for key, value in dic_credentials.items():
+ if key.startswith('OS_'):
+ click.echo("{}={}".format(key, value))
import functest.utils.functest_vacation as vacation
-class CliTestcase(object):
+class Testcase(object):
def __init__(self):
self.tiers = tb.TierBuilder(
for tier in self.tiers.get_tiers():
for test in tier.get_tests():
summary += (" %s\n" % test.get_name())
- click.echo(summary)
+ return summary
def show(self, testname):
description = self.tiers.get_test(testname)
- if description is None:
- click.echo("The test case '%s' does not exist or is not supported."
- % testname)
-
- click.echo(description)
+ return description
@staticmethod
def run(testname, noclean=False, report=False):
for test in tests:
cmd = "run_tests {}-t {}".format(flags, test)
ft_utils.execute_command(cmd)
+
+
+class CliTestcase(Testcase):
+
+ def __init__(self):
+ super(CliTestcase, self).__init__()
+
+ def list(self):
+ click.echo(super(CliTestcase, self).list())
+
+ def show(self, testname):
+ testcase_show = super(CliTestcase, self).show(testname)
+ if testcase_show:
+ click.echo(testcase_show)
+ else:
+ click.echo("The test case '%s' does not exist or is not supported."
+ % testname)
import functest.utils.functest_utils as ft_utils
-class CliTier(object):
+class Tier(object):
def __init__(self):
self.tiers = tb.TierBuilder(
% (tier.get_order(),
tier.get_name(),
tier.get_test_names()))
- click.echo(summary)
+ return summary
def show(self, tiername):
tier = self.tiers.get_tier(tiername)
if tier is None:
- tier_names = self.tiers.get_tier_names()
- click.echo("The tier with name '%s' does not exist. "
- "Available tiers are:\n %s\n" % (tiername, tier_names))
+ return None
else:
- click.echo(self.tiers.get_tier(tiername))
+ tier_info = self.tiers.get_tier(tiername)
+ return tier_info
def gettests(self, tiername):
tier = self.tiers.get_tier(tiername)
if tier is None:
- tier_names = self.tiers.get_tier_names()
- click.echo("The tier with name '%s' does not exist. "
- "Available tiers are:\n %s\n" % (tiername, tier_names))
+ return None
else:
tests = tier.get_test_names()
- click.echo("Test cases in tier '%s':\n %s\n" % (tiername, tests))
+ return tests
@staticmethod
def run(tiername, noclean=False, report=False):
else:
cmd = "run_tests {}-t {}".format(flags, tiername)
ft_utils.execute_command(cmd)
+
+
+class CliTier(Tier):
+
+ def __init__(self):
+ super(CliTier, self).__init__()
+
+ def list(self):
+ click.echo(super(CliTier, self).list())
+
+ def show(self, tiername):
+ tier_info = super(CliTier, self).show(tiername)
+ if tier_info:
+ click.echo(tier_info)
+ else:
+ tier_names = self.tiers.get_tier_names()
+ click.echo("The tier with name '%s' does not exist. "
+ "Available tiers are:\n %s\n" % (tiername, tier_names))
+
+ def gettests(self, tiername):
+ tests = super(CliTier, self).gettests(tiername)
+ if tests:
+ click.echo("Test cases in tier '%s':\n %s\n" % (tiername, tests))
+ else:
+ tier_names = self.tiers.get_tier_names()
+ click.echo("The tier with name '%s' does not exist. "
+ "Available tiers are:\n %s\n" % (tiername, tier_names))
click
openbaton-cli
cloudify_rest_client
+Flask!=0.11,<1.0,>=0.10 # BSD
+Flask-RESTful>=0.3.5 # BSD
mock>=2.0 # BSD
iniparse==0.4
PrettyTable<0.8,>=0.7.1 # BSD
prepare_env = functest.ci.prepare_env:main
run_tests = functest.ci.run_tests:main
check_deployment = functest.ci.check_deployment:main
+ functest_restapi = functest.api.server:main
basepython = python2.7
whitelist_externals = bash
modules =
+ functest.api
functest.core
functest.opnfv_tests.sdn.odl
functest.tests.unit.core
Code Review / functest-xtesting.git / commitdiff