`[15]`_: Testing OpenStack Tempest part 1
-`[16]`_: Running Functest through internal REST API
-
-`[17]`_: OPNFV Test API
+`[16]`_: OPNFV Test API
`OPNFV main site`_: OPNFV official web site
.. _`[13]`: https://wiki.opnfv.org/display/PROJ/SNAPS-OO
.. _`[14]`: https://github.com/oolorg/opnfv-functest-vrouter
.. _`[15]`: https://aptira.com/testing-openstack-tempest-part-1/
-.. _`[16]`: https://wiki.opnfv.org/display/functest/Running+test+cases+via+new+Functest+REST+API
-.. _`[17]`: http://docs.opnfv.org/en/latest/testing/testing-dev.html
+.. _`[16]`: http://docs.opnfv.org/en/latest/testing/testing-dev.html
.. _`OPNFV main site`: http://www.opnfv.org
.. _`Functest page`: https://wiki.opnfv.org/functest
.. _`OpenRC`: http://docs.openstack.org/user-guide/common/cli_set_environment_variables_using_openstack_rc.html
Reporting results to the test Database
======================================
In OPNFV CI we collect all the results from CI. A test API shall be available
-as well as a test database `[17]`_.
-
-Functest internal API
-=====================
-
-An internal API has been introduced in Euphrates. The goal is to trigger
-Functest operations through an API in addition of the CLI.
-This could be considered as a first step towards a pseudo micro services
-approach where the different test projects could expose and consume APIs to the
-other test projects.
-
-In Euphrates the main method of the APIs are:
-
- * Show credentials
- * Update openrc file
- * Show environment
- * Update hosts info for domain name
- * Prepare environment
- * List all testcases
- * Show a testcase
- * Run a testcase
- * List all tiers
- * Show a tier
- * List all testcases within given tier
- * Get the result of the specified task
- * Get the log of the specified task
-
-See `[16]`_ to get examples on how to use the API.
+as well as a test database `[16]`_.
.. _`[1]`: http://docs.opnfv.org/en/latest/submodules/functest/docs/testing/user/configguide/index.html
-.. _`[16]`: https://wiki.opnfv.org/display/functest/Running+test+cases+via+new+Functest+REST+API
-.. _`[17]`: http://docs.opnfv.org/en/latest/testing/testing-dev.html
+.. _`[16]`: http://docs.opnfv.org/en/latest/testing/testing-dev.html
These test cases can be run inside the container, using new Functest CLI as
follows::
- $ functest testcase run vping_ssh
- $ functest testcase run vping_userdata
+ $ run_tests -t vping_ssh
+ $ run_tests -t vping_userdata
The Functest CLI is designed to route a call to the corresponding internal
python scripts, located in paths::
+++ /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
-
-
-LOGGER = logging.getLogger(__name__)
-
-
-class ApiResource(Resource):
- """ API Resource class"""
-
- 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 _get_args(self): # pylint: disable=no-self-use
- """ Convert the unicode to string for request.args """
- args = api_utils.change_to_str_in_dict(request.args)
- return 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:
- api_utils.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
-
-from flask import jsonify
-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
-
-
-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
-
-"""
-Used to handle multi-thread tasks
-"""
-
-import logging
-import threading
-
-from oslo_serialization import jsonutils
-
-
-LOGGER = logging.getLogger(__name__)
-
-
-class TaskThread(threading.Thread):
- """ Task Thread Class """
-
- def __init__(self, target, args, handler):
- super(TaskThread, self).__init__(target=target, args=args)
- self.target = target
- self.args = args
- self.handler = handler
-
- def run(self):
- """ Override the function run: run testcase and update database """
- update_data = {'task_id': self.args.get('task_id'),
- 'status': 'IN PROGRESS'}
- self.handler.insert(update_data)
-
- LOGGER.info('Starting running test case')
-
- try:
- data = self.target(self.args)
- except Exception as err: # pylint: disable=broad-except
- LOGGER.exception('Task Failed')
- update_data = {'status': 'FAIL', 'error': str(err)}
- self.handler.update_attr(self.args.get('task_id'), update_data)
- else:
- LOGGER.info('Task Finished')
- LOGGER.debug('Result: %s', data)
- new_data = {'status': 'FINISHED',
- 'result': jsonutils.dumps(data.get('result', {}))}
-
- self.handler.update_attr(self.args.get('task_id'), new_data)
+++ /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
-
-"""
-Create database to store task results using sqlalchemy
-"""
-
-from sqlalchemy import create_engine
-from sqlalchemy.ext.declarative import declarative_base
-from sqlalchemy.orm import scoped_session, sessionmaker
-
-
-SQLITE = 'sqlite:////tmp/functest.db'
-
-ENGINE = create_engine(SQLITE, convert_unicode=True)
-DB_SESSION = scoped_session(sessionmaker(autocommit=False,
- autoflush=False,
- bind=ENGINE))
-BASE = declarative_base()
-BASE.query = DB_SESSION.query_property()
+++ /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 tasks: insert the task info into database and update it
-"""
-
-from functest.api.database.db import DB_SESSION
-from functest.api.database.v1.models import Tasks
-
-
-class TasksHandler(object):
- """ Tasks Handler Class """
-
- def insert(self, kwargs): # pylint: disable=no-self-use
- """ To insert the task info into database """
- task = Tasks(**kwargs)
- DB_SESSION.add(task) # pylint: disable=maybe-no-member
- DB_SESSION.commit() # pylint: disable=maybe-no-member
- return task
-
- def get_task_by_taskid(self, task_id): # pylint: disable=no-self-use
- """ Obtain the task by task id """
- # pylint: disable=maybe-no-member
- task = Tasks.query.filter_by(task_id=task_id).first()
- if not task:
- raise ValueError
-
- return task
-
- def update_attr(self, task_id, attr):
- """ Update the required attributes of the task """
- task = self.get_task_by_taskid(task_id)
-
- for key, value in attr.items():
- setattr(task, key, value)
- DB_SESSION.commit() # pylint: disable=maybe-no-member
+++ /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 tables for tasks
-"""
-
-from sqlalchemy import Column
-from sqlalchemy import Integer
-from sqlalchemy import String
-from sqlalchemy import Text
-
-from functest.api.database.db import BASE
-
-
-class Tasks(BASE): # pylint: disable=too-few-public-methods, no-init
- """ Create a table for tasks"""
-
- __tablename__ = 'tasks'
- id = Column(Integer, primary_key=True) # pylint: disable=invalid-name
- task_id = Column(String(50))
- status = Column(Integer)
- error = Column(String(120))
- result = Column(Text)
-
- def __repr__(self):
- return '<Task %r>' % Tasks.task_id
+++ /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
-"""
-
-import collections
-import logging
-import socket
-
-from flask import jsonify
-from flasgger.utils import swag_from
-import pkg_resources
-from xtesting.ci import run_tests
-
-from functest.api.base import ApiResource
-from functest.api.common import api_utils
-from functest.cli.commands.cli_os import OpenStack
-from functest.utils import constants
-
-LOGGER = logging.getLogger(__name__)
-
-ADDRESS = socket.gethostbyname(socket.gethostname())
-ENDPOINT_CREDS = ('http://{}:5000/api/v1/functest/openstack'.format(ADDRESS))
-
-
-class V1Creds(ApiResource):
- """ V1Creds Resource class"""
-
- @swag_from(
- pkg_resources.resource_filename('functest', 'api/swagger/creds.yaml'),
- endpoint='{0}/credentials'.format(ENDPOINT_CREDS))
- def get(self): # pylint: disable=no-self-use
- """ Get credentials """
- run_tests.Runner.source_envfile(constants.ENV_FILE)
- credentials_show = OpenStack.show_credentials()
- return jsonify(credentials_show)
-
- @swag_from(
- pkg_resources.resource_filename('functest',
- 'api/swagger/creds_action.yaml'),
- endpoint='{0}/action'.format(ENDPOINT_CREDS))
- def post(self):
- """ Used to handle post request """
- return self._dispatch_post()
-
- def update_openrc(self, args): # pylint: disable=no-self-use
- """ Used to update the OpenStack RC file """
- try:
- openrc_vars = args['openrc']
- except KeyError:
- return api_utils.result_handler(
- status=0, data='openrc must be provided')
- else:
- if not isinstance(openrc_vars, collections.Mapping):
- return api_utils.result_handler(
- status=0, data='args should be a dict')
-
- lines = ['export {}={}\n'.format(k, v) for k, v in openrc_vars.items()]
-
- rc_file = constants.ENV_FILE
- with open(rc_file, 'w') as creds_file:
- creds_file.writelines(lines)
-
- LOGGER.info("Sourcing the OpenStack RC file...")
- try:
- run_tests.Runner.source_envfile(rc_file)
- except Exception as err: # pylint: disable=broad-except
- LOGGER.exception('Failed to source the OpenStack RC file')
- return api_utils.result_handler(status=0, data=str(err))
-
- return api_utils.result_handler(
- status=0, data='Update openrc successfully')
+++ /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
-"""
-
-import socket
-
-import IPy
-from flask import jsonify
-from flasgger.utils import swag_from
-import pkg_resources
-
-from functest.api.base import ApiResource
-from functest.api.common import api_utils
-from functest.cli.commands.cli_env import Env
-
-ADDRESS = socket.gethostbyname(socket.gethostname())
-ENDPOINT_ENVS = ('http://{}:5000/api/v1/functest/envs'.format(ADDRESS))
-
-
-class V1Envs(ApiResource):
- """ V1Envs Resource class"""
-
- @swag_from(
- pkg_resources.resource_filename('functest', 'api/swagger/envs.yaml'),
- endpoint=ENDPOINT_ENVS)
- def get(self): # pylint: disable=no-self-use
- """ Get environment """
- environment_show = Env().show()
- return jsonify(environment_show)
-
- @swag_from(
- pkg_resources.resource_filename('functest',
- 'api/swagger/envs_action.yaml'),
- endpoint='{0}/action'.format(ENDPOINT_ENVS))
- def post(self):
- """ Used to handle post request """
- return self._dispatch_post()
-
- def update_hosts(self, hosts_info): # pylint: disable=no-self-use
- """ Update hosts info """
-
- if not isinstance(hosts_info, dict):
- return api_utils.result_handler(
- status=1, data='Error, args should be a dict')
-
- for key, value in hosts_info.items():
- if key:
- try:
- IPy.IP(value)
- except Exception: # pylint: disable=broad-except
- return api_utils.result_handler(
- status=1, data='The IP %s is invalid' % value)
- else:
- return api_utils.result_handler(
- status=1, data='Domain name is absent')
-
- try:
- functest_flag = "# SUT hosts info for Functest"
- hosts_list = ('\n{} {} {}'.format(ip, host_name, functest_flag)
- for host_name, ip in hosts_info.items())
-
- with open("/etc/hosts", 'r') as file_hosts:
- origin_lines = [line for line in file_hosts
- if functest_flag not in line]
-
- with open("/etc/hosts", 'w') as file_hosts:
- file_hosts.writelines(origin_lines)
- file_hosts.write(functest_flag)
- file_hosts.writelines(hosts_list)
- except Exception: # pylint: disable=broad-except
- return api_utils.result_handler(
- status=1, data='Error when updating hosts info')
- else:
- return api_utils.result_handler(
- status=0, data='Update hosts info successfully')
+++ /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 retrieve the task results
-"""
-
-import errno
-import json
-import logging
-import os
-import uuid
-
-from flask import jsonify
-from flasgger.utils import swag_from
-import pkg_resources
-
-from functest.api.base import ApiResource
-from functest.api.common import api_utils
-from functest.api.database.v1.handlers import TasksHandler
-from functest.utils import config
-
-
-LOGGER = logging.getLogger(__name__)
-
-
-class V1Task(ApiResource):
- """ V1Task Resource class"""
-
- @swag_from(pkg_resources.resource_filename(
- 'functest', 'api/swagger/task.yaml'))
- def get(self, task_id): # pylint: disable=no-self-use
- """ GET the result of the task id """
- try:
- uuid.UUID(task_id)
- except ValueError:
- return api_utils.result_handler(status=1, data='Invalid task id')
-
- task_handler = TasksHandler()
- try:
- task = task_handler.get_task_by_taskid(task_id)
- except ValueError:
- return api_utils.result_handler(status=1, data='No such task id')
-
- status = task.status
- LOGGER.debug('Task status is: %s', status)
-
- if status not in ['IN PROGRESS', 'FAIL', 'FINISHED']:
- return api_utils.result_handler(status=1,
- data='internal server error')
-
- switcher = {'IN PROGRESS': 0, 'FAIL': 1, 'FINISHED': 2}
- if status == 'IN PROGRESS':
- result = {'status': switcher.get(status), 'result': ''}
- elif status == 'FAIL':
- result = {'status': switcher.get(status), 'error': task.error}
- else:
- result = {'status': switcher.get(status),
- 'result': json.loads(task.result)}
-
- return jsonify(result)
-
-
-class V1TaskLog(ApiResource):
- """ V1TaskLog Resource class"""
-
- @swag_from(pkg_resources.resource_filename(
- 'functest', 'api/swagger/task_log.yaml'))
- def get(self, task_id): # pylint: disable=no-self-use
- """ GET the log of the task id """
- try:
- uuid.UUID(task_id)
- except ValueError:
- return api_utils.result_handler(status=1, data='Invalid task id')
-
- task_handler = TasksHandler()
- try:
- task = task_handler.get_task_by_taskid(task_id)
- except ValueError:
- return api_utils.result_handler(status=1, data='No such task id')
-
- task_log_dir = getattr(config.CONF, 'dir_results')
- # pylint: disable=maybe-no-member
- index = int(self._get_args().get('index', 0))
-
- try:
- with open(os.path.join(task_log_dir,
- '{}.log'.format(task_id)), 'r') as log_file:
- log_file.seek(index)
- data = log_file.readlines()
- index = log_file.tell()
- except OSError as err:
- if err.errno == errno.ENOENT:
- return api_utils.result_handler(
- status=1, data='Log file does not exist')
-
- return api_utils.result_handler(
- status=1, data='Error with log file')
-
- return_data = {'data': data, 'index': index}
-
- switcher = {'IN PROGRESS': 0, 'FAIL': 1, 'FINISHED': 2}
-
- return api_utils.result_handler(status=switcher.get(task.status),
- data=return_data)
+++ /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
-"""
-
-import logging
-import os
-import re
-import socket
-import uuid
-
-from flask import jsonify
-from flasgger.utils import swag_from
-import pkg_resources
-from six.moves import configparser
-
-from functest.api.base import ApiResource
-from functest.api.common import api_utils, thread
-from functest.cli.commands.cli_testcase import Testcase
-from functest.api.database.v1.handlers import TasksHandler
-from functest.utils import config
-from functest.utils import env
-import functest.utils.functest_utils as ft_utils
-
-LOGGER = logging.getLogger(__name__)
-
-ADDRESS = socket.gethostbyname(socket.gethostname())
-ENDPOINT_TESTCASES = ('http://{}:5000/api/v1/functest/testcases'
- .format(ADDRESS))
-
-
-class V1Testcases(ApiResource):
- """ V1Testcases Resource class"""
-
- @swag_from(pkg_resources.resource_filename(
- 'functest', 'api/swagger/testcases.yaml'))
- def get(self): # pylint: disable=no-self-use
- """ GET all testcases """
- testcases_list = Testcase().list()
- result = {'testcases': re.split(' |\n ', testcases_list)[1:]}
- return jsonify(result)
-
-
-class V1Testcase(ApiResource):
- """ V1Testcase Resource class"""
-
- @swag_from(
- pkg_resources.resource_filename('functest',
- 'api/swagger/testcase.yaml'),
- endpoint='{0}/<testcase_name>'.format(ENDPOINT_TESTCASES))
- def get(self, testcase_name): # pylint: disable=no-self-use
- """ GET the info of one testcase"""
- testcase = Testcase().show(testcase_name)
- if not testcase:
- return api_utils.result_handler(
- status=1,
- data="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)
-
- @swag_from(
- pkg_resources.resource_filename('functest',
- 'api/swagger/testcase_run.yaml'),
- endpoint='{0}/action'.format(ENDPOINT_TESTCASES))
- def post(self):
- """ Used to handle post request """
- return self._dispatch_post()
-
- def run_test_case(self, args):
- """ Run a testcase """
- try:
- case_name = args['testcase']
- except KeyError:
- return api_utils.result_handler(
- status=1, data='testcase name must be provided')
-
- testcase = Testcase().show(case_name)
- if not testcase:
- return api_utils.result_handler(
- status=1,
- data="The test case '%s' does not exist or is not supported"
- % case_name)
-
- task_id = str(uuid.uuid4())
-
- task_args = {'testcase': case_name, 'task_id': task_id}
-
- task_args.update(args.get('opts', {}))
-
- task_thread = thread.TaskThread(self._run, task_args, TasksHandler())
- task_thread.start()
-
- result = {'testcase': case_name, 'task_id': task_id}
- return jsonify({'result': result})
-
- def _run(self, args): # pylint: disable=no-self-use
- """ The built_in function to run a test case """
-
- case_name = args.get('testcase')
- self._update_logging_ini(args.get('task_id'))
-
- try:
- cmd = "run_tests -t {}".format(case_name)
- runner = ft_utils.execute_command(cmd)
- except Exception: # pylint: disable=broad-except
- result = 'FAIL'
- LOGGER.exception("Running test case %s failed!", case_name)
- if runner == os.EX_OK:
- result = 'PASS'
- else:
- result = 'FAIL'
-
- env_info = {
- 'installer': env.get('INSTALLER_TYPE'),
- 'scenario': env.get('DEPLOY_SCENARIO'),
- 'build_tag': env.get('BUILD_TAG'),
- 'ci_loop': env.get('CI_LOOP')
- }
- result = {
- 'task_id': args.get('task_id'),
- 'testcase': case_name,
- 'env_info': env_info,
- 'result': result
- }
-
- return {'result': result}
-
- def _update_logging_ini(self, task_id): # pylint: disable=no-self-use
- """ Update the log file for each task"""
- rconfig = configparser.RawConfigParser()
- rconfig.read(
- pkg_resources.resource_filename('xtesting', 'ci/logging.ini'))
- log_path = os.path.join(getattr(config.CONF, 'dir_results'),
- '{}.log'.format(task_id))
- rconfig.set('handler_file', 'args', '("{}",)'.format(log_path))
-
- with open(
- pkg_resources.resource_filename(
- 'xtesting', 'ci/logging.ini'), 'wb') as configfile:
- rconfig.write(configfile)
+++ /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 jsonify
-from flasgger.utils import swag_from
-import pkg_resources
-
-from functest.api.base import ApiResource
-from functest.api.common import api_utils
-from functest.cli.commands.cli_tier import Tier
-
-
-class V1Tiers(ApiResource):
- """ V1Tiers Resource class """
-
- @swag_from(pkg_resources.resource_filename(
- 'functest', 'api/swagger/tiers.yaml'))
- 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].lstrip('- ').rstrip(':'): 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 """
-
- @swag_from(pkg_resources.resource_filename(
- 'functest', 'api/swagger/tier.yaml'))
- def get(self, tier_name): # pylint: disable=no-self-use
- """ GET the info of one tier """
- tier_info = Tier().show(tier_name)
- if not tier_info:
- return api_utils.result_handler(
- status=1,
- data="The tier with name '%s' does not exist." % tier_name)
- tier_info.__dict__.pop('name')
- tier_info.__dict__.pop('tests_array')
- tier_info.__dict__.pop('skipped_tests_array')
- testcases = Tier().gettests(tier_name)
- result = {'tier': tier_name, 'testcases': testcases}
- result.update(tier_info.__dict__)
- return jsonify(result)
-
-
-class V1TestcasesinTier(ApiResource):
- """ V1TestcasesinTier Resource class """
-
- @swag_from(pkg_resources.resource_filename(
- 'functest', 'api/swagger/testcases_in_tier.yaml'))
- def get(self, tier_name): # pylint: disable=no-self-use
- """ GET all testcases within given tier """
- tier_info = Tier().show(tier_name)
- if not tier_info:
- return api_utils.result_handler(
- status=1,
- data="The tier with name '%s' does not exist." % tier_name)
- testcases = Tier().gettests(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 inspect
-import logging
-import socket
-import pkg_resources
-
-from flask import Flask
-from flask_restful import Api
-from flasgger import Swagger
-import six
-
-from functest.api.base import ApiResource
-from functest.api.common import api_utils
-from functest.api.database.db import BASE
-from functest.api.database.db import DB_SESSION
-from functest.api.database.db import ENGINE
-from functest.api.database.v1 import models
-from functest.api.urls import URLPATTERNS
-
-
-LOGGER = logging.getLogger(__name__)
-
-APP = Flask(__name__)
-API = Api(APP)
-Swagger(APP)
-
-
-@APP.teardown_request
-def shutdown_session(exception=None): # pylint: disable=unused-argument
- """
- To be called at the end of each request whether it is successful
- or an exception is raised
- """
- DB_SESSION.remove()
-
-
-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 six.moves.urllib.parse.urljoin(
- 'http://{}:5000'.format(address), url)
-
-
-def api_add_resource():
- """
- 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 init_db():
- """
- Import all modules here that might define models so that
- they will be registered properly on the metadata, and then
- create a database
- """
- def func(subcls):
- """ To check the subclasses of BASE"""
- try:
- if issubclass(subcls[1], BASE):
- return True
- except TypeError:
- pass
- return False
- # pylint: disable=bad-option-value,bad-builtin,
- subclses = filter(func, inspect.getmembers(models, inspect.isclass))
- LOGGER.debug('Import models: %s', [subcls[1] for subcls in subclses])
- BASE.metadata.create_all(bind=ENGINE)
-
-
-def main():
- """Entry point"""
- logging.config.fileConfig(pkg_resources.resource_filename(
- 'functest', 'ci/logging.ini'))
- logging.captureWarnings(True)
- LOGGER.info('Starting Functest server')
- api_add_resource()
- init_db()
- APP.run(host='0.0.0.0')
+++ /dev/null
----
-Show credentials
-
-This api offers the interface to show credentials.
-The credentials dict will be returned.
----
-tags:
- - Creds
-definitions:
- Credentials:
- type: object
- properties:
- creds_name:
- $ref: '#/definitions/Name'
- Name:
- type: dict
-responses:
- 200:
- description: Show credentials
- schema:
- $ref: '#/definitions/Credentials'
- examples:
- "OS_AUTH_URL": "https://192.16.1.222:5000/v3"
- "OS_AUTH_VERSION": "3"
- "OS_CACERT": "/home/opnfv/functest/conf/os_cacert"
- "OS_IDENTITY_API_VERSION": "3"
- "OS_INTERFACE": "public"
- "OS_NO_CACHE": "1"
- "OS_PASSWORD": "990232e0885da343ac805528522d"
- "OS_PROJECT_DOMAIN_NAME": "Default"
- "OS_PROJECT_NAME": "admin"
- "OS_REGION_NAME": "RegionOne"
- "OS_TENANT_NAME": "admin"
- "OS_USERNAME": "admin"
- "OS_USER_DOMAIN_NAME": "Default"
+++ /dev/null
----
-Update openrc
-
-This api offers the interface to Update openstack.creds.
----
-tags:
- - Creds
-parameters:
- - in: body
- name: body
- description: this is the input json dict
- schema:
- required:
- - action
- - args
- properties:
- action:
- type: string
- description: this is action for creds
- default: update_openrc
- args:
- schema:
- required:
- - openrc
- properties:
- openrc:
- type: string
- description: this is the test case name
- default:
- "OS_AUTH_URL": "http://192.16.1.222:5000/v3"
- "OS_IDENTITY_API_VERSION": "3"
- "OS_INTERFACE": "public"
- "OS_PASSWORD": "admn"
- "OS_PROJECT_DOMAIN_NAME": "Default"
- "OS_PROJECT_NAME": "admin"
- "OS_REGION_NAME": "RegionOne"
- "OS_TENANT_NAME": "admin"
- "OS_USERNAME": "admin"
- "OS_USER_DOMAIN_NAME": "Default"
-definitions:
- Credentials:
- type: object
- properties:
- creds_name:
- $ref: '#/definitions/Name'
- Name:
- type: dict
-responses:
- 200:
- description: Update openrc
- schema:
- $ref: '#/definitions/Credentials'
- examples:
- 'status': 0
- 'result': 'Update openrc successfully'
+++ /dev/null
----
-Show environment
-
-This api offers the interface to show environment.
-The environment dict will be returned.
----
-tags:
- - Envs
-definitions:
- Environment:
- type: object
- properties:
- creds_name:
- $ref: '#/definitions/Name'
- Name:
- type: dict
-responses:
- 200:
- description: Show environment
- schema:
- $ref: '#/definitions/Environment'
- examples:
- "DEBUG FLAG": "false"
- "INSTALLER": "compass, 192.168.200.2"
- "POD": "unknown_pod"
- "SCENARIO": "os-nosdn-nofeature-noha"
- "STATUS": "ready"
+++ /dev/null
----
-Update hosts info
-
-This api offers the interface to update hosts info.
----
-tags:
- - Envs
-parameters:
- - in: body
- name: body
- description: this is the input json dict
- schema:
- required:
- - action
- - args
- properties:
- action:
- type: string
- description: this is action for envs
- default: update_hosts
- args:
- type: string
- description: Hosts info to be updated
- default:
- "identity.ac.dz.com": "8.20.11.22"
- "image.ac.dz.com": "8.20.11.22"
-definitions:
- Environment:
- type: object
- properties:
- creds_name:
- $ref: '#/definitions/Name'
- Name:
- type: dict
-responses:
- 200:
- description: Update hosts info
- schema:
- $ref: '#/definitions/Environment'
- examples:
- 'status': 0
- 'result': 'Update hosts info successfully'
+++ /dev/null
----
-Get the result of the specified task
-
-This api offers the interface to get the result of the specified task.
----
-tags:
- - Tasks
-parameters:
- - name: task_id
- description: task id
- in: path
- type: string
- required: true
-definitions:
- Task:
- type: object
- properties:
- creds_name:
- $ref: '#/definitions/Result'
- Result:
- type: dict
-responses:
- 200:
- description: Get the result of the specified task
- schema:
- $ref: '#/definitions/Task'
- examples:
- "result": {
- "case_name": "vping_ssh",
- "env_info": {
- "build_tag": null,
- "ci_loop": "weekly",
- "installer": "compass",
- "scenario": "os-nosdn-nofeature-noha"},
- "result": "PASS",
- "task_id": "1a9f3c5d-ce0b-4354-862e-dd08b26fc484"}
- "status": 2
+++ /dev/null
----
-Get the log of the specified task
-
-This api offers the interface to get the log of the specified task.
----
-tags:
- - Tasks
-parameters:
- - name: task_id
- description: task id
- in: path
- type: string
- required: true
-definitions:
- Task:
- type: object
- properties:
- creds_name:
- $ref: '#/definitions/Result'
- Result:
- type: dict
-responses:
- 200:
- description: Get the log of the specified task
- schema:
- $ref: '#/definitions/Task'
- examples:
- "result": {
- "data": [
- "2017-09-14 06:46:26,106 - functest.ci.run_tests - DEBUG ",
- "- Sourcing the OpenStack RC file... ",
- "2017-09-14 06:46:26,107 - functest.ci.run_tests - DEBUG ",
- "- Test args: connection_check ",
- "2017-09-14 06:46:26,107 - functest.ci.run_tests - INFO ",
- "- Running test case 'connection_check'... ",
- "..."]}
- "status": 2
+++ /dev/null
----
-Show the info of one testcase
-
-This api offers the interface to show the detailed info of one testcase.
-The info of one testcase will be returned.
----
-tags:
- - Testcases
-parameters:
- - name: testcase_name
- description: testcase name
- in: path
- type: string
- required: true
-definitions:
- Testcases:
- type: object
- properties:
- case_name:
- $ref: '#/definitions/Tests'
- Tests:
- type: dict
-responses:
- 200:
- description: Show the detailed info of one testcase
- schema:
- $ref: '#/definitions/Testcases'
- examples:
- "testcase": ""
- "blocking":
- "criteria":
- "dependency": {
- "installer": "",
- "scenario": ""}
- "description": ""
- "enabled":
+++ /dev/null
----
-Run a test case
-
-This api offers the interface to run a test case
----
-tags:
- - Testcases
-parameters:
- - in: body
- name: body
- description: this is the input json dict
- schema:
- required:
- - action
- - args
- properties:
- action:
- type: string
- description: this is action for creds
- default: run_test_case
- args:
- schema:
- required:
- - testcase
- properties:
- testcase:
- type: string
- description: this is the test case name
- default:
- vping_ssh
-definitions:
- Testcases:
- type: object
- properties:
- creds_name:
- $ref: '#/definitions/Tests'
- Tests:
- type: dict
-responses:
- 200:
- description: Run a test case
- schema:
- $ref: '#/definitions/Testcases'
- examples:
- 'task_id': '94c8ec94-d873-466f-a205-bf592f31ff5b'
- 'testcase': 'vping_ssh'
+++ /dev/null
----
-List all test cases
-
-This api offers the interface to list all test cases
-The testcases list will be returned
----
-tags:
- - Testcases
-definitions:
- Testcases:
- type: object
- properties:
- creds_name:
- $ref: '#/definitions/Tests'
- Tests:
- type: dict
-responses:
- 200:
- description: List all test cases
- schema:
- $ref: '#/definitions/Testcases'
- examples:
- "testcases": []
+++ /dev/null
----
-List all testcases within given tier
-
-This api offers the interface to list all testcases within given tier.
-All testcases within given tier will be returned.
----
-tags:
- - Tiers
-parameters:
- - name: tier_name
- description: tier name
- in: path
- type: string
- required: true
-definitions:
- Testcases:
- type: object
- properties:
- creds_name:
- $ref: '#/definitions/Tests'
- Tests:
- type: dict
-responses:
- 200:
- description: List all testcases within given tier
- schema:
- $ref: '#/definitions/Testcases'
- examples:
- "tier": ""
- "testcases": []
+++ /dev/null
----
-Show the info of one tier
-
-This api offers the interface to show the detailed info of one tier.
-The info of one tier will be returned.
----
-tags:
- - Tiers
-parameters:
- - name: tier_name
- description: tier name
- in: path
- type: string
- required: true
-definitions:
- Tiers:
- type: object
- properties:
- creds_name:
- $ref: '#/definitions/Name'
- Name:
- type: string
-responses:
- 200:
- description: Show the detailed info of one tier
- schema:
- $ref: '#/definitions/Tiers'
- examples:
- "tier": ""
- "ci_loop": ""
- "description": ""
- "order":
- "testcases": []
+++ /dev/null
----
-List all tiers
-
-This api offers the interface to list all tiers.
-The tiers list will be returned.
----
-tags:
- - Tiers
-definitions:
- Tiers:
- type: object
- properties:
- creds_name:
- $ref: '#/definitions/Tier'
- Tier:
- type: dict
-responses:
- 200:
- description: List all tiers
- schema:
- $ref: '#/definitions/Tiers'
- examples:
- "tiers": ""
+++ /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":"update_hosts", "args": {}} => Update hosts info
- Url('/api/v1/functest/envs/action', 'v1_envs'),
-
- # GET /api/v1/functest/openstack/credentials => GET credentials
- Url('/api/v1/functest/openstack/credentials', 'v1_creds'),
-
- # POST /api/v1/functest/openstack/action
- # {"action":"update_openrc", "args": {"openrc": {}}} => Update openrc
- Url('/api/v1/functest/openstack/action', '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'),
-
- # POST /api/v1/functest/testcases/action
- # {"action":"run_test_case", "args": {"opts": {}, "testcase": "vping_ssh"}}
- # => Run a testcase
- Url('/api/v1/functest/testcases/action', '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'),
-
- # GET /api/v1/functest/tasks/<task_id>
- # => GET the result of the task id
- Url('/api/v1/functest/tasks/<task_id>', 'v1_task'),
-
- # GET /api/v1/functest/tasks/<task_id>/log
- # => GET the log of the task
- Url('/api/v1/functest/tasks/<task_id>/log', 'v1_task_log')
-]
+++ /dev/null
-#!/usr/bin/env python
-#
-# jose.lausuch@ericsson.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
-
-# pylint: disable=missing-docstring
-
-import logging.config
-import pkg_resources
-
-import click
-
-from functest.cli.commands.cli_env import CliEnv
-from functest.cli.commands.cli_os import CliOpenStack
-from functest.cli.commands.cli_testcase import CliTestcase
-from functest.cli.commands.cli_tier import CliTier
-
-
-CONTEXT_SETTINGS = dict(help_option_names=['-h', '--help'])
-
-
-@click.group(context_settings=CONTEXT_SETTINGS)
-@click.version_option(version='opnfv colorado.0.1 ')
-def cli():
- logging.config.fileConfig(pkg_resources.resource_filename(
- 'functest', 'ci/logging.ini'))
- logging.captureWarnings(True)
-
-
-ENV = CliEnv()
-OPENSTACK = CliOpenStack()
-TESTCASE = CliTestcase()
-TIER = CliTier()
-
-
-@cli.group()
-@click.pass_context
-def env(ctx): # pylint: disable=unused-argument
- pass
-
-
-@cli.group()
-@click.pass_context
-def openstack(ctx): # pylint: disable=unused-argument
- pass
-
-
-@cli.group()
-@click.pass_context
-def testcase(ctx): # pylint: disable=unused-argument
- pass
-
-
-@cli.group()
-@click.pass_context
-def tier(ctx): # pylint: disable=unused-argument
- pass
-
-
-@openstack.command('check', help="Checks connectivity and status "
- "to the OpenStack deployment.")
-def os_check():
- OPENSTACK.check()
-
-
-@openstack.command('show-credentials',
- help="Prints the OpenStack credentials.")
-def os_show_credentials():
- OPENSTACK.show_credentials()
-
-
-@env.command('show', help="Shows information about the current environment.")
-def env_show():
- ENV.show()
-
-
-@testcase.command('list', help="Lists the available testcases.")
-def testcase_list():
- TESTCASE.list()
-
-
-@testcase.command('show', help="Shows information about a test case.")
-@click.argument('testname', type=click.STRING, required=True)
-def testcase_show(testname):
- TESTCASE.show(testname)
-
-
-@testcase.command('run', help="Executes a test case.")
-@click.argument('testname', type=click.STRING, required=True)
-@click.option('-n', '--noclean', is_flag=True, default=False,
- help='The created openstack resources by the test'
- 'will not be cleaned after the execution.')
-@click.option('-r', '--report', is_flag=True, default=False,
- help='Push results to the results DataBase. Only CI Pods'
- 'have rights to do that.')
-def testcase_run(testname, noclean, report):
- TESTCASE.run(testname, noclean, report)
-
-
-@tier.command('list', help="Lists the available tiers.")
-def tier_list():
- TIER.list()
-
-
-@tier.command('show', help="Shows information about a tier.")
-@click.argument('tiername', type=click.STRING, required=True)
-def tier_show(tiername):
- TIER.show(tiername)
-
-
-@tier.command('get-tests', help="Prints the tests in a tier.")
-@click.argument('tiername', type=click.STRING, required=True)
-def tier_gettests(tiername):
- TIER.gettests(tiername)
-
-
-@tier.command('run', help="Executes all the tests within a tier.")
-@click.argument('tiername', type=click.STRING, required=True)
-@click.option('-n', '--noclean', is_flag=True, default=False,
- help='The created openstack resources by the tests'
- 'will not be cleaned after the execution.')
-@click.option('-r', '--report', is_flag=True, default=False,
- help='Push results to the results DataBase. Only CI Pods'
- 'have rights to do that.')
-def tier_run(tiername, noclean, report):
- TIER.run(tiername, noclean, report)
+++ /dev/null
-#!/usr/bin/env python
-#
-# jose.lausuch@ericsson.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
-
-# pylint: disable=missing-docstring
-
-import click
-import prettytable
-import six
-
-from functest.utils import env
-
-
-class Env(object): # pylint: disable=too-few-public-methods
-
- @staticmethod
- def show():
- install_type = env.get('INSTALLER_TYPE')
- scenario = env.get('DEPLOY_SCENARIO')
- node = env.get('NODE_NAME')
- build_tag = env.get('BUILD_TAG')
- if build_tag:
- build_tag = build_tag.lstrip(
- "jenkins-").lstrip("functest").lstrip("-")
-
- env_info = {'INSTALLER': install_type,
- 'SCENARIO': scenario,
- 'POD': node,
- 'BUILD_TAG': build_tag}
-
- return env_info
-
-
-class CliEnv(object): # pylint: disable=too-few-public-methods
-
- @staticmethod
- def show():
- env_info = Env.show()
- msg = prettytable.PrettyTable(
- header_style='upper', padding_width=5,
- field_names=['Functest Environment', 'value'])
- for key, value in six.iteritems(env_info):
- if key is not None:
- msg.add_row([key, value])
- click.echo(msg.get_string())
+++ /dev/null
-#!/usr/bin/env python
-#
-# jose.lausuch@ericsson.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
-
-# pylint: disable=missing-docstring
-
-import os
-
-import click
-from six.moves import urllib
-
-from functest.ci import check_deployment
-from functest.utils import constants
-
-
-class OpenStack(object):
-
- def __init__(self):
- self.os_auth_url = os.environ.get('OS_AUTH_URL', None)
- self.endpoint_ip = None
- self.endpoint_port = None
- self.openstack_creds = constants.ENV_FILE
- if self.os_auth_url:
- self.endpoint_ip = urllib.parse.urlparse(self.os_auth_url).hostname
- self.endpoint_port = urllib.parse.urlparse(self.os_auth_url).port
-
- def ping_endpoint(self):
- if self.os_auth_url is None:
- click.echo("Source the OpenStack credentials first")
- exit(0)
- response = os.system("ping -c 1 " + self.endpoint_ip + ">/dev/null")
- if response == 0:
- return 0
- else:
- click.echo("Cannot talk to the endpoint %s\n" % self.endpoint_ip)
- exit(0)
-
- @staticmethod
- def show_credentials():
- dic_credentials = {}
- for key, value in os.environ.items():
- if key.startswith('OS_'):
- dic_credentials.update({key: value})
- return dic_credentials
-
- def check(self):
- self.ping_endpoint()
- deployment = check_deployment.CheckDeployment()
- deployment.check_all()
-
-
-class CliOpenStack(OpenStack):
-
- @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))
+++ /dev/null
-#!/usr/bin/env python
-#
-# jose.lausuch@ericsson.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
-
-# pylint: disable=missing-docstring
-
-import click
-
-from functest.cli.commands import cli_tier
-from functest.utils import functest_utils
-
-
-class Testcase(cli_tier.Tier):
-
- def list(self):
- summary = ""
- for tier in self.tiers.get_tiers():
- for test in tier.get_tests():
- summary += (" %s\n" % test.get_name())
- return summary
-
- def show(self, name):
- description = self.tiers.get_test(name)
- return description
-
- @staticmethod
- def run(name, noclean=False, report=False):
- tests = name.split(",")
- for test in tests:
- cmd = "run_tests {}-t {}".format(
- Testcase.get_flags(noclean, report), test)
- functest_utils.execute_command(cmd)
-
-
-class CliTestcase(Testcase):
-
- def list(self):
- click.echo(super(CliTestcase, self).list())
-
- def show(self, name):
- testcase_show = super(CliTestcase, self).show(name)
- if testcase_show:
- click.echo(testcase_show)
- else:
- click.echo("The test case '%s' does not exist or is not supported."
- % name)
+++ /dev/null
-#!/usr/bin/env python
-#
-# jose.lausuch@ericsson.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
-
-# pylint: disable=missing-docstring
-
-import pkg_resources
-
-import click
-from xtesting.ci import tier_builder
-
-from functest.utils import functest_utils
-from functest.utils import env
-
-
-class Tier(object):
-
- def __init__(self):
- self.tiers = tier_builder.TierBuilder(
- env.get('INSTALLER_TYPE'),
- env.get('DEPLOY_SCENARIO'),
- pkg_resources.resource_filename('xtesting', 'ci/testcases.yaml'))
-
- def list(self):
- summary = ""
- for tier in self.tiers.get_tiers():
- summary += (" - %s. %s:\n\t %s\n"
- % (tier.get_order(),
- tier.get_name(),
- tier.get_test_names()))
- return summary
-
- def show(self, name):
- tier = self.tiers.get_tier(name)
- if tier is None:
- return None
- tier_info = self.tiers.get_tier(name)
- return tier_info
-
- def gettests(self, name):
- tier = self.tiers.get_tier(name)
- if tier is None:
- return None
- tests = tier.get_test_names()
- return tests
-
- @staticmethod
- def get_flags(noclean=False, report=False):
- flags = ""
- if noclean:
- flags += "-n "
- if report:
- flags += "-r "
- return flags
-
- @staticmethod
- def run(name, noclean=False, report=False):
- cmd = "run_tests {}-t {}".format(Tier.get_flags(noclean, report), name)
- functest_utils.execute_command(cmd)
-
-
-class CliTier(Tier):
-
- def list(self):
- click.echo(super(CliTier, self).list())
-
- def show(self, name):
- tier_info = super(CliTier, self).show(name)
- 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" % (name, tier_names))
-
- def gettests(self, name):
- tests = super(CliTier, self).gettests(name)
- if tests:
- click.echo("Test cases in tier '%s':\n %s\n" % (name, 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" % (name, tier_names))
+++ /dev/null
-#!/usr/bin/env python
-
-# 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
-
-# pylint: disable=missing-docstring
-
-import logging
-import os
-import re
-import unittest
-
-import mock
-
-from functest.cli.commands import cli_env
-
-
-class RegexMatch(object): # pylint: disable=too-few-public-methods
- def __init__(self, msg):
- self.msg = msg
-
- def __eq__(self, other):
- match = re.search(self.msg, other)
- return match is not None
-
-
-class CliEnvTesting(unittest.TestCase):
-
- def setUp(self):
- self.cli_environ = cli_env.CliEnv()
-
- def _test_show_missing_env_var(self, var, *args):
- # pylint: disable=unused-argument
- if var == 'INSTALLER_TYPE':
- os.environ['INSTALLER_TYPE'] = ''
- reg_string = r"| INSTALLER: Unknown, \S+\s*|"
- elif var == 'SCENARIO':
- os.environ['DEPLOY_SCENARIO'] = ''
- reg_string = r"| SCENARIO: Unknown\s*|"
- elif var == 'NODE':
- os.environ['NODE_NAME'] = ''
- reg_string = r"| POD: Unknown\s*|"
- elif var == 'BUILD_TAG':
- os.environ['BUILD_TAG'] = ''
- reg_string = r"| BUILD TAG: None|"
-
- with mock.patch('functest.cli.commands.cli_env.click.echo') \
- as mock_click_echo:
- self.cli_environ.show()
- mock_click_echo.assert_called_with(RegexMatch(reg_string))
-
- def test_show_ci_installer_type_ko(self, *args):
- self._test_show_missing_env_var('INSTALLER_TYPE', *args)
-
- def test_show_missing_ci_scenario(self, *args):
- self._test_show_missing_env_var('SCENARIO', *args)
-
- def test_show_missing_ci_node(self, *args):
- self._test_show_missing_env_var('NODE', *args)
-
- def test_show_missing_ci_build_tag(self, *args):
- self._test_show_missing_env_var('BUILD_TAG', *args)
-
-
-if __name__ == "__main__":
- logging.disable(logging.CRITICAL)
- unittest.main(verbosity=2)
+++ /dev/null
-#!/usr/bin/env python
-#
-# jose.lausuch@ericsson.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
-
-# pylint: disable=missing-docstring
-
-import logging
-import unittest
-import os
-
-import mock
-
-from functest.cli.commands import cli_os
-
-
-class CliOpenStackTesting(unittest.TestCase):
-
- def setUp(self):
- self.endpoint_ip = 'test_ip'
- self.os_auth_url = 'http://test_ip:test_port/v2.0'
- self.installer_type = 'test_installer_type'
- self.openstack_creds = 'test_env_file'
- self.snapshot_file = 'test_snapshot_file'
- os.environ["OS_AUTH_URL"] = ''
- self.cli_os = cli_os.CliOpenStack()
-
- def test_ping_endpoint_default(self):
- self.cli_os.os_auth_url = self.os_auth_url
- self.cli_os.endpoint_ip = self.endpoint_ip
- with mock.patch('functest.cli.commands.cli_os.os.system',
- return_value=0):
- self.assertEqual(self.cli_os.ping_endpoint(), 0)
-
- @mock.patch('functest.cli.commands.cli_os.exit', side_effect=Exception)
- @mock.patch('functest.cli.commands.cli_os.click.echo')
- def test_ping_endpoint_auth_url_ko(self, mock_click_echo, mock_exit):
- with self.assertRaises(Exception):
- self.cli_os.os_auth_url = None
- self.cli_os.ping_endpoint()
- mock_click_echo.assert_called_once_with(
- "Source the OpenStack credentials first")
- mock_exit.assert_called_once_with(0)
-
- @mock.patch('functest.cli.commands.cli_os.exit')
- @mock.patch('functest.cli.commands.cli_os.click.echo')
- def test_ping_endpoint_system_fails(self, mock_click_echo, mock_exit):
- self.cli_os.os_auth_url = self.os_auth_url
- self.cli_os.endpoint_ip = self.endpoint_ip
- with mock.patch('functest.cli.commands.cli_os.os.system',
- return_value=1):
- self.cli_os.ping_endpoint()
- mock_click_echo.assert_called_once_with(
- "Cannot talk to the endpoint %s\n" % self.endpoint_ip)
- mock_exit.assert_called_once_with(0)
-
- def test_check(self):
- with mock.patch.object(self.cli_os, 'ping_endpoint'), \
- mock.patch('functest.cli.commands.cli_os.check_deployment.'
- 'CheckDeployment') as mock_check_deployment:
- self.cli_os.check()
- self.assertTrue(mock_check_deployment.called)
-
- @mock.patch('functest.cli.commands.cli_os.click.echo')
- def test_show_credentials(self, mock_click_echo):
- key = 'OS_KEY'
- value = 'OS_VALUE'
- with mock.patch.dict(os.environ, {key: value}):
- self.cli_os.show_credentials()
- mock_click_echo.assert_any_call("{}={}".format(key, value))
-
-
-if __name__ == "__main__":
- logging.disable(logging.CRITICAL)
- unittest.main(verbosity=2)
+++ /dev/null
-#!/usr/bin/env python
-
-# 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
-
-# pylint: disable=missing-docstring
-
-import logging
-import unittest
-
-import mock
-
-from functest.cli.commands import cli_testcase
-
-
-class CliTestCasesTesting(unittest.TestCase):
-
- def setUp(self):
- self.testname = 'testname'
- with mock.patch('xtesting.ci.tier_builder'):
- self.cli_tests = cli_testcase.CliTestcase()
-
- @mock.patch('functest.utils.functest_utils.execute_command')
- def test_run_default(self, mock_ft_utils):
- cmd = "run_tests -n -r -t {}".format(self.testname)
- self.cli_tests.run(self.testname, noclean=True, report=True)
- mock_ft_utils.assert_called_with(cmd)
-
- @mock.patch('functest.utils.functest_utils.execute_command')
- def test_run_noclean_missing_report(self, mock_ft_utils):
- cmd = "run_tests -n -t {}".format(self.testname)
- self.cli_tests.run(self.testname, noclean=True, report=False)
- mock_ft_utils.assert_called_with(cmd)
-
- @mock.patch('functest.utils.functest_utils.execute_command')
- def test_run_report_missing_noclean(self, mock_ft_utils):
- cmd = "run_tests -r -t {}".format(self.testname)
- self.cli_tests.run(self.testname, noclean=False, report=True)
- mock_ft_utils.assert_called_with(cmd)
-
- @mock.patch('functest.utils.functest_utils.execute_command')
- def test_run_missing_noclean_report(self, mock_ft_utils):
- cmd = "run_tests -t {}".format(self.testname)
- self.cli_tests.run(self.testname, noclean=False, report=False)
- mock_ft_utils.assert_called_with(cmd)
-
- @mock.patch('functest.cli.commands.cli_testcase.click.echo')
- def test_list(self, mock_click_echo):
- with mock.patch.object(self.cli_tests.tiers, 'get_tiers',
- return_value=[]):
- self.cli_tests.list()
- mock_click_echo.assert_called_with("")
-
- @mock.patch('functest.cli.commands.cli_testcase.click.echo')
- def test_show_default_desc_none(self, mock_click_echo):
- with mock.patch.object(self.cli_tests.tiers, 'get_test',
- return_value=None):
- self.cli_tests.show(self.testname)
- mock_click_echo.assert_any_call("The test case '%s' "
- "does not exist or is"
- " not supported."
- % self.testname)
-
- @mock.patch('functest.cli.commands.cli_testcase.click.echo')
- def test_show_default(self, mock_click_echo):
- mock_obj = mock.Mock()
- with mock.patch.object(self.cli_tests.tiers, 'get_test',
- return_value=mock_obj):
- self.cli_tests.show(self.testname)
- mock_click_echo.assert_called_with(mock_obj)
-
-
-if __name__ == "__main__":
- logging.disable(logging.CRITICAL)
- unittest.main(verbosity=2)
+++ /dev/null
-#!/usr/bin/env python
-
-# 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
-
-# pylint: disable=missing-docstring
-
-import logging
-import unittest
-
-import mock
-
-from functest.cli.commands import cli_tier
-
-
-class CliTierTesting(unittest.TestCase):
-
- def setUp(self):
- self.tiername = 'tiername'
- self.testnames = 'testnames'
- with mock.patch('xtesting.ci.tier_builder'):
- self.cli_tier = cli_tier.CliTier()
-
- @mock.patch('functest.cli.commands.cli_tier.click.echo')
- def test_list(self, mock_click_echo):
- with mock.patch.object(self.cli_tier.tiers, 'get_tiers',
- return_value=[]):
- self.cli_tier.list()
- mock_click_echo.assert_called_with("")
-
- @mock.patch('functest.cli.commands.cli_tier.click.echo')
- def test_show_default(self, mock_click_echo):
- with mock.patch.object(self.cli_tier.tiers, 'get_tier',
- return_value=self.tiername):
- self.cli_tier.show(self.tiername)
- mock_click_echo.assert_called_with(self.tiername)
-
- @mock.patch('functest.cli.commands.cli_tier.click.echo')
- def test_show_missing_tier(self, mock_click_echo):
- with mock.patch.object(self.cli_tier.tiers, 'get_tier',
- return_value=None), \
- mock.patch.object(self.cli_tier.tiers, 'get_tier_names',
- return_value='tiernames'):
- self.cli_tier.show(self.tiername)
- mock_click_echo.assert_called_with("The tier with name '%s' does "
- "not exist. Available tiers are"
- ":\n %s\n" % (self.tiername,
- 'tiernames'))
-
- @mock.patch('functest.cli.commands.cli_tier.click.echo')
- def test_gettests_default(self, mock_click_echo):
- mock_obj = mock.Mock()
- attrs = {'get_test_names.return_value': self.testnames}
- mock_obj.configure_mock(**attrs)
-
- with mock.patch.object(self.cli_tier.tiers, 'get_tier',
- return_value=mock_obj):
- self.cli_tier.gettests(self.tiername)
- mock_click_echo.assert_called_with(
- "Test cases in tier '%s':\n %s\n" % (
- self.tiername, self.testnames))
-
- @mock.patch('functest.cli.commands.cli_tier.click.echo')
- def test_gettests_missing_tier(self, mock_click_echo):
- with mock.patch.object(self.cli_tier.tiers, 'get_tier',
- return_value=None), \
- mock.patch.object(self.cli_tier.tiers, 'get_tier_names',
- return_value='tiernames'):
- self.cli_tier.gettests(self.tiername)
- mock_click_echo.assert_called_with("The tier with name '%s' does "
- "not exist. Available tiers are"
- ":\n %s\n" % (self.tiername,
- 'tiernames'))
-
- @mock.patch('functest.utils.functest_utils.execute_command')
- def test_run_default(self, mock_ft_utils):
- cmd = "run_tests -n -r -t {}".format(self.tiername)
- self.cli_tier.run(self.tiername, noclean=True, report=True)
- mock_ft_utils.assert_called_with(cmd)
-
- @mock.patch('functest.utils.functest_utils.execute_command')
- def test_run_report_missing_noclean(self, mock_ft_utils):
- cmd = "run_tests -r -t {}".format(self.tiername)
- self.cli_tier.run(self.tiername, noclean=False, report=True)
- mock_ft_utils.assert_called_with(cmd)
-
- @mock.patch('functest.utils.functest_utils.execute_command')
- def test_run_noclean_missing_report(self, mock_ft_utils):
- cmd = "run_tests -n -t {}".format(self.tiername)
- self.cli_tier.run(self.tiername, noclean=True, report=False)
- mock_ft_utils.assert_called_with(cmd)
-
- @mock.patch('functest.utils.functest_utils.execute_command')
- def test_run_missing_noclean_report(self, mock_ft_utils):
- cmd = "run_tests -t {}".format(self.tiername)
- self.cli_tier.run(self.tiername, noclean=False, report=False)
- mock_ft_utils.assert_called_with(cmd)
-
-
-if __name__ == "__main__":
- logging.disable(logging.CRITICAL)
- unittest.main(verbosity=2)
+++ /dev/null
-#!/usr/bin/env python
-
-# Copyright (c) 2016 Orange 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
-
-# pylint: disable=missing-docstring
-
-import logging
-import os
-import unittest
-
-import mock
-from click.testing import CliRunner
-
-with mock.patch('functest.cli.commands.cli_testcase.CliTestcase.__init__',
- mock.Mock(return_value=None)), \
- mock.patch('functest.cli.commands.cli_tier.CliTier.__init__',
- mock.Mock(return_value=None)):
- os.environ['OS_AUTH_URL'] = ''
- from functest.cli import cli_base
-
-
-class CliBaseTesting(unittest.TestCase):
-
- def setUp(self):
- self.runner = CliRunner()
- self._openstack = cli_base.OPENSTACK
- self._env = cli_base.ENV
- self._testcase = cli_base.TESTCASE
- self._tier = cli_base.TIER
-
- def test_os_check(self):
- with mock.patch.object(self._openstack, 'check') as mock_method:
- result = self.runner.invoke(cli_base.os_check)
- self.assertEqual(result.exit_code, 0)
- self.assertTrue(mock_method.called)
-
- def test_os_show_credentials(self):
- with mock.patch.object(self._openstack, 'show_credentials') \
- as mock_method:
- result = self.runner.invoke(cli_base.os_show_credentials)
- self.assertEqual(result.exit_code, 0)
- self.assertTrue(mock_method.called)
-
- def test_env_show(self):
- with mock.patch.object(self._env, 'show') as mock_method:
- result = self.runner.invoke(cli_base.env_show)
- self.assertEqual(result.exit_code, 0)
- self.assertTrue(mock_method.called)
-
- def test_testcase_list(self):
- with mock.patch.object(self._testcase, 'list') as mock_method:
- result = self.runner.invoke(cli_base.testcase_list)
- self.assertEqual(result.exit_code, 0)
- self.assertTrue(mock_method.called)
-
- def test_testcase_show(self):
- with mock.patch.object(self._testcase, 'show') as mock_method:
- result = self.runner.invoke(cli_base.testcase_show, ['testname'])
- self.assertEqual(result.exit_code, 0)
- self.assertTrue(mock_method.called)
-
- def test_testcase_run(self):
- with mock.patch.object(self._testcase, 'run') as mock_method:
- result = self.runner.invoke(cli_base.testcase_run,
- ['testname', '--noclean'])
- self.assertEqual(result.exit_code, 0)
- self.assertTrue(mock_method.called)
-
- def test_tier_list(self):
- with mock.patch.object(self._tier, 'list') as mock_method:
- result = self.runner.invoke(cli_base.tier_list)
- self.assertEqual(result.exit_code, 0)
- self.assertTrue(mock_method.called)
-
- def test_tier_show(self):
- with mock.patch.object(self._tier, 'show') as mock_method:
- result = self.runner.invoke(cli_base.tier_show, ['tiername'])
- self.assertEqual(result.exit_code, 0)
- self.assertTrue(mock_method.called)
-
- def test_tier_gettests(self):
- with mock.patch.object(self._tier, 'gettests') as mock_method:
- result = self.runner.invoke(cli_base.tier_gettests, ['tiername'])
- self.assertEqual(result.exit_code, 0)
- self.assertTrue(mock_method.called)
-
- def test_tier_run(self):
- with mock.patch.object(self._tier, 'run') as mock_method:
- result = self.runner.invoke(cli_base.tier_run,
- ['tiername', '--noclean'])
- self.assertEqual(result.exit_code, 0)
- self.assertTrue(mock_method.called)
-
-
-if __name__ == "__main__":
- logging.disable(logging.CRITICAL)
- unittest.main(verbosity=2)
[entry_points]
console_scripts =
- functest = functest.cli.cli_base:cli
functest_odl = functest.opnfv_tests.sdn.odl.odl:main
check_deployment = functest.ci.check_deployment:main
- functest_restapi = functest.api.server:main
[build_sphinx]
all_files = 1
basepython = python2.7
whitelist_externals = bash
modules =
- functest.api
functest.ci
- functest.cli
functest.opnfv_tests.openstack.rally
functest.opnfv_tests.openstack.refstack
functest.opnfv_tests.openstack.snaps
functest.opnfv_tests.sdn.odl
functest.opnfv_tests.vnf.router
functest.tests.unit.ci
- functest.tests.unit.cli
functest.tests.unit.odl
functest.tests.unit.openstack.rally
functest.tests.unit.openstack.snaps
files =
.travis.yml
docker
- functest/api
functest/ci
functest/opnfv_tests/vnf
commands =
[testenv:py35]
dirs =
functest/tests/unit/ci
- functest/tests/unit/cli
functest/tests/unit/core
functest/tests/unit/odl
functest/tests/unit/openstack
Code Review / functest.git / commitdiff