bundler \
crudini \
curl \
+dnsmasq \
gcc \
git \
libffi-dev \
python-mock \
python-pip \
postgresql \
-ruby1.9.1-dev \
+ruby \
+ruby-dev \
+ruby-bundler \
ssh \
sshpass \
wget \
# other repositories
RUN git clone --depth 1 -b $ODL_TAG https://git.opendaylight.org/gerrit/p/integration/test.git /src/odl_test
-RUN git clone --depth 1 -b $VIMS_TAG https://github.com/boucherv-orange/clearwater-live-test ${REPOS_VNFS_DIR}/vims-test
+RUN git clone --depth 1 -b $VIMS_TAG https://github.com/boucherv-orange/clearwater-live-test /src/vims-test
RUN git clone --depth 1 -b $VROUTER_TAG https://github.com/oolorg/opnfv-functest-vrouter.git ${REPOS_VNFS_DIR}/vrouter
RUN git clone --depth 1 https://github.com/wuwenbin2/OnosSystemTest.git ${REPOS_DIR}/onos
-RUN gpg --keyserver hkp://p80.pool.sks-keyservers.net:80 --recv-keys 409B6B1796C275462A1703113804BB82D39DC0E3
-RUN curl -L https://get.rvm.io | bash -s stable
-
# SFC integration
RUN /bin/bash -c ". /usr/local/lib/python2.7/dist-packages/sfc/tests/functest/setup_scripts/tacker_client_install.sh"
RUN ln -s /src/tempest /src/refstack-client/.tempest \
&& virtualenv --system-site-packages /src/tempest/.venv
-RUN /bin/bash -c ". /etc/profile.d/rvm.sh \
- && cd ${REPOS_VNFS_DIR}/vims-test \
- && rvm autolibs enable"
-RUN /bin/bash -c ". /etc/profile.d/rvm.sh \
- && cd ${REPOS_VNFS_DIR}/vims-test \
- && rvm install 1.9.3"
-RUN /bin/bash -c ". /etc/profile.d/rvm.sh \
- && cd ${REPOS_VNFS_DIR}/vims-test \
- && rvm use 1.9.3"
-RUN /bin/bash -c ". /etc/profile.d/rvm.sh \
- && cd ${REPOS_VNFS_DIR}/vims-test \
- && bundle install"
+RUN cd /src/vims-test && bundle install
RUN sh -c 'curl -sL https://deb.nodesource.com/setup_4.x | sudo -E bash -' \
&& sudo apt-get install -y nodejs \
baro_tests
sdnvpn
-opera
securityscanning
sfc
promise
--- /dev/null
+FROM opnfv/functest-core
+
+ARG VIMS_TAG=stable
+
+RUN apk --no-cache add --update \
+ ruby ruby-dev ruby-bundler ruby-irb ruby-rdoc dnsmasq \
+ procps git g++ make libxslt-dev libxml2-dev zlib-dev libffi-dev && \
+ git clone --depth 1 -b $VIMS_TAG https://github.com/boucherv-orange/clearwater-live-test /src/vims-test && \
+ rm -r /src/vims-test/.git && \
+ cd /src/vims-test && bundle config build.nokogiri --use-system-libraries && bundle install --system
+COPY testcases.yaml /usr/lib/python2.7/site-packages/functest/ci/testcases.yaml
+CMD ["bash","-c","prepare_env start && run_tests -t all"]
--- /dev/null
+#!/bin/bash
+
+from="${DOCKER_REPO%/*}/functest-core"
+sed -i "s|^FROM.*$|FROM ${from}|" Dockerfile
+
+exit $?
--- /dev/null
+tiers:
+ -
+ name: vnf
+ order: 4
+ ci_loop: '(daily)|(weekly)'
+ description : >-
+ Collection of VNF test cases.
+ testcases:
+ -
+ case_name: cloudify_ims
+ project_name: functest
+ criteria: 100
+ blocking: false
+ description: >-
+ This test case deploys an OpenSource vIMS solution from Clearwater
+ using the Cloudify orchestrator. It also runs some signaling traffic.
+ dependencies:
+ installer: ''
+ scenario: 'os-nosdn-nofeature-ha'
+ run:
+ module: 'functest.opnfv_tests.vnf.ims.cloudify_ims'
+ class: 'CloudifyIms'
+ -
+ case_name: orchestra_openims
+ project_name: functest
+ criteria: 100
+ blocking: false
+ description: >-
+ OpenIMS VNF deployment with Open Baton (Orchestra)
+ dependencies:
+ installer: ''
+ scenario: 'os-nosdn-nofeature-ha'
+ run:
+ module: 'functest.opnfv_tests.vnf.ims.orchestra_openims'
+ class: 'OpenImsVnf'
+
+ -
+ case_name: orchestra_clearwaterims
+ project_name: functest
+ criteria: 100
+ blocking: false
+ description: >-
+ ClearwaterIMS VNF deployment with Open Baton (Orchestra)
+ dependencies:
+ installer: ''
+ scenario: 'os-nosdn-nofeature-ha'
+ run:
+ module: 'functest.opnfv_tests.vnf.ims.orchestra_clearwaterims'
+ class: 'ClearwaterImsVnf'
from flask import request
from flask_restful import Resource
-from functest.api.common import api_utils, error
+from functest.api.common import api_utils
LOGGER = logging.getLogger(__name__)
try:
return getattr(self, action)(args)
except AttributeError:
- error.result_handler(status=1, data='No such action')
+ api_utils.result_handler(status=1, data='No such action')
# Import modules from package "functest.api.resources"
import sys
from oslo_utils import importutils
+from flask import jsonify
import six
import functest
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 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
+
+"""
+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
Resources to handle openstack related requests
"""
+import collections
+import logging
+
from flask import jsonify
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 openstack_utils as os_utils
from functest.utils.constants import CONST
+LOGGER = logging.getLogger(__name__)
+
class V1Creds(ApiResource):
""" V1Creds Resource class"""
os_utils.source_credentials(CONST.__getattribute__('openstack_creds'))
credentials_show = OpenStack.show_credentials()
return jsonify(credentials_show)
+
+ 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 = CONST.__getattribute__('openstack_creds')
+ with open(rc_file, 'w') as creds_file:
+ creds_file.writelines(lines)
+
+ LOGGER.info("Sourcing the OpenStack RC file...")
+ try:
+ os_utils.source_credentials(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')
from functest.api.base import ApiResource
from functest.cli.commands.cli_env import Env
+from functest.api.common import api_utils
import functest.utils.functest_utils as ft_utils
def prepare(self, args): # pylint: disable=no-self-use, unused-argument
""" Prepare environment """
- ft_utils.execute_command("prepare_env start")
+ try:
+ ft_utils.execute_command("prepare_env start")
+ except Exception as err: # pylint: disable=broad-except
+ return api_utils.result_handler(status=1, data=str(err))
+ return api_utils.result_handler(
+ status=0, data="Prepare env 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 json
+import logging
+import uuid
+
+from flask import jsonify
+
+from functest.api.base import ApiResource
+from functest.api.common import api_utils
+from functest.api.database.v1.handlers import TasksHandler
+
+
+LOGGER = logging.getLogger(__name__)
+
+
+class V1Tasks(ApiResource):
+ """ V1Tasks Resource class"""
+
+ 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')
+ if status == 'IN PROGRESS':
+ result = {'status': status, 'result': ''}
+ elif status == 'FAIL':
+ result = {'status': status, 'error': task.error}
+ else:
+ result = {'status': status, 'result': json.loads(task.result)}
+
+ return jsonify(result)
Resources to handle testcase related requests
"""
+import os
+import logging
+import uuid
+
from flask import abort, jsonify
from functest.api.base import ApiResource
-from functest.api.common import api_utils
+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.constants import CONST
+import functest.utils.functest_utils as ft_utils
+
+LOGGER = logging.getLogger(__name__)
class V1Testcases(ApiResource):
result.update(testcase_info)
result.update({'dependency': dependency_dict})
return jsonify(result)
+
+ 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')
+
+ 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()
+
+ results = {'testcase': case_name, 'task_id': task_id}
+ return jsonify(results)
+
+ def _run(self, args): # pylint: disable=no-self-use
+ """ The built_in function to run a test case """
+
+ case_name = args.get('testcase')
+
+ if not os.path.isfile(CONST.__getattribute__('env_active')):
+ raise Exception("Functest environment is not ready.")
+ else:
+ 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': CONST.__getattribute__('INSTALLER_TYPE'),
+ 'scenario': CONST.__getattribute__('DEPLOY_SCENARIO'),
+ 'build_tag': CONST.__getattribute__('BUILD_TAG'),
+ 'ci_loop': CONST.__getattribute__('CI_LOOP')
+ }
+ result = {
+ 'task_id': args.get('task_id'),
+ 'case_name': case_name,
+ 'env_info': env_info,
+ 'result': result
+ }
+
+ return {'result': result}
"""
+import inspect
import logging
import socket
from urlparse import urljoin
from flask_restful import Api
from functest.api.base import ApiResource
-from functest.api.urls import URLPATTERNS
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)
+
+
+@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 """
return urljoin('http://{}:5000'.format(address), url)
-def api_add_resource(api):
+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
"""
for url_pattern in URLPATTERNS:
try:
- api.add_resource(
+ 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-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'))
LOGGER.info('Starting Functest server')
- app = Flask(__name__)
- api = Api(app)
- api_add_resource(api)
- app.run(host='0.0.0.0')
+ api_add_resource()
+ init_db()
+ APP.run(host='0.0.0.0')
# 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 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>/testcases
# => GET all testcases within given tier
- Url('/api/v1/functest/tiers/<tier_name>/testcases', 'v1_testcases_in_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_tasks')
]
hw_firmware_type: 'uefi'
short_id: 'ubuntu16.04'
hw_video_model: 'vga'
+ ubuntu:
+ disk_file: /home/opnfv/functest/images/ubuntu-14.04-server-cloudimg-arm64-uefi1.img
+ extra_properties:
+ hw_firmware_type: 'uefi'
+ hw_video_model: 'vga'
+ centos:
+ disk_file: /home/opnfv/functest/images/CentOS-7-aarch64-GenericCloud.qcow2
+
vping:
image_name: TestVM
dir_repo_rally: /home/opnfv/repos/rally
repo_tempest: /src/tempest
dir_repo_releng: /home/opnfv/repos/releng
- repo_vims_test: /home/opnfv/repos/vnfs/vims-test
+ repo_vims_test: /src/vims-test
repo_onos: /home/opnfv/repos/onos
repo_barometer: /home/opnfv/repos/barometer
repo_doctor: /home/opnfv/repos/doctor
vm_name_2: opnfv-vping-2
image_name: functest-vping
private_net_name: vping-net
+ # network_type: vlan
+ # physical_network: physnet2
+ # segmentation_id: 2366
private_subnet_name: vping-subnet
private_subnet_cidr: 192.168.130.0/24
router_name: vping-router
router_name: rally-router
vnf:
- aaa:
- tenant_name: aaa
- tenant_description: Freeradius server
- tenant_images: {}
juju_epc:
tenant_name: epc
tenant_description: OAI EPC deployed with Juju
tenant_name: orchestra_clearwaterims
tenant_description: Clearwater IMS deployed with Open Baton
config: orchestra.yaml
- opera_ims:
- tenant_name: opera_ims
- tenant_description: ims deployed with open-o
ONOS:
general:
#!/bin/bash
-CIRROS_REPO_URL=http://download.cirros-cloud.net
-CIRROS_AARCH64_TAG=161201
-CIRROS_X86_64_TAG=0.3.5
-
-RED='\033[1;31m'
-NC='\033[0m' # No Color
-
-function usage(){
- echo -e "${RED}USAGE: $script <destination_folder> <scenario_name> [arch]${NC}"
- exit 0
-}
-
-script=`basename "$0"`
-IMAGES_FOLDER_DIR=$1
-SCENARIO=$2
-ARCH=$3
-
-if [[ -z $IMAGES_FOLDER_DIR ]]; then usage; fi;
-
set -ex
-mkdir -p ${IMAGES_FOLDER_DIR}
-
-
-####################
-# MANDATORY IMAGES #
-####################
-# These images should be present in Functest for the tests to work
-
-# Functest:
-wget -nc ${CIRROS_REPO_URL}/${CIRROS_X86_64_TAG}/cirros-${CIRROS_X86_64_TAG}-x86_64-disk.img -P ${IMAGES_FOLDER_DIR}
-wget -nc ${CIRROS_REPO_URL}/${CIRROS_X86_64_TAG}/cirros-${CIRROS_X86_64_TAG}-x86_64-lxc.tar.gz -P ${IMAGES_FOLDER_DIR}
-
-# SNAPS:
-wget -nc http://uec-images.ubuntu.com/releases/trusty/14.04/ubuntu-14.04-server-cloudimg-amd64-disk1.img -P ${IMAGES_FOLDER_DIR}
-wget -nc http://cloud.centos.org/centos/7/images/CentOS-7-x86_64-GenericCloud.qcow2 -P ${IMAGES_FOLDER_DIR}
-
-
-###################
-# OPTIONAL IMAGES #
-###################
-# Optional images can be commented if they are not going to be used by the tests
-
-# SDNVPN (odl-bgpvpn scenarios):
-if [[ ${SCENARIO} == *"bgpvpn"* ]]; then
- wget -nc http://artifacts.opnfv.org/sdnvpn/ubuntu-16.04-server-cloudimg-amd64-disk1.img -P ${IMAGES_FOLDER_DIR}
-fi
-
-# ONOS (onos-sfc scenarios):
-if [[ ${SCENARIO} == *"onos-sfc"* ]]; then
- wget -nc http://artifacts.opnfv.org/onosfw/images/firewall_block_image.img -P ${IMAGES_FOLDER_DIR}
-fi
-
-if [[ ${ARCH} == "arm" ]] || [[ ${ARCH} == "aarch64" ]]; then
- # ARM (aarch64 cirros images):
- wget -nc ${CIRROS_REPO_URL}/daily/20${CIRROS_AARCH64_TAG}/cirros-d${CIRROS_AARCH64_TAG}-aarch64-disk.img -P ${IMAGES_FOLDER_DIR}
- wget -nc ${CIRROS_REPO_URL}/daily/20${CIRROS_AARCH64_TAG}/cirros-d${CIRROS_AARCH64_TAG}-aarch64-initramfs -P ${IMAGES_FOLDER_DIR}
- wget -nc ${CIRROS_REPO_URL}/daily/20${CIRROS_AARCH64_TAG}/cirros-d${CIRROS_AARCH64_TAG}-aarch64-kernel -P ${IMAGES_FOLDER_DIR}
-fi
-set +ex
\ No newline at end of file
+wget_opts="-N --tries=1 --connect-timeout=30"
+
+cat << EOF | wget ${wget_opts} -i - -P ${1:-/home/opnfv/functest/images}
+http://download.cirros-cloud.net/0.3.5/cirros-0.3.5-x86_64-disk.img
+https://cloud-images.ubuntu.com/releases/14.04/release/ubuntu-14.04-server-cloudimg-amd64-disk1.img
+https://cloud.centos.org/centos/7/images/CentOS-7-x86_64-GenericCloud.qcow2
+https://cloud-images.ubuntu.com/releases/16.04/release/ubuntu-16.04-server-cloudimg-amd64-disk1.img
+http://repository.cloudifysource.org/cloudify/4.0.1/sp-release/cloudify-manager-premium-4.0.1.qcow2
+http://cloud-images.ubuntu.com/trusty/current/trusty-server-cloudimg-amd64-disk1.img
+http://download.cirros-cloud.net/0.3.5/cirros-0.3.5-x86_64-lxc.tar.gz
+http://download.cirros-cloud.net/daily/20161201/cirros-d161201-aarch64-disk.img
+http://download.cirros-cloud.net/daily/20161201/cirros-d161201-aarch64-initramfs
+http://download.cirros-cloud.net/daily/20161201/cirros-d161201-aarch64-kernel
+https://cloud-images.ubuntu.com/releases/14.04/release/ubuntu-14.04-server-cloudimg-arm64-uefi1.img
+http://cloud.centos.org/altarch/7/images/aarch64/CentOS-7-aarch64-GenericCloud.qcow2.xz
+EOF
+
+# vnf requires also the next image which cannot be downloaded by jjobs
+# http://marketplace.openbaton.org:8082/api/v1/images/52e2ccc0-1dce-4663-894d-28aab49323aa/img
+
+xz --decompress ${1:-/home/opnfv/functest/images}/CentOS-7-aarch64-GenericCloud.qcow2.xz
+
+exit $?
logger = logging.getLogger('functest.ci.prepare_env')
handler = None
# set the architecture to default
-pod_arch = None
+pod_arch = os.getenv("HOST_ARCH", None)
arch_filter = ['aarch64']
CONFIG_FUNCTEST_PATH = pkg_resources.resource_filename(
-
name: vnf
order: 4
- ci_loop: 'daily'
+ ci_loop: '(daily)|(weekly)'
description : >-
Collection of VNF test cases.
testcases:
run:
module: 'functest.opnfv_tests.vnf.ims.cloudify_ims'
class: 'CloudifyIms'
-
- -
- case_name: aaa
- enabled: false
- project_name: functest
- criteria: 100
- blocking: false
- description: >-
- Test suite from Parser project.
- dependencies:
- installer: ''
- scenario: ''
- run:
- module: 'functest.opnfv_tests.vnf.aaa.aaa'
- class: 'AaaVnf'
-
-
case_name: orchestra_openims
project_name: functest
module: 'functest.opnfv_tests.vnf.ims.orchestra_clearwaterims'
class: 'ClearwaterImsVnf'
- -
- case_name: opera_vims
- enabled: false
- project_name: opera
- criteria: 100
- blocking: false
- description: >-
- VNF deployment with OPEN-O
- dependencies:
- installer: 'compass'
- scenario: 'os-nosdn-openo-ha'
- run:
- module: 'functest.opnfv_tests.vnf.ims.opera_ims'
- class: 'OperaIms'
-
-
case_name: vyos_vrouter
enabled: false
'refstack')
TEMPEST_CONF_YAML = pkg_resources.resource_filename(
'functest', 'opnfv_tests/openstack/tempest/custom_tests/tempest_conf.yaml')
+TEST_ACCOUNTS_FILE = pkg_resources.resource_filename(
+ 'functest',
+ 'opnfv_tests/openstack/tempest/custom_tests/test_accounts.yaml')
CI_INSTALLER_TYPE = CONST.__getattribute__('INSTALLER_TYPE')
CI_INSTALLER_IP = CONST.__getattribute__('INSTALLER_IP')
def create_tempest_resources(use_custom_images=False,
use_custom_flavors=False):
- keystone_client = os_utils.get_keystone_client()
-
- logger.debug("Creating tenant and user for Tempest suite")
- tenant_id = os_utils.create_tenant(
- keystone_client,
- CONST.__getattribute__('tempest_identity_tenant_name'),
- CONST.__getattribute__('tempest_identity_tenant_description'))
- if not tenant_id:
- logger.error("Failed to create %s tenant"
- % CONST.__getattribute__('tempest_identity_tenant_name'))
-
- user_id = os_utils.create_user(
- keystone_client,
- CONST.__getattribute__('tempest_identity_user_name'),
- CONST.__getattribute__('tempest_identity_user_password'),
- None, tenant_id)
- if not user_id:
- logger.error("Failed to create %s user" %
- CONST.__getattribute__('tempest_identity_user_name'))
logger.debug("Creating private network for Tempest suite")
network_dic = os_utils.create_shared_network_full(
return img_flavor_dict
+def create_tenant_user():
+ keystone_client = os_utils.get_keystone_client()
+
+ logger.debug("Creating tenant and user for Tempest suite")
+ tenant_id = os_utils.create_tenant(
+ keystone_client,
+ CONST.__getattribute__('tempest_identity_tenant_name'),
+ CONST.__getattribute__('tempest_identity_tenant_description'))
+ if not tenant_id:
+ logger.error("Failed to create %s tenant"
+ % CONST.__getattribute__('tempest_identity_tenant_name'))
+
+ user_id = os_utils.create_user(
+ keystone_client,
+ CONST.__getattribute__('tempest_identity_user_name'),
+ CONST.__getattribute__('tempest_identity_user_password'),
+ None, tenant_id)
+ if not user_id:
+ logger.error("Failed to create %s user" %
+ CONST.__getattribute__('tempest_identity_user_name'))
+
+ return tenant_id
+
+
def get_verifier_id():
"""
Returns verifer id for current Tempest
config.set('DEFAULT', 'log_file', '{}/tempest.log'.format(deployment_dir))
config.set('oslo_concurrency', 'lock_path',
'{}/lock_files'.format(deployment_dir))
+ generate_test_accounts_file()
+ config.set('auth', 'test_accounts_file', TEST_ACCOUNTS_FILE)
config.set('scenario', 'img_dir', '{}'.format(deployment_dir))
config.set('scenario', 'img_file', 'tempest-image')
config.set('compute', 'image_ref', img_flavor_dict.get("image_id"))
shutil.copyfile(conf_file, confpath)
+def generate_test_accounts_file():
+ """
+ Add needed tenant and user params into test_accounts.yaml
+ """
+
+ logger.debug("Add needed params into test_accounts.yaml...")
+ tenant_id = create_tenant_user()
+ accounts_list = [
+ {
+ 'tenant_name':
+ CONST.__getattribute__('tempest_identity_tenant_name'),
+ 'tenant_id': str(tenant_id),
+ 'username': CONST.__getattribute__('tempest_identity_user_name'),
+ 'password':
+ CONST.__getattribute__('tempest_identity_user_password')
+ }
+ ]
+
+ with open(TEST_ACCOUNTS_FILE, "w") as f:
+ yaml.dump(accounts_list, f, default_flow_style=False)
+
+
def configure_tempest_update_params(tempest_conf_file,
IMAGE_ID=None, FLAVOR_ID=None):
"""
config.set('compute', 'flavor_ref', FLAVOR_ID)
if FLAVOR_ID_ALT is not None:
config.set('compute', 'flavor_ref_alt', FLAVOR_ID_ALT)
- config.set('identity', 'tenant_name',
- CONST.__getattribute__('tempest_identity_tenant_name'))
- config.set('identity', 'username',
- CONST.__getattribute__('tempest_identity_user_name'))
- config.set('identity', 'password',
- CONST.__getattribute__('tempest_identity_user_password'))
config.set('identity', 'region', 'RegionOne')
if os_utils.is_keystone_v3():
auth_version = 'v3'
try:
self.result = 100 * int(num_success) / int(num_executed)
except ZeroDivisionError:
- logger.error("No test has been executed")
self.result = 0
- return
+ if int(num_tests) > 0:
+ logger.info("All tests have been skipped")
+ else:
+ logger.error("No test has been executed")
+ return
with open(os.path.join(conf_utils.TEMPEST_RESULTS_DIR,
"tempest.log"), 'r') as logfile:
'vping_private_subnet_name') + self.guid
private_subnet_cidr = CONST.__getattribute__(
'vping_private_subnet_cidr')
+
+ vping_network_type = None
+ vping_physical_network = None
+ vping_segmentation_id = None
+
+ if (hasattr(CONST, 'network_type')):
+ vping_network_type = CONST.__getattribute__(
+ 'vping_network_type')
+ if (hasattr(CONST, 'physical_network')):
+ vping_physical_network = CONST.__getattribute__(
+ 'vping_physical_network')
+ if (hasattr(CONST, 'segmentation_id')):
+ vping_segmentation_id = CONST.__getattribute__(
+ 'vping_segmentation_id')
+
self.logger.info(
"Creating network with name: '%s'" % private_net_name)
self.network_creator = deploy_utils.create_network(
self.os_creds,
- NetworkSettings(name=private_net_name,
- subnet_settings=[SubnetSettings(
- name=private_subnet_name,
- cidr=private_subnet_cidr)]))
+ NetworkSettings(
+ name=private_net_name,
+ network_type=vping_network_type,
+ physical_network=vping_physical_network,
+ segmentation_id=vping_segmentation_id,
+ subnet_settings=[SubnetSettings(
+ name=private_subnet_name,
+ cidr=private_subnet_cidr)]))
self.creators.append(self.network_creator)
self.logger.info(
+++ /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
-
-import logging
-
-import functest.core.vnf as vnf
-
-
-class AaaVnf(vnf.VnfOnBoarding):
- """AAA VNF sample"""
-
- logger = logging.getLogger(__name__)
-
- def __init__(self, **kwargs):
- if "case_name" not in kwargs:
- kwargs["case_name"] = "aaa"
- super(AaaVnf, self).__init__(**kwargs)
-
- def deploy_orchestrator(self):
- self.logger.info("No VNFM needed to deploy a free radius here")
- return True
-
- def deploy_vnf(self):
- self.logger.info("Freeradius VNF deployment")
- # find a way to deploy freeradius and tester (heat,manual, ..)
- deploy_vnf = {'status': 'PASS', 'version': 'xxxx'}
- self.details['deploy_vnf'] = deploy_vnf
- return True
-
- def test_vnf(self):
- self.logger.info("Run test towards freeradius")
- # once the freeradius is deployed..make some tests
- test_vnf = {'status': 'PASS', 'version': 'xxxx'}
- self.details['test_vnf'] = test_vnf
- return True
import logging
import os
import pkg_resources
+import shlex
import shutil
+import subprocess
import time
import requests
bono_ip=None, ellis_ip=None,
signup_code='secret'):
self.logger.info('Run Clearwater live test')
- nameservers = ft_utils.get_resolvconf_ns()
- resolvconf = ['{0}{1}{2}'.format(os.linesep, 'nameserver ', ns)
- for ns in nameservers]
- self.logger.debug('resolvconf: %s', resolvconf)
dns_file = '/etc/resolv.conf'
dns_file_bak = '/etc/resolv.conf.bak'
+ self.logger.debug('Backup %s -> %s', dns_file, dns_file_bak)
shutil.copy(dns_file, dns_file_bak)
- script = ('echo -e "nameserver {0}{1}" > {2};'
- 'source /etc/profile.d/rvm.sh;'
- 'cd {3};'
- 'rake test[{4}] SIGNUP_CODE={5}'
- .format(dns_ip,
- ''.join(resolvconf),
+ cmd = ("dnsmasq -d -u root --server=/clearwater.opnfv/{0} "
+ "-r /etc/resolv.conf.bak".format(dns_ip))
+ dnsmasq_process = subprocess.Popen(shlex.split(cmd))
+ script = ('echo -e "nameserver {0}" > {1};'
+ 'cd {2};'
+ 'rake test[{3}] SIGNUP_CODE={4}'
+ .format('127.0.0.1',
dns_file,
self.test_dir,
public_domain,
ft_utils.execute_command(cmd,
error_msg='Clearwater live test failed',
output_file=output_file)
-
+ dnsmasq_process.kill()
with open(dns_file_bak, 'r') as bak_file:
result = bak_file.read()
with open(dns_file, 'w') as f:
# needs some images
self.__logger.info("Upload some OS images if it doesn't exist")
- for image_name, image_url in self.images.iteritems():
- self.__logger.info("image: %s, url: %s", image_name, image_url)
- if image_url and image_name:
+ for image_name, image_file in self.images.iteritems():
+ self.__logger.info("image: %s, file: %s", image_name, image_file)
+ if image_file and image_name:
image_creator = OpenStackImage(
self.snaps_creds,
ImageSettings(name=image_name,
image_user='cloud',
img_format='qcow2',
- url=image_url))
+ image_file=image_file))
image_creator.create()
# self.created_object.append(image_creator)
tenant_images:
- ubuntu_14.04: http://cloud-images.ubuntu.com/trusty/current/trusty-server-cloudimg-amd64-disk1.img
- cloudify_manager_4.0: http://repository.cloudifysource.org/cloudify/4.0.1/sp-release/cloudify-manager-premium-4.0.1.qcow2
+ ubuntu_14.04: /home/opnfv/functest/images/trusty-server-cloudimg-amd64-disk1.img
+ cloudify_manager_4.0: /home/opnfv/functest/images/cloudify-manager-premium-4.0.1.qcow2
orchestrator:
name: cloudify
version: '4.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
-
-import json
-import logging
-import os
-import time
-
-from opera import openo_connect
-import requests
-
-import functest.opnfv_tests.vnf.ims.clearwater_ims_base as clearwater_ims_base
-from functest.utils.constants import CONST
-
-
-class OperaIms(clearwater_ims_base.ClearwaterOnBoardingBase):
-
- def __init__(self, **kwargs):
- if "case_name" not in kwargs:
- kwargs["case_name"] = "opera_ims"
- super(OperaIms, self).__init__(**kwargs)
- self.logger = logging.getLogger(__name__)
- self.ellis_file = os.path.join(
- CONST.__getattribute__('dir_results'), 'ellis.info')
- self.live_test_file = os.path.join(
- CONST.__getattribute__('dir_results'), 'live_test_report.json')
- try:
- self.openo_msb_endpoint = os.environ['OPENO_MSB_ENDPOINT']
- except KeyError:
- raise Exception('OPENO_MSB_ENDPOINT is not specified,'
- ' put it as <OPEN-O ip>:<port>')
- else:
- self.logger.info('OPEN-O endpoint is: %s', self.openo_msb_endpoint)
-
- def prepare(self):
- pass
-
- def clean(self):
- pass
-
- def deploy_vnf(self):
- try:
- openo_connect.create_service(self.openo_msb_endpoint,
- 'functest_opera',
- 'VNF for functest testing')
- except Exception as e:
- self.logger.error(e)
- return {'status': 'FAIL', 'result': e}
- else:
- self.logger.info('vIMS deployment is kicked off')
- return {'status': 'PASS', 'result': ''}
-
- def dump_info(self, info_file, result):
- with open(info_file, 'w') as f:
- self.logger.debug('Save information to file: %s', info_file)
- json.dump(result, f)
-
- def test_vnf(self):
- vnfm_ip = openo_connect.get_vnfm_ip(self.openo_msb_endpoint)
- self.logger.info('VNFM IP: %s', vnfm_ip)
- vnf_status_url = 'http://{0}:5000/api/v1/model/status'.format(vnfm_ip)
- vnf_alive = False
- retry = 40
-
- self.logger.info('Check the VNF status')
- while retry > 0:
- rq = requests.get(vnf_status_url, timeout=90)
- response = rq.json()
- vnf_alive = response['vnf_alive']
- msg = response['msg']
- self.logger.info(msg)
- if vnf_alive:
- break
- self.logger.info('check again in one and half a minute...')
- retry = retry - 1
- time.sleep(90)
-
- if not vnf_alive:
- raise Exception('VNF failed to start: {0}'.format(msg))
-
- ellis_config_url = ('http://{0}:5000/api/v1/model/ellis/configure'
- .format(vnfm_ip))
- rq = requests.get(ellis_config_url, timeout=90)
- if rq.json() and not rq.json()['ellis_ok']:
- self.logger.error(rq.json()['data'])
- raise Exception('Failed to configure Ellis')
-
- self.logger.info('Get Clearwater deployment detail')
- vnf_info_url = ('http://{0}:5000/api/v1/model/output'
- .format(vnfm_ip))
- rq = requests.get(vnf_info_url, timeout=90)
- data = rq.json()['data']
- self.logger.info(data)
- bono_ip = data['bono_ip']
- ellis_ip = data['ellis_ip']
- dns_ip = data['dns_ip']
- result = self.config_ellis(ellis_ip, 'signup', True)
- self.logger.debug('Ellis Result: %s', result)
- self.dump_info(self.ellis_file, result)
-
- if dns_ip:
- vims_test_result = self.run_clearwater_live_test(
- dns_ip,
- 'clearwater.local',
- bono_ip,
- ellis_ip,
- 'signup')
- if vims_test_result != '':
- self.dump_info(self.live_test_file, vims_test_result)
- return {'status': 'PASS', 'result': vims_test_result}
- else:
- return {'status': 'FAIL', 'result': ''}
-
- def main(self, **kwargs):
- self.logger.info("Start to run Opera vIMS VNF onboarding test")
- self.execute()
- self.logger.info("Opera vIMS VNF onboarding test finished")
- if self.result is "PASS":
- return self.EX_OK
- else:
- return self.EX_RUN_ERROR
-
- def run(self):
- kwargs = {}
- return self.main(**kwargs)
tenant_images:
orchestrator:
- ubuntu-14.04-server-cloudimg-amd64-disk1: http://cloud-images.ubuntu.com/trusty/current/trusty-server-cloudimg-amd64-disk1.img
+ ubuntu-14.04-server-cloudimg-amd64-disk1: /home/opnfv/functest/images/trusty-server-cloudimg-amd64-disk1.img
orchestra_openims:
- openims: http://marketplace.openbaton.org:8082/api/v1/images/52e2ccc0-1dce-4663-894d-28aab49323aa/img
+ openims: /home/opnfv/functest/images/img
orchestra_clearwaterims:
- ubuntu-14.04-server-cloudimg-amd64-disk1: http://cloud-images.ubuntu.com/trusty/current/trusty-server-cloudimg-amd64-disk1.img
+ ubuntu-14.04-server-cloudimg-amd64-disk1: /home/opnfv/functest/images/trusty-server-cloudimg-amd64-disk1.img
mano:
name: OpenBaton
version: '3.2.0'
ram_min: 2048
disk: 5
vcpus: 2
- test:
\ No newline at end of file
+ test:
if not os.path.exists(self.data_dir):
os.makedirs(self.data_dir)
- self.images = get_config(
- "tenant_images.%s" %
- self.case_name, config_file)
+ self.images = get_config("tenant_images.orchestrator", config_file)
self.images.update(
get_config(
"tenant_images.%s" %
def prepare_images(self):
"""Upload images if they doen't exist yet"""
self.logger.info("Upload images if they doen't exist yet")
- for image_name, image_url in self.images.iteritems():
- self.logger.info("image: %s, url: %s", image_name, image_url)
- if image_url and image_name:
+ for image_name, image_file in self.images.iteritems():
+ self.logger.info("image: %s, file: %s", image_name, image_file)
+ if image_file and image_name:
image = OpenStackImage(
self.snaps_creds,
ImageSettings(name=image_name,
image_user='cloud',
img_format='qcow2',
- url=image_url))
+ image_file=image_file))
image.create()
# self.created_resources.append(image);
if not os.path.exists(self.data_dir):
os.makedirs(self.data_dir)
- self.images = get_config("tenant_images.%s" %
- self.case_name, config_file)
+ self.images = get_config("tenant_images.orchestrator", config_file)
self.images.update(get_config("tenant_images.%s" %
self.case_name, config_file))
self.snaps_creds = None
def prepare_images(self):
"""Upload images if they doen't exist yet"""
self.logger.info("Upload images if they doen't exist yet")
- for image_name, image_url in self.images.iteritems():
- self.logger.info("image: %s, url: %s", image_name, image_url)
- if image_url and image_name:
+ for image_name, image_file in self.images.iteritems():
+ self.logger.info("image: %s, file: %s", image_name, image_file)
+ if image_file and image_name:
image = OpenStackImage(
self.snaps_creds,
ImageSettings(name=image_name,
image_user='cloud',
img_format='qcow2',
- url=image_url))
+ image_file=image_file))
image.create()
# self.created_resources.append(image);
def test_create_tempest_resources_missing_network_dic(self):
with mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'os_utils.get_keystone_client',
- return_value=mock.Mock()), \
- mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'os_utils.create_tenant',
- return_value='test_tenant_id'), \
- mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'os_utils.create_user',
- return_value='test_user_id'), \
- mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'os_utils.create_shared_network_full',
- return_value=None), \
+ 'os_utils.create_shared_network_full',
+ return_value=None), \
self.assertRaises(Exception) as context:
conf_utils.create_tempest_resources()
msg = 'Failed to create private network'
def test_create_tempest_resources_missing_image(self):
with mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'os_utils.get_keystone_client',
+ 'os_utils.create_shared_network_full',
return_value=mock.Mock()), \
- mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'os_utils.create_tenant',
- return_value='test_tenant_id'), \
- mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'os_utils.create_user',
- return_value='test_user_id'), \
- mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'os_utils.create_shared_network_full',
- return_value=mock.Mock()), \
mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
'os_utils.get_or_create_image',
return_value=(mock.Mock(), None)), \
def test_create_tempest_resources_missing_flavor(self):
with mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'os_utils.get_keystone_client',
+ 'os_utils.create_shared_network_full',
return_value=mock.Mock()), \
- mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'os_utils.create_tenant',
- return_value='test_tenant_id'), \
- mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'os_utils.create_user',
- return_value='test_user_id'), \
- mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
- 'os_utils.create_shared_network_full',
- return_value=mock.Mock()), \
mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
'os_utils.get_or_create_image',
return_value=(mock.Mock(), 'image_id')), \
msg = 'Failed to create flavor'
self.assertTrue(msg in context)
+ @mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
+ 'logger.error')
+ def create_tenant_user_and_tenant_ok(self, mock_logger_error):
+ with mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
+ 'os_utils.get_keystone_client',
+ return_value=mock.Mock()), \
+ mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
+ 'os_utils.create_tenant',
+ return_value='test_tenant_id'):
+ conf_utils.create_tenant_user()
+ mock_logger_error.assert_not_called()
+
+ @mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
+ 'logger.error')
+ def create_tenant_user_and_user_ok(self, mock_logger_error):
+ with mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
+ 'os_utils.get_keystone_client',
+ return_value=mock.Mock()), \
+ mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
+ 'os_utils.create_user',
+ return_value='test_user_id'):
+ conf_utils.create_tenant_user()
+ mock_logger_error.assert_not_called()
+
+ @mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
+ 'logger.error')
+ def create_tenant_user_and_tenant_failed(self, mock_logger_error):
+ with mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
+ 'os_utils.get_keystone_client',
+ return_value=mock.Mock()), \
+ mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
+ 'os_utils.create_tenant',
+ return_value=None):
+ conf_utils.create_tenant_user()
+ msg = ("Failed to create %s tenant"
+ % CONST.__getattribute__('tempest_identity_tenant_name'))
+ mock_logger_error.assert_any_call(msg)
+
+ @mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
+ 'logger.error')
+ def create_tenant_user_and_user_failed(self, mock_logger_error):
+ with mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
+ 'os_utils.get_keystone_client',
+ return_value=mock.Mock()), \
+ mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
+ 'os_utils.create_user',
+ return_value=None):
+ conf_utils.create_tenant_user()
+ msg = ("Failed to create %s user"
+ % CONST.__getattribute__('tempest_identity_user_name'))
+ mock_logger_error.assert_any_call(msg)
+
def test_get_verifier_id_missing_verifier(self):
CONST.__setattr__('tempest_deployment_name', 'test_deploy_name')
with mock.patch('functest.opnfv_tests.openstack.tempest.'
'conf_utils.ConfigParser.RawConfigParser.'
'write') as mwrite, \
mock.patch('__builtin__.open', mock.mock_open()), \
+ mock.patch('functest.opnfv_tests.openstack.tempest.'
+ 'conf_utils.generate_test_accounts_file'), \
mock.patch('functest.opnfv_tests.openstack.tempest.'
'conf_utils.shutil.copyfile'):
conf_utils.configure_tempest_defcore('test_dep_dir',
self.assertTrue(mread.called)
self.assertTrue(mwrite.called)
+ def test_generate_test_accounts_file_default(self):
+ with mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
+ 'create_tenant_user',
+ return_value='test_tenant_id') as mock_create, \
+ mock.patch("__builtin__.open", mock.mock_open()), \
+ mock.patch('functest.opnfv_tests.openstack.tempest.conf_utils.'
+ 'yaml.dump') as mock_dump:
+ conf_utils.generate_test_accounts_file()
+ self.assertTrue(mock_create.called)
+ self.assertTrue(mock_dump.called)
+
def _test_missing_param(self, params, image_id, flavor_id):
with mock.patch('functest.opnfv_tests.openstack.tempest.'
'conf_utils.ConfigParser.RawConfigParser.'
if k not in os.environ:
self.__setattr__(k, v)
self._set_ci_run()
- self._set_ci_loop()
+ if 'CI_LOOP' not in os.environ:
+ self._set_ci_loop()
def _set_ci_run(self):
if self.BUILD_TAG:
from novaclient import client as novaclient
from keystoneclient import client as keystoneclient
from neutronclient.neutron import client as neutronclient
-from functest.utils.constants import CONST
+from functest.utils.constants import CONST
import functest.utils.functest_utils as ft_utils
logger = logging.getLogger(__name__)
def get_external_net(neutron_client):
+ if (hasattr(CONST, 'EXTERNAL_NETWORK')):
+ return CONST.__getattribute__('EXTERNAL_NETWORK')
for network in neutron_client.list_networks()['networks']:
if network['router:external']:
return network['name']
def get_external_net_id(neutron_client):
+ if (hasattr(CONST, 'EXTERNAL_NETWORK')):
+ networks = neutron_client.list_networks(
+ name=CONST.__getattribute__('EXTERNAL_NETWORK'))
+ net_id = networks['networks'][0]['id']
+ return net_id
for network in neutron_client.list_networks()['networks']:
if network['router:external']:
return network['id']
git+https://gerrit.opnfv.org/gerrit/snaps#egg=snaps
git+https://gerrit.opnfv.org/gerrit/barometer#egg=baro_tests
git+https://gerrit.opnfv.org/gerrit/sdnvpn#egg=sdnvpn
-git+https://gerrit.opnfv.org/gerrit/opera#egg=opera
git+https://gerrit.opnfv.org/gerrit/securityscanning#egg=securityscanning
git+https://gerrit.opnfv.org/gerrit/sfc#egg=sfc
-e git+https://gerrit.opnfv.org/gerrit/promise#egg=promise
Code Review / functest-xtesting.git / commitdiff