aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--docker/Dockerfile13
-rw-r--r--docker/core/Dockerfile13
-rw-r--r--docker/features/Dockerfile3
-rw-r--r--docker/features/testcases.yaml8
-rw-r--r--functest/api/resources/v1/envs.py9
-rw-r--r--functest/ci/testcases.yaml2
-rw-r--r--functest/core/vnf.py57
-rw-r--r--functest/opnfv_tests/openstack/rally/blacklist.txt1
-rw-r--r--functest/opnfv_tests/vnf/ims/cloudify_ims.py27
-rw-r--r--functest/opnfv_tests/vnf/ims/cloudify_ims.yaml7
-rw-r--r--functest/opnfv_tests/vnf/ims/orchestra_clearwaterims.py20
-rw-r--r--functest/opnfv_tests/vnf/ims/orchestra_openims.py22
-rw-r--r--functest/tests/unit/core/test_vnf.py129
-rw-r--r--functest/tests/unit/utils/test_openstack_tacker.py524
-rw-r--r--functest/tests/unit/vnf/ims/test_cloudify_ims.py106
-rw-r--r--functest/tests/unit/vnf/ims/test_orchestra_clearwaterims.py107
-rw-r--r--functest/tests/unit/vnf/ims/test_orchestra_openims.py107
-rw-r--r--functest/tests/unit/vnf/router/test_cloudify_vrouter.py72
-rw-r--r--functest/utils/openstack_tacker.py317
-rw-r--r--requirements.txt12
-rw-r--r--upper-constraints.txt3
21 files changed, 167 insertions, 1392 deletions
diff --git a/docker/Dockerfile b/docker/Dockerfile
index d60ce53b3..abb9b0d70 100644
--- a/docker/Dockerfile
+++ b/docker/Dockerfile
@@ -13,7 +13,7 @@ LABEL version="0.1" description="OPNFV Functest Docker container"
# Environment variables
ARG BRANCH=master
-ARG RALLY_TAG=0.8.1
+ARG RALLY_TAG=stable/0.9
ARG ODL_TAG=release/carbon
ARG OPENSTACK_TAG=stable/ocata
ARG VIMS_TAG=stable
@@ -77,6 +77,13 @@ RUN wget -q -O- https://git.openstack.org/cgit/openstack/requirements/plain/uppe
> upper-constraints.txt && \
pip install --src /src -cupper-constraints.txt \
-chttps://git.opnfv.org/functest/plain/upper-constraints.txt?h=$BRANCH \
+ -e git+https://github.com/openstack/requirements@stable/ocata#egg=openstack_requirements && \
+ git clone --depth 1 https://github.com/openstack/rally.git -b $RALLY_TAG /src/rally && \
+ update-requirements -s --source /src/openstack-requirements /src/rally/ && \
+ pip install --src /src -cupper-constraints.txt \
+ -chttps://git.opnfv.org/functest/plain/upper-constraints.txt?h=$BRANCH /src/rally/ && \
+ pip install --src /src -cupper-constraints.txt \
+ -chttps://git.opnfv.org/functest/plain/upper-constraints.txt?h=$BRANCH \
git+https://gerrit.opnfv.org/gerrit/functest@$BRANCH#egg=functest \
-rthirdparty-requirements.txt && \
mkdir -p /etc/rally && \
@@ -85,8 +92,6 @@ RUN wget -q -O- https://git.openstack.org/cgit/openstack/requirements/plain/uppe
rm thirdparty-requirements.txt upper-constraints.txt
# OPNFV repositories
-RUN git clone --depth 1 -b $BRANCH https://gerrit.opnfv.org/gerrit/domino /src/domino
-RUN git clone --depth 1 -b $BRANCH https://gerrit.opnfv.org/gerrit/doctor ${REPOS_DIR}/doctor
RUN git clone --depth 1 -b $BRANCH https://gerrit.opnfv.org/gerrit/fds /src/fds
# other repositories
@@ -105,3 +110,5 @@ RUN sh -c 'curl -sL https://deb.nodesource.com/setup_4.x | sudo -E bash -' \
&& cd /src/promise/source && npm install
RUN echo ". ${FUNCTEST_DIR}/cli/functest-complete.sh" >> /root/.bashrc
+
+CMD ["functest_restapi"]
diff --git a/docker/core/Dockerfile b/docker/core/Dockerfile
index ebd76a262..0ab809aad 100644
--- a/docker/core/Dockerfile
+++ b/docker/core/Dockerfile
@@ -2,6 +2,7 @@ FROM alpine:3.6
ARG BRANCH=master
ARG OPENSTACK_TAG=stable/ocata
+ARG RALLY_TAG=stable/0.9
RUN apk --no-cache add --update \
python libffi libssl1.0 libjpeg-turbo py-pip bash \
@@ -12,10 +13,20 @@ RUN apk --no-cache add --update \
wget -q -O- https://git.openstack.org/cgit/openstack/requirements/plain/upper-constraints.txt?h=$OPENSTACK_TAG | \
sed -E s/^tempest==+\(.*\)$/-e\ git+https:\\/\\/github.com\\/openstack\\/tempest@\\1#egg=tempest/ \
> upper-constraints.txt && \
+ pip install -U --no-cache-dir -cupper-constraints.txt \
+ -chttps://git.opnfv.org/functest/plain/upper-constraints.txt?h=$BRANCH \
+ git+https://github.com/pypa/pip.git && \
+ pip install --no-cache-dir --src /src -cupper-constraints.txt \
+ -chttps://git.opnfv.org/functest/plain/upper-constraints.txt?h=$BRANCH \
+ -e git+https://github.com/openstack/requirements@stable/ocata#egg=openstack_requirements && \
+ git clone --depth 1 https://github.com/openstack/rally.git -b $RALLY_TAG /src/rally && \
+ update-requirements -s --source /src/openstack-requirements /src/rally/ && \
+ pip install --no-cache-dir --src /src -cupper-constraints.txt \
+ -chttps://git.opnfv.org/functest/plain/upper-constraints.txt?h=$BRANCH /src/rally/ && \
pip install --no-cache-dir --src /src -cupper-constraints.txt \
-chttps://git.opnfv.org/functest/plain/upper-constraints.txt?h=$BRANCH \
git+https://gerrit.opnfv.org/gerrit/functest@$BRANCH#egg=functest && \
- rm upper-constraints.txt && \
+ rm -r upper-constraints.txt /src/rally/.git && \
mkdir -p /etc/rally && \
printf "[database]\nconnection = 'sqlite:////var/lib/rally/database/rally.sqlite'" > /etc/rally/rally.conf && \
mkdir -p /var/lib/rally/database && rally-manage db create && \
diff --git a/docker/features/Dockerfile b/docker/features/Dockerfile
index 54bffe827..e1214c1e8 100644
--- a/docker/features/Dockerfile
+++ b/docker/features/Dockerfile
@@ -14,8 +14,7 @@ RUN apk --no-cache add --update nodejs nodejs-npm && \
-rthirdparty-requirements.txt && \
npm -g install npm@latest && \
(cd /src/promise/source && npm install) && \
- git clone --depth 1 -b $BRANCH https://gerrit.opnfv.org/gerrit/domino /src/domino && \
- rm -r thirdparty-requirements.txt /src/domino/.git && \
+ rm -r thirdparty-requirements.txt && \
apk del .build-deps
COPY testcases.yaml /usr/lib/python2.7/site-packages/functest/ci/testcases.yaml
CMD ["bash","-c","prepare_env start && run_tests -t all"]
diff --git a/docker/features/testcases.yaml b/docker/features/testcases.yaml
index da83232b3..ec9962f7a 100644
--- a/docker/features/testcases.yaml
+++ b/docker/features/testcases.yaml
@@ -58,15 +58,15 @@ tiers:
-
case_name: functest-odl-sfc
- enabled: false
+ enabled: true
project_name: sfc
criteria: 100
blocking: false
description: >-
- Test suite for odl-sfc to test two chains and two SFs
+ Test suite for odl-sfc to test two chains with one SF and one chain with two SFs
dependencies:
- installer: '(apex)|(fuel)'
- scenario: 'odl_l2-sfc'
+ installer: ''
+ scenario: 'odl.*sfc'
run:
module: 'functest.core.feature'
class: 'BashFeature'
diff --git a/functest/api/resources/v1/envs.py b/functest/api/resources/v1/envs.py
index fb76fa63f..8020544fb 100644
--- a/functest/api/resources/v1/envs.py
+++ b/functest/api/resources/v1/envs.py
@@ -33,10 +33,11 @@ class V1Envs(ApiResource):
def prepare(self, args): # pylint: disable=no-self-use, unused-argument
""" Prepare environment """
- 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))
+
+ result_env = ft_utils.execute_command("prepare_env start")
+ if not result_env == 0:
+ return api_utils.result_handler(
+ status=1, data="Failed to prepare env")
return api_utils.result_handler(
status=0, data="Prepare env successfully")
diff --git a/functest/ci/testcases.yaml b/functest/ci/testcases.yaml
index 2af686219..5364035f2 100644
--- a/functest/ci/testcases.yaml
+++ b/functest/ci/testcases.yaml
@@ -303,7 +303,7 @@ tiers:
-
case_name: functest-odl-sfc
- enabled: false
+ enabled: true
project_name: sfc
criteria: 100
blocking: false
diff --git a/functest/core/vnf.py b/functest/core/vnf.py
index a329212d8..868b8b48f 100644
--- a/functest/core/vnf.py
+++ b/functest/core/vnf.py
@@ -14,7 +14,9 @@ import time
import functest.core.testcase as base
from functest.utils.constants import CONST
-import functest.utils.openstack_utils as os_utils
+from snaps.openstack.create_user import UserSettings, OpenStackUser
+from snaps.openstack.create_project import ProjectSettings, OpenStackProject
+from snaps.openstack.tests import openstack_tests
__author__ = ("Morgan Richomme <morgan.richomme@orange.com>, "
"Valentin Boucher <valentin.boucher@orange.com>")
@@ -43,10 +45,11 @@ class VnfOnBoarding(base.TestCase):
def __init__(self, **kwargs):
super(VnfOnBoarding, self).__init__(**kwargs)
- self.exist_obj = {'tenant': False, 'user': False}
self.tenant_name = CONST.__getattribute__(
'vnf_{}_tenant_name'.format(self.case_name))
- self.creds = {}
+ self.snaps_creds = {}
+ self.created_object = []
+ self.os_project = None
def run(self, **kwargs):
"""
@@ -100,20 +103,29 @@ class VnfOnBoarding(base.TestCase):
'vnf_{}_tenant_description'.format(self.case_name))
self.__logger.info("Prepare VNF: %s, description: %s",
self.tenant_name, tenant_description)
- keystone_client = os_utils.get_keystone_client()
- self.exist_obj['tenant'] = (
- not os_utils.get_or_create_tenant_for_vnf(
- keystone_client,
- self.tenant_name,
- tenant_description))
- self.exist_obj['user'] = not os_utils.get_or_create_user_for_vnf(
- keystone_client, self.tenant_name)
- self.creds = {
- "tenant": self.tenant_name,
- "username": self.tenant_name,
- "password": self.tenant_name,
- "auth_url": os_utils.get_credentials()['auth_url']
- }
+ snaps_creds = openstack_tests.get_credentials(
+ os_env_file=CONST.__getattribute__('openstack_creds'))
+
+ project_creator = OpenStackProject(
+ snaps_creds,
+ ProjectSettings(
+ name=self.tenant_name,
+ description=tenant_description
+ ))
+ project_creator.create()
+ self.created_object.append(project_creator)
+ self.os_project = project_creator
+
+ user_creator = OpenStackUser(snaps_creds,
+ UserSettings(
+ name=self.tenant_name,
+ password=self.tenant_name))
+ self.created_object.append(user_creator)
+
+ project_creator.assoc_user(user_creator.create())
+
+ self.snaps_creds = user_creator.get_os_creds(self.tenant_name)
+
return base.TestCase.EX_OK
except Exception: # pylint: disable=broad-except
self.__logger.exception("Exception raised during VNF preparation")
@@ -185,8 +197,9 @@ class VnfOnBoarding(base.TestCase):
* the tenant
"""
self.__logger.info("test cleaning")
- keystone_client = os_utils.get_keystone_client()
- if not self.exist_obj['tenant']:
- os_utils.delete_tenant(keystone_client, self.tenant_name)
- if not self.exist_obj['user']:
- os_utils.delete_user(keystone_client, self.tenant_name)
+ self.__logger.info('Remove the cloudify manager OS object ..')
+ for creator in reversed(self.created_object):
+ try:
+ creator.clean()
+ except Exception as exc: # pylint: disable=broad-except
+ self.__logger.error('Unexpected error cleaning - %s', exc)
diff --git a/functest/opnfv_tests/openstack/rally/blacklist.txt b/functest/opnfv_tests/openstack/rally/blacklist.txt
index 0623368d8..cdb5be663 100644
--- a/functest/opnfv_tests/openstack/rally/blacklist.txt
+++ b/functest/opnfv_tests/openstack/rally/blacklist.txt
@@ -48,6 +48,7 @@ scenario:
# panko in the deployment. This is not currently fulfilled
# Ref: https://docs.openstack.org/releasenotes/ceilometer/ocata.html
- 'CeilometerEvents..*'
+ - 'CeilometerTraits..*'
functionality:
-
diff --git a/functest/opnfv_tests/vnf/ims/cloudify_ims.py b/functest/opnfv_tests/vnf/ims/cloudify_ims.py
index 8ed7f0dda..c8c2c509c 100644
--- a/functest/opnfv_tests/vnf/ims/cloudify_ims.py
+++ b/functest/opnfv_tests/vnf/ims/cloudify_ims.py
@@ -24,7 +24,6 @@ import functest.opnfv_tests.vnf.ims.clearwater_ims_base as clearwater_ims_base
from functest.utils.constants import CONST
import functest.utils.openstack_utils as os_utils
-from snaps.openstack.os_credentials import OSCreds
from snaps.openstack.create_network import (NetworkSettings, SubnetSettings,
OpenStackNetwork)
from snaps.openstack.create_security_group import (SecurityGroupSettings,
@@ -63,7 +62,6 @@ class CloudifyIms(clearwater_ims_base.ClearwaterOnBoardingBase):
raise Exception("VNF config file not found")
self.snaps_creds = ''
- self.created_object = []
config_file = os.path.join(self.case_dir, self.config)
self.orchestrator = dict(
@@ -101,12 +99,19 @@ class CloudifyIms(clearwater_ims_base.ClearwaterOnBoardingBase):
self.__logger.info("Additional pre-configuration steps")
- self.snaps_creds = OSCreds(
- username=self.creds['username'],
- password=self.creds['password'],
- auth_url=self.creds['auth_url'],
- project_name=self.creds['tenant'],
- identity_api_version=int(os_utils.get_keystone_client_version()))
+ compute_quotas = self.os_project.get_compute_quotas()
+ network_quotas = self.os_project.get_network_quotas()
+
+ for key, value in (
+ self.vnf['requirements']['compute_quotas'].items()):
+ setattr(compute_quotas, key, value)
+
+ for key, value in (
+ self.vnf['requirements']['network_quotas'].items()):
+ setattr(network_quotas, key, value)
+
+ compute_quotas = self.os_project.update_compute_quotas(compute_quotas)
+ network_quotas = self.os_project.update_network_quotas(network_quotas)
# needs some images
self.__logger.info("Upload some OS images if it doesn't exist")
@@ -402,12 +407,6 @@ class CloudifyIms(clearwater_ims_base.ClearwaterOnBoardingBase):
self.__logger.warn("Some issue during the undeployment ..")
self.__logger.warn("Tenant clean continue ..")
- self.__logger.info('Remove the cloudify manager OS object ..')
- for creator in reversed(self.created_object):
- try:
- creator.clean()
- except Exception as exc:
- self.logger.error('Unexpected error cleaning - %s', exc)
super(CloudifyIms, self).clean()
@energy.enable_recording
diff --git a/functest/opnfv_tests/vnf/ims/cloudify_ims.yaml b/functest/opnfv_tests/vnf/ims/cloudify_ims.yaml
index 280e0a6b8..ad7fa70a9 100644
--- a/functest/opnfv_tests/vnf/ims/cloudify_ims.yaml
+++ b/functest/opnfv_tests/vnf/ims/cloudify_ims.yaml
@@ -21,6 +21,13 @@ vnf:
flavor:
name: m1.small
ram_min: 2048
+ compute_quotas:
+ cores: 50
+ instances: 15
+ network_quotas:
+ security_group: 20
+ security_group_rule: 100
+ port: 50
inputs:
image_id: 'ubuntu_14.04'
flavor_id: 'm1.small'
diff --git a/functest/opnfv_tests/vnf/ims/orchestra_clearwaterims.py b/functest/opnfv_tests/vnf/ims/orchestra_clearwaterims.py
index 5154dea20..c924a3470 100644
--- a/functest/opnfv_tests/vnf/ims/orchestra_clearwaterims.py
+++ b/functest/opnfv_tests/vnf/ims/orchestra_clearwaterims.py
@@ -31,7 +31,6 @@ from snaps.openstack.create_network import (
SubnetSettings,
PortSettings)
from snaps.openstack.create_router import OpenStackRouter, RouterSettings
-from snaps.openstack.os_credentials import OSCreds
from snaps.openstack.create_instance import (
VmInstanceSettings,
OpenStackVmInstance)
@@ -208,15 +207,13 @@ class ClearwaterImsVnf(vnf.VnfOnBoarding):
super(ClearwaterImsVnf, self).prepare()
self.logger.info("Additional pre-configuration steps")
- self.logger.info("creds %s", (self.creds))
-
- self.snaps_creds = OSCreds(
- username=self.creds['username'],
- password=self.creds['password'],
- auth_url=self.creds['auth_url'],
- project_name=self.creds['tenant'],
- identity_api_version=int(os_utils.get_keystone_client_version()))
+ self.creds = {
+ "tenant": self.tenant_name,
+ "username": self.tenant_name,
+ "password": self.tenant_name,
+ "auth_url": os_utils.get_credentials()['auth_url']
+ }
self.prepare_images()
self.prepare_flavor()
self.prepare_security_groups()
@@ -661,8 +658,9 @@ class ClearwaterImsVnf(vnf.VnfOnBoarding):
try:
neutron_client = os_utils.get_neutron_client(self.creds)
self.logger.info("Deleting Open Baton Port...")
- port = snaps_utils.neutron_utils.get_port_by_name(
- neutron_client, '%s_port' % self.case_name)
+ port = snaps_utils.neutron_utils.get_port(
+ neutron_client,
+ port_name='%s_port' % self.case_name)
snaps_utils.neutron_utils.delete_port(neutron_client, port)
time.sleep(10)
except Exception as exc:
diff --git a/functest/opnfv_tests/vnf/ims/orchestra_openims.py b/functest/opnfv_tests/vnf/ims/orchestra_openims.py
index 45440704e..aae35146f 100644
--- a/functest/opnfv_tests/vnf/ims/orchestra_openims.py
+++ b/functest/opnfv_tests/vnf/ims/orchestra_openims.py
@@ -32,7 +32,6 @@ from snaps.openstack.create_network import (
SubnetSettings,
PortSettings)
from snaps.openstack.create_router import OpenStackRouter, RouterSettings
-from snaps.openstack.os_credentials import OSCreds
from snaps.openstack.create_instance import (
VmInstanceSettings, OpenStackVmInstance)
from functest.opnfv_tests.openstack.snaps import snaps_utils
@@ -198,22 +197,18 @@ class OpenImsVnf(vnf.VnfOnBoarding):
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(self):
"""Prepare testscase (Additional pre-configuration steps)."""
super(OpenImsVnf, self).prepare()
self.logger.info("Additional pre-configuration steps")
- self.logger.info("creds %s", (self.creds))
-
- self.snaps_creds = OSCreds(
- username=self.creds['username'],
- password=self.creds['password'],
- auth_url=self.creds['auth_url'],
- project_name=self.creds['tenant'],
- identity_api_version=int(os_utils.get_keystone_client_version()))
-
+ self.creds = {
+ "tenant": self.tenant_name,
+ "username": self.tenant_name,
+ "password": self.tenant_name,
+ "auth_url": os_utils.get_credentials()['auth_url']
+ }
self.prepare_images()
self.prepare_flavor()
self.prepare_security_groups()
@@ -697,8 +692,9 @@ class OpenImsVnf(vnf.VnfOnBoarding):
try:
neutron_client = os_utils.get_neutron_client(self.creds)
self.logger.info("Deleting Open Baton Port...")
- port = snaps_utils.neutron_utils.get_port_by_name(
- neutron_client, '%s_port' % self.case_name)
+ port = snaps_utils.neutron_utils.get_port(
+ neutron_client,
+ port_name='%s_port' % self.case_name)
snaps_utils.neutron_utils.delete_port(neutron_client, port)
time.sleep(10)
except Exception as exc:
diff --git a/functest/tests/unit/core/test_vnf.py b/functest/tests/unit/core/test_vnf.py
index 2ebbbe379..00a29ead1 100644
--- a/functest/tests/unit/core/test_vnf.py
+++ b/functest/tests/unit/core/test_vnf.py
@@ -18,6 +18,8 @@ from functest.core import vnf
from functest.core import testcase
from functest.utils import constants
+from snaps.openstack.os_credentials import OSCreds
+
class VnfBaseTesting(unittest.TestCase):
"""The class testing VNF."""
@@ -102,91 +104,62 @@ class VnfBaseTesting(unittest.TestCase):
with self.assertRaises(vnf.VnfTestException):
self.test.test_vnf()
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client')
- @mock.patch('functest.core.vnf.os_utils.delete_user',
- return_value=True)
- def test_clean_user_already_exist(self, *args):
- self.test.exist_obj['user'] = True
- self.test.clean()
- args[0].assert_not_called()
- args[1].assert_called_once_with()
-
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client')
- @mock.patch('functest.core.vnf.os_utils.delete_user',
- return_value=True)
- def test_clean_user_created(self, *args):
- self.test.exist_obj['user'] = False
- self.test.clean()
- args[0].assert_called_once_with(mock.ANY, self.tenant_name)
- args[1].assert_called_once_with()
-
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client')
- @mock.patch('functest.core.vnf.os_utils.delete_tenant',
- return_value=True)
- def test_clean_tenant_already_exist(self, *args):
- self.test.exist_obj['tenant'] = True
- self.test.clean()
- args[0].assert_not_called()
- args[1].assert_called_once_with()
-
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client')
- @mock.patch('functest.core.vnf.os_utils.delete_tenant',
- return_value=True)
- def test_clean_tenant_created(self, *args):
- self.test.exist_obj['tenant'] = False
- self.test.clean()
- args[0].assert_called_once_with(mock.ANY, self.tenant_name)
- args[1].assert_called_once_with()
-
def test_deploy_orch_unimplemented(self):
self.assertTrue(self.test.deploy_orchestrator())
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
- return_value='test')
- @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_credentials',
- return_value={'auth_url': 'test'})
- def test_prepare(self, *args):
- self.assertEqual(self.test.prepare(),
- testcase.TestCase.EX_OK)
- args[0].assert_called_once_with()
- args[1].assert_called_once_with('test', self.tenant_name)
- args[2].assert_called_once_with(
- 'test', self.tenant_name, self.tenant_description)
- args[3].assert_called_once_with()
-
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
+# @mock.patch('snaps.openstack.tests.openstack_tests.get_credentials',
+# return_value='test')
+# @mock.patch('snaps.openstack.create_project.OpenStackProject',
+# return_value=True)
+# @mock.patch('snaps.openstack.create_user.OpenStackUser',
+# return_value=True)
+# def test_prepare(self, *args):
+# self.assertEqual(self.test.prepare(),
+# testcase.TestCase.EX_OK)
+# args[0].assert_called_once_with()
+# args[1].assert_called_once_with('test', self.tenant_name)
+# args[2].assert_called_once_with(
+# 'test', self.tenant_name, self.tenant_description)
+# args[3].assert_called_once_with()
+
+ @mock.patch('snaps.openstack.tests.openstack_tests.get_credentials',
+ return_value=OSCreds(
+ username='user', password='pass',
+ auth_url='http://foo.com:5000/v3', project_name='bar'),
side_effect=Exception)
def test_prepare_keystone_client_ko(self, *args):
with self.assertRaises(vnf.VnfPreparationException):
self.test.prepare()
- args[0].assert_called_once_with()
-
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client')
- @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
- side_effect=Exception)
- def test_prepare_tenant_creation_ko(self, *args):
- with self.assertRaises(vnf.VnfPreparationException):
- self.test.prepare()
- args[0].assert_called_once_with(
- mock.ANY, self.tenant_name, self.tenant_description)
- args[1].assert_called_once_with()
-
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client')
- @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
- return_value=0)
- @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf',
- side_effect=Exception)
- def test_prepare_user_creation_ko(self, *args):
- with self.assertRaises(vnf.VnfPreparationException):
- self.test.prepare()
- args[0].assert_called_once_with(mock.ANY, self.tenant_name)
- args[1].assert_called_once_with(
- mock.ANY, self.tenant_name, self.tenant_description)
- args[2].assert_called_once_with()
+ args[0].assert_called_once()
+
+# @mock.patch('snaps.openstack.tests.openstack_tests.get_credentials',
+# return_value=OS_CREDS)
+# @mock.patch('snaps.openstack.create_project.OpenStackProject')
+# @mock.patch('snaps.openstack.create_project.OpenStackProject.create',
+# side_effect=Exception)
+# def test_prepare_tenant_creation_ko(self, *args):
+# with self.assertRaises(vnf.VnfPreparationException):
+# self.test.prepare()
+# args[2].assert_called_once()
+# args[1].assert_called_once_with(OS_CREDS,
+# ProjectSettings(
+# name=self.tenant_name,
+# description=self.tenant_description,
+# ))
+# args[0].assert_called_once()
+
+# @mock.patch('snaps.openstack.tests.openstack_tests.get_credentials')
+# @mock.patch('snaps.openstack.create_project.OpenStackProject',
+# return_value=0)
+# @mock.patch('snaps.openstack.create_user.OpenStackUser',
+# side_effect=Exception)
+# def test_prepare_user_creation_ko(self, *args):
+# with self.assertRaises(vnf.VnfPreparationException):
+# self.test.prepare()
+# args[0].assert_called_once_with(mock.ANY, self.tenant_name)
+# args[1].assert_called_once_with(
+# mock.ANY, self.tenant_name, self.tenant_description)
+# args[2].assert_called_once_with()
if __name__ == "__main__":
diff --git a/functest/tests/unit/utils/test_openstack_tacker.py b/functest/tests/unit/utils/test_openstack_tacker.py
deleted file mode 100644
index 3c0fc3d0d..000000000
--- a/functest/tests/unit/utils/test_openstack_tacker.py
+++ /dev/null
@@ -1,524 +0,0 @@
-#!/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
-
-import logging
-import unittest
-
-import mock
-from tackerclient.v1_0 import client as tackerclient
-
-from functest.utils import openstack_tacker
-from functest.tests.unit import test_utils
-
-
-class OSTackerTesting(unittest.TestCase):
-
- def setUp(self):
- self.tacker_client = mock.Mock()
- self.getresponse = {'vnfds': [{'id': 'test_id'}],
- 'vnfs': [{'id': 'test_id'}],
- 'sfcs': [{'id': 'test_id'}]}
- self.vnfdlist = {'vnfds': [{'id': 'test_vnfd1'}, {'id': 'test_vnfd2'}]}
- self.vnflist = {'vnfs': [{'id': 'test_vnf1'}, {'id': 'test_vnf2'}]}
- self.sfclist = {'sfcs': [{'id': 'test_sfc1'}, {'id': 'test_sfc2'}]}
- self.sfc_classifierlist = {'sfc_classifiers': [{'id': 'test_sfc_cl1'},
- {'id': 'test_sfc_cl2'}]}
-
- self.createvnfd = {"vnfd": {"attributes": {"vnfd": 'vnfd_body'}}}
- self.createvnf = {"vnf": {"attributes": {"vnf": 'vnf_body'}}}
- self.createsfc = {"sfc": {"attributes": {"sfc": 'sfc_body'}}}
- self.createsfc_clf = {"sfc_classifier": {"attributes":
- {"sfc_clf": 'sfc_clf_body'}}}
-
- self.resource_type = 'vnfd'
- self.resource_name = 'resource_name'
- self.tosca_file = 'test_tosca_file'
- self.vnfd = 'test_vnfd'
- self.vnf = 'test_vnf'
- self.sfc = 'test_sfc'
- self.sfc_clf = 'test_sfc_clf'
-
- def _get_creds(self):
- cred_dict = {
- 'OS_USERNAME': 'username',
- 'OS_PASSWORD': 'password',
- 'OS_AUTH_URL': 'auth_url',
- 'OS_TENANT_NAME': 'tenant_name',
- 'OS_USER_DOMAIN_NAME': 'user_domain_name',
- 'OS_PROJECT_DOMAIN_NAME': 'project_domain_name',
- 'OS_PROJECT_NAME': 'project_name',
- 'OS_ENDPOINT_TYPE': 'endpoint_type',
- 'OS_REGION_NAME': 'region_name'
- }
- return cred_dict
-
- def test_get_tacker_client(self):
- with mock.patch('functest.utils.openstack_tacker.'
- 'os_utils.get_session'):
- tackerclient.Client = mock.Mock
- ret = openstack_tacker.get_tacker_client()
- self.assertTrue(isinstance(ret, mock.Mock))
-
- def test_get_id_from_name(self):
- with mock.patch.object(self.tacker_client, 'get',
- return_value=self.getresponse):
- resp = openstack_tacker.get_id_from_name(self.tacker_client,
- self.resource_type,
- self.resource_name)
- self.assertEqual(resp, 'test_id')
-
- @mock.patch('functest.utils.openstack_tacker.logger.error')
- def test_get_id_from_name_exception(self, mock_logger_error):
- with mock.patch.object(self.tacker_client, 'get',
- side_effect=Exception):
- resp = openstack_tacker.get_id_from_name(self.tacker_client,
- self.resource_type,
- self.resource_name)
- self.assertIsNone(resp)
- mock_logger_error.assert_called_once_with(test_utils.
- SubstrMatch("Error [get"
- "_id_from_"
- "name(tacker"
- "_client"
- ", resource_"
- "type, "
- "resource_"
- "name)]:"))
-
- @mock.patch('functest.utils.openstack_tacker.get_id_from_name')
- def test_get_vnfd_id(self, mock_get_id_from_name):
- openstack_tacker.get_vnfd_id(self.tacker_client, self.resource_name)
- mock_get_id_from_name.assert_called_once_with(self.tacker_client,
- 'vnfd',
- self.resource_name)
-
- @mock.patch('functest.utils.openstack_tacker.get_id_from_name')
- def test_get_vnf_id(self, mock_get_id_from_name):
- openstack_tacker.get_vnf_id(self.tacker_client, self.resource_name)
- mock_get_id_from_name.assert_called_once_with(self.tacker_client,
- 'vnf',
- self.resource_name)
-
- @mock.patch('functest.utils.openstack_tacker.get_id_from_name')
- def test_get_sfc_id(self, mock_get_id_from_name):
- openstack_tacker.get_sfc_id(self.tacker_client, self.resource_name)
- mock_get_id_from_name.assert_called_once_with(self.tacker_client,
- 'sfc',
- self.resource_name)
-
- @mock.patch('functest.utils.openstack_tacker.get_id_from_name')
- def test_get_sfc_classifier_id(self, mock_get_id_from_name):
- openstack_tacker.get_sfc_classifier_id(self.tacker_client,
- self.resource_name)
- mock_get_id_from_name.assert_called_once_with(self.tacker_client,
- 'sfc-classifier',
- self.resource_name)
-
- def test_list_vnfds(self):
- with mock.patch.object(self.tacker_client, 'list_vnfds',
- return_value=self.vnfdlist):
- resp = openstack_tacker.list_vnfds(self.tacker_client,
- verbose=False)
- self.assertEqual(resp, ['test_vnfd1', 'test_vnfd2'])
-
- def test_list_vnfds_verbose(self):
- with mock.patch.object(self.tacker_client, 'list_vnfds',
- return_value=self.vnfdlist):
- resp = openstack_tacker.list_vnfds(self.tacker_client,
- verbose=True)
- self.assertEqual(resp, self.vnfdlist)
-
- @mock.patch('functest.utils.openstack_tacker.logger.error')
- def test_list_vnfds_exception(self, mock_logger_error):
- with mock.patch.object(self.tacker_client, 'list_vnfds',
- side_effect=Exception):
- resp = openstack_tacker.list_vnfds(self.tacker_client,
- verbose=False)
- mock_logger_error.assert_called_once_with(test_utils.
- SubstrMatch("Error"
- " [list"
- "_vnfds("
- "tacker_"
- "client)]:"))
- self.assertIsNone(resp)
-
- def test_create_vnfd_missing_file(self):
- with mock.patch.object(self.tacker_client, 'create_vnfd',
- return_value=self.createvnfd):
- resp = openstack_tacker.create_vnfd(self.tacker_client,
- tosca_file=None)
- self.assertEqual(resp, self.createvnfd)
-
- def test_create_vnfd_default(self):
- with mock.patch.object(self.tacker_client, 'create_vnfd',
- return_value=self.createvnfd), \
- mock.patch('__builtin__.open', mock.mock_open(read_data='1')) \
- as m:
- resp = openstack_tacker.create_vnfd(self.tacker_client,
- tosca_file=self.tosca_file)
- m.assert_called_once_with(self.tosca_file)
- self.assertEqual(resp, self.createvnfd)
-
- @mock.patch('functest.utils.openstack_tacker.logger.error')
- def test_create_vnfd_exception(self, mock_logger_error):
- with mock.patch.object(self.tacker_client, 'create_vnfd',
- side_effect=Exception):
- resp = openstack_tacker.create_vnfd(self.tacker_client,
- tosca_file=self.tosca_file)
- mock_logger_error.assert_called_once_with(test_utils.
- SubstrMatch("Error"
- " [create"
- "_vnfd("
- "tacker_"
- "client, "
- "'%s')]"
- % self.
- tosca_file))
- self.assertIsNone(resp)
-
- def test_delete_vnfd(self):
- with mock.patch('functest.utils.openstack_tacker.get_vnfd_id',
- return_value=self.vnfd), \
- mock.patch.object(self.tacker_client, 'delete_vnfd',
- return_value=self.vnfd):
- resp = openstack_tacker.delete_vnfd(self.tacker_client,
- vnfd_id='vnfd_id',
- vnfd_name=self.vnfd)
- self.assertEqual(resp, self.vnfd)
-
- def test_delete_vnfd_missing_vnfd_name(self):
- with mock.patch('functest.utils.openstack_tacker.get_vnfd_id',
- return_value=self.vnfd), \
- self.assertRaises(Exception) as context:
- resp = openstack_tacker.delete_vnfd(self.tacker_client,
- vnfd_id=None,
- vnfd_name=None)
- self.assertIsNone(resp)
- msg = 'You need to provide VNFD id or VNFD name'
- self.assertTrue(msg in context)
-
- @mock.patch('functest.utils.openstack_tacker.logger.error')
- def test_delete_vnfd_exception(self, mock_logger_error):
- with mock.patch('functest.utils.openstack_tacker.get_vnfd_id',
- return_value=self.vnfd), \
- mock.patch.object(self.tacker_client, 'delete_vnfd',
- side_effect=Exception):
- resp = openstack_tacker.delete_vnfd(self.tacker_client,
- vnfd_id=None,
- vnfd_name=None)
- self.assertIsNone(resp)
- self.assertTrue(mock_logger_error.called)
-
- def test_list_vnfs(self):
- with mock.patch.object(self.tacker_client, 'list_vnfs',
- return_value=self.vnflist):
- resp = openstack_tacker.list_vnfs(self.tacker_client,
- verbose=False)
- self.assertEqual(resp, ['test_vnf1', 'test_vnf2'])
-
- def test_list_vnfs_verbose(self):
- with mock.patch.object(self.tacker_client, 'list_vnfs',
- return_value=self.vnflist):
- resp = openstack_tacker.list_vnfs(self.tacker_client,
- verbose=True)
- self.assertEqual(resp, self.vnflist)
-
- @mock.patch('functest.utils.openstack_tacker.logger.error')
- def test_list_vnfs_exception(self, mock_logger_error):
- with mock.patch.object(self.tacker_client, 'list_vnfs',
- side_effect=Exception):
- resp = openstack_tacker.list_vnfs(self.tacker_client,
- verbose=False)
- mock_logger_error.assert_called_once_with(test_utils.
- SubstrMatch("Error"
- " [list"
- "_vnfs("
- "tacker_"
- "client)]:"))
- self.assertIsNone(resp)
-
- def test_create_vnf_default(self):
- with mock.patch.object(self.tacker_client, 'create_vnf',
- return_value=self.createvnf), \
- mock.patch('functest.utils.openstack_tacker.get_vnfd_id',
- return_value=self.vnf):
- resp = openstack_tacker.create_vnf(self.tacker_client,
- vnf_name=self.vnf,
- vnfd_id='vnfd_id',
- vnfd_name=self.vnfd)
- self.assertEqual(resp, self.createvnf)
-
- @mock.patch('functest.utils.openstack_tacker.logger.error')
- def test_create_vnf_exception(self, mock_logger_error):
- with mock.patch.object(self.tacker_client, 'create_vnf',
- side_effect=Exception):
- resp = openstack_tacker.create_vnf(self.tacker_client,
- vnf_name=self.vnf,
- vnfd_id='vnfd_id',
- vnfd_name=self.vnfd)
- mock_logger_error.assert_called_once_with(test_utils.
- SubstrMatch("error"
- " [create"
- "_vnf("
- "tacker_"
- "client"))
- self.assertIsNone(resp)
-
- def test_wait_for_vnf_vnf_retrieval_failed(self):
- with mock.patch('functest.utils.openstack_tacker.get_vnf',
- return_value=None), \
- self.assertRaises(Exception) as context:
- openstack_tacker.wait_for_vnf(self.tacker_client,
- vnf_id='vnf_id',
- vnf_name='vnf_name')
- msg = ("Could not retrieve VNF - id='vnf_id', "
- "name='vnf_name'")
- self.assertTrue(msg in context)
- with mock.patch('functest.utils.openstack_tacker.get_vnf',
- side_effect=Exception):
- ret = openstack_tacker.wait_for_vnf(self.tacker_client,
- vnf_id='vnf_id',
- vnf_name='vnf_name')
- self.assertEqual(ret, None)
-
- def test_wait_for_vnf_vnf_status_error(self):
- vnf = {'id': 'vnf_id',
- 'status': 'ERROR'}
- with mock.patch('functest.utils.openstack_tacker.get_vnf',
- return_value=vnf), \
- self.assertRaises(Exception) as context:
- openstack_tacker.wait_for_vnf(self.tacker_client,
- vnf_id='vnf_id',
- vnf_name='vnf_name')
- msg = ('Error when booting vnf vnf_id')
- self.assertTrue(msg in context)
-
- def test_wait_for_vnf_vnf_timeout(self):
- vnf = {'id': 'vnf_id',
- 'status': 'PENDING_CREATE'}
- with mock.patch('functest.utils.openstack_tacker.get_vnf',
- return_value=vnf), \
- self.assertRaises(Exception) as context:
- openstack_tacker.wait_for_vnf(self.tacker_client,
- vnf_id='vnf_id',
- vnf_name='vnf_name',
- timeout=2)
- msg = ('Timeout when booting vnf vnf_id')
- self.assertTrue(msg in context)
-
- def test_delete_vnf(self):
- with mock.patch('functest.utils.openstack_tacker.get_vnf_id',
- return_value=self.vnf), \
- mock.patch.object(self.tacker_client, 'delete_vnf',
- return_value=self.vnf):
- resp = openstack_tacker.delete_vnf(self.tacker_client,
- vnf_id='vnf_id',
- vnf_name=self.vnf)
- self.assertEqual(resp, self.vnf)
-
- def test_delete_vnf_missing_vnf_name(self):
- with self.assertRaises(Exception) as context:
- openstack_tacker.delete_vnf(self.tacker_client,
- vnf_id=None,
- vnf_name=None)
- msg = 'You need to provide a VNF id or name'
- self.assertTrue(msg in context)
-
- @mock.patch('functest.utils.openstack_tacker.logger.error')
- def test_delete_vnf_exception(self, mock_logger_error):
- with mock.patch('functest.utils.openstack_tacker.get_vnf_id',
- return_value=self.vnf), \
- mock.patch.object(self.tacker_client, 'delete_vnf',
- side_effect=Exception):
- resp = openstack_tacker.delete_vnf(self.tacker_client,
- vnf_id=None,
- vnf_name=None)
- self.assertIsNone(resp)
- self.assertTrue(mock_logger_error.called)
-
- def test_list_sfcs(self):
- with mock.patch.object(self.tacker_client, 'list_sfcs',
- return_value=self.sfclist):
- resp = openstack_tacker.list_sfcs(self.tacker_client,
- verbose=False)
- self.assertEqual(resp, ['test_sfc1', 'test_sfc2'])
-
- def test_list_sfcs_verbose(self):
- with mock.patch.object(self.tacker_client, 'list_sfcs',
- return_value=self.sfclist):
- resp = openstack_tacker.list_sfcs(self.tacker_client,
- verbose=True)
- self.assertEqual(resp, self.sfclist)
-
- @mock.patch('functest.utils.openstack_tacker.logger.error')
- def test_list_sfcs_exception(self, mock_logger_error):
- with mock.patch.object(self.tacker_client, 'list_sfcs',
- side_effect=Exception):
- resp = openstack_tacker.list_sfcs(self.tacker_client,
- verbose=False)
- mock_logger_error.assert_called_once_with(test_utils.
- SubstrMatch("Error"
- " [list"
- "_sfcs("
- "tacker_"
- "client)]:"))
- self.assertIsNone(resp)
-
- def test_create_sfc_default(self):
- with mock.patch.object(self.tacker_client, 'create_sfc',
- return_value=self.createsfc), \
- mock.patch('functest.utils.openstack_tacker.get_vnf_id',
- return_value=self.vnf):
- resp = openstack_tacker.create_sfc(self.tacker_client,
- sfc_name=self.sfc,
- chain_vnf_ids=['chain_vnf_id'],
- chain_vnf_names=[self.vnf])
- self.assertEqual(resp, self.createsfc)
-
- @mock.patch('functest.utils.openstack_tacker.logger.error')
- def test_create_sfc_exception(self, mock_logger_error):
- with mock.patch.object(self.tacker_client, 'create_sfc',
- side_effect=Exception):
- resp = openstack_tacker.create_sfc(self.tacker_client,
- sfc_name=self.sfc,
- chain_vnf_ids=['chain_vnf_id'],
- chain_vnf_names=[self.vnf])
- mock_logger_error.assert_called_once_with(test_utils.
- SubstrMatch("error"
- " [create"
- "_sfc("
- "tacker_"
- "client"))
- self.assertIsNone(resp)
-
- def test_delete_sfc(self):
- with mock.patch('functest.utils.openstack_tacker.get_sfc_id',
- return_value=self.sfc), \
- mock.patch.object(self.tacker_client, 'delete_sfc',
- return_value=self.sfc):
- resp = openstack_tacker.delete_sfc(self.tacker_client,
- sfc_id='sfc_id',
- sfc_name=self.sfc)
- self.assertEqual(resp, self.sfc)
-
- def test_delete_sfc_missing_sfc_name(self):
- with self.assertRaises(Exception) as context:
- openstack_tacker.delete_sfc(self.tacker_client,
- sfc_id=None,
- sfc_name=None)
- msg = 'You need to provide an SFC id or name'
- self.assertTrue(msg in context)
-
- @mock.patch('functest.utils.openstack_tacker.logger.error')
- def test_delete_sfc_exception(self, mock_logger_error):
- with mock.patch('functest.utils.openstack_tacker.get_sfc_id',
- return_value=self.sfc), \
- mock.patch.object(self.tacker_client, 'delete_sfc',
- side_effect=Exception):
- resp = openstack_tacker.delete_sfc(self.tacker_client,
- sfc_id=None,
- sfc_name=None)
- self.assertIsNone(resp)
- self.assertTrue(mock_logger_error.called)
-
- def test_list_sfc_classifiers(self):
- with mock.patch.object(self.tacker_client, 'list_sfc_classifiers',
- return_value=self.sfc_classifierlist):
- resp = openstack_tacker.list_sfc_classifiers(self.tacker_client,
- verbose=False)
- self.assertEqual(resp, ['test_sfc_cl1', 'test_sfc_cl2'])
-
- def test_list_sfc_classifiers_verbose(self):
- with mock.patch.object(self.tacker_client, 'list_sfc_classifiers',
- return_value=self.sfc_classifierlist):
- resp = openstack_tacker.list_sfc_classifiers(self.tacker_client,
- verbose=True)
- self.assertEqual(resp, self.sfc_classifierlist)
-
- @mock.patch('functest.utils.openstack_tacker.logger.error')
- def test_list_sfc_classifiers_exception(self, mock_logger_error):
- with mock.patch.object(self.tacker_client, 'list_sfc_classifiers',
- side_effect=Exception):
- resp = openstack_tacker.list_sfc_classifiers(self.tacker_client,
- verbose=False)
- mock_logger_error.assert_called_once_with(test_utils.
- SubstrMatch("Error"
- " [list"
- "_sfc_cl"
- "assifiers("
- "tacker_"
- "client)]:"))
- self.assertIsNone(resp)
-
- def test_create_sfc_classifier_default(self):
- with mock.patch.object(self.tacker_client, 'create_sfc_classifier',
- return_value=self.createsfc_clf), \
- mock.patch('functest.utils.openstack_tacker.get_sfc_id',
- return_value=self.sfc):
- cl = self.sfc_clf
- resp = openstack_tacker.create_sfc_classifier(self.tacker_client,
- sfc_clf_name=cl,
- sfc_id='sfc_id',
- sfc_name=self.sfc)
- self.assertEqual(resp, self.createsfc_clf)
-
- @mock.patch('functest.utils.openstack_tacker.logger.error')
- def test_sfc_classifier_exception(self, mock_logger_error):
- with mock.patch.object(self.tacker_client, 'create_sfc_classifier',
- side_effect=Exception):
- cl = self.sfc_clf
- resp = openstack_tacker.create_sfc_classifier(self.tacker_client,
- sfc_clf_name=cl,
- sfc_id='sfc_id',
- sfc_name=self.sfc)
- mock_logger_error.assert_called_once_with(test_utils.
- SubstrMatch("error"
- " [create"
- "_sfc_cl"
- "assifier("
- "tacker_"
- "client"))
- self.assertIsNone(resp)
-
- def test_delete_sfc_classifier(self):
- with mock.patch('functest.utils.openstack_tacker.get_sfc_'
- 'classifier_id',
- return_value=self.sfc_clf), \
- mock.patch.object(self.tacker_client, 'delete_sfc_classifier',
- return_value=self.sfc_clf):
- cl = self.sfc_clf
- resp = openstack_tacker.delete_sfc_classifier(self.tacker_client,
- sfc_clf_id='sfc_id',
- sfc_clf_name=cl)
- self.assertEqual(resp, cl)
-
- def test_delete_sfc_classifier_missing_sfc_name(self):
- with self.assertRaises(Exception) as context:
- openstack_tacker.delete_vnf(self.tacker_client,
- sfc_clf_id=None,
- sfc_clf_name=None)
- msg = 'You need to provide an SFCclassifier id or name'
- self.assertTrue(msg in context)
-
- @mock.patch('functest.utils.openstack_tacker.logger.error')
- def test_delete_sfc_classifier_exception(self, mock_logger_error):
- with mock.patch('functest.utils.openstack_tacker.get_sfc_'
- 'classifier_id',
- return_value=self.sfc_clf), \
- mock.patch.object(self.tacker_client, 'delete_sfc_classifier',
- side_effect=Exception):
- cl = self.sfc_clf
- resp = openstack_tacker.delete_sfc_classifier(self.tacker_client,
- sfc_clf_id='sfc_id',
- sfc_clf_name=cl)
- self.assertIsNone(resp)
- self.assertTrue(mock_logger_error.called)
-
-
-if __name__ == "__main__":
- logging.disable(logging.CRITICAL)
- unittest.main(verbosity=2)
diff --git a/functest/tests/unit/vnf/ims/test_cloudify_ims.py b/functest/tests/unit/vnf/ims/test_cloudify_ims.py
index f0483c69f..cdd657aac 100644
--- a/functest/tests/unit/vnf/ims/test_cloudify_ims.py
+++ b/functest/tests/unit/vnf/ims/test_cloudify_ims.py
@@ -13,8 +13,6 @@ import mock
from functest.core import vnf
from functest.opnfv_tests.vnf.ims import cloudify_ims
-from snaps.openstack.os_credentials import OSCreds
-
class CloudifyImsTesting(unittest.TestCase):
@@ -57,114 +55,10 @@ class CloudifyImsTesting(unittest.TestCase):
'vnf': {},
'test_vnf': {}}
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
- return_value='test')
- @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_credentials',
- return_value={'auth_url': 'test/v1'})
- @mock.patch('snaps.openstack.create_image.OpenStackImage.create')
- def test_prepare_default(self, *args):
- self.assertIsNone(self.ims_vnf.prepare())
- args[4].assert_called_once_with()
-
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
- return_value='test')
- @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_credentials',
- return_value={'auth_url': 'test/no_v'})
- @mock.patch('snaps.openstack.create_image.OpenStackImage.create')
- def test_prepare_bad_auth_url(self, *args):
- with self.assertRaises(Exception):
- self.ims_vnf.image_creator(
- OSCreds(username='user', password='pass', auth_url='url',
- project_name='project', identity_api_version=3),
- mock.Mock())
- args[0].assert_not_called()
-
def test_prepare_missing_param(self):
with self.assertRaises(vnf.VnfPreparationException):
self.ims_vnf.prepare()
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
- side_effect=Exception)
- def test_prepare_keystone_exception(self, *args):
- with self.assertRaises(vnf.VnfPreparationException):
- self.ims_vnf.prepare()
- args[0].assert_called_once_with()
-
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
- return_value='test')
- @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
- side_effect=Exception)
- def test_prepare_tenant_exception(self, *args):
- with self.assertRaises(vnf.VnfPreparationException):
- self.ims_vnf.prepare()
- args[1].assert_called_once_with()
-
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
- return_value='test')
- @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf',
- side_effect=Exception)
- def test_prepare_user_exception(self, *args):
- with self.assertRaises(vnf.VnfPreparationException):
- self.ims_vnf.prepare()
- args[2].assert_called_once_with()
-
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
- return_value='test')
- @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_credentials',
- side_effect=Exception)
- def test_prepare_credentials_exception(self, *args):
- with self.assertRaises(vnf.VnfPreparationException):
- self.ims_vnf.prepare()
- args[0].assert_called_once_with()
-
- # @mock.patch('snaps.openstack.create_keypairs.OpenStackKeypair',
- # side_effect=Exception)
- # def test_deploy_orchestrator_keypair_exception(self, *args):
- # with self.assertRaises(vnf.OrchestratorDeploymentException):
- # self.ims_vnf.deploy_orchestrator()
-
- # def test_deploy_orchestrator_network_creation_fail(self):
- # def test_deploy_orchestrator_floatting_ip_creation_fail(self):
- # def test_deploy_orchestrator_flavor_fail(self):
- # def test_deploy_orchestrator_get_image_id_fail(self):
- # def test_deploy_orchestrator_create_instance_fail(self):
- # def test_deploy_orchestrator_secgroup_fail(self):
- # def test_deploy_orchestrator_add_floating_ip_fail(self):
- # def test_deploy_orchestrator_get_endpoint_fail(self):
- # def test_deploy_orchestrator_initiate CloudifyClient_fail(self):
- # def test_deploy_orchestrator_get_status_fail(self):
- #
-
- # def test_deploy_vnf(self):
- # def test_deploy_vnf_publish_fail(self):
- # def test_deploy_vnf_get_flavor_fail(self):
- # def test_deploy_vnf_get_external_net_fail(self):
- # def test_deploy_vnf_deployment_create_fail(self):
- # def test_deploy_vnf_start_fail(self):
- #
- # def test_test_vnf(self):
- # def test_test_vnf_deployment_get_fail(self):
- # def test_test_vnf_run_live_test_fail(self):
- #
- # def test_clean(self):
- # def test_clean_execution_start_fail(self):
- # def test_clean_deployment_delete_fail(self):
- # def test_clean_blueprint_delete_fail(self):
-
if __name__ == "__main__":
logging.disable(logging.CRITICAL)
diff --git a/functest/tests/unit/vnf/ims/test_orchestra_clearwaterims.py b/functest/tests/unit/vnf/ims/test_orchestra_clearwaterims.py
index ef227ca43..2e83f30a4 100644
--- a/functest/tests/unit/vnf/ims/test_orchestra_clearwaterims.py
+++ b/functest/tests/unit/vnf/ims/test_orchestra_clearwaterims.py
@@ -11,7 +11,6 @@ import logging
import unittest
import mock
-from snaps.openstack.os_credentials import OSCreds
from functest.core import vnf
from functest.opnfv_tests.vnf.ims import orchestra_clearwaterims
@@ -110,117 +109,11 @@ class OrchestraClearwaterImsTesting(unittest.TestCase):
'vnf': {},
'test_vnf': {}}
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
- return_value='test')
- @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_credentials',
- return_value={'auth_url': 'test/v1'})
- @mock.patch(
- 'functest.utils.openstack_utils.get_tenant_id',
- return_value={'mocked_tenant_id'})
- @mock.patch(
- 'functest.utils.openstack_utils.get_floating_ips',
- return_value=[])
- @mock.patch('snaps.openstack.create_image.OpenStackImage.create')
- @mock.patch('snaps.openstack.create_flavor.OpenStackFlavor.create')
- @mock.patch(
- 'snaps.openstack.create_security_group.OpenStackSecurityGroup.create')
- @mock.patch('snaps.openstack.create_network.OpenStackNetwork.create')
- @mock.patch('snaps.openstack.create_router.OpenStackRouter.create')
- @mock.patch(
- 'functest.opnfv_tests.openstack.snaps.snaps_utils.get_ext_net_name')
- @mock.patch(
- 'functest.opnfv_tests.openstack.snaps.'
- 'snaps_utils.neutron_utils.create_floating_ip')
- def test_prepare_default(self, *args):
- """Testing prepare function without any exceptions expected"""
- self.assertIsNone(self.ims_vnf.prepare())
- args[4].assert_called_once_with()
-
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
- return_value='test')
- @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_credentials',
- return_value={'auth_url': 'test/no_v'})
- @mock.patch('snaps.openstack.create_image.OpenStackImage.create')
- def test_prepare_bad_auth_url(self, *args):
- """Testing prepare function with bad auth url"""
- with self.assertRaises(Exception):
- self.ims_vnf.image_creator(
- OSCreds(username='user', password='pass', auth_url='url',
- project_name='project', identity_api_version=3),
- mock.Mock())
- args[0].assert_not_called()
-
def test_prepare_missing_param(self):
"""Testing prepare function with missing param"""
with self.assertRaises(vnf.VnfPreparationException):
self.ims_vnf.prepare()
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
- side_effect=Exception)
- def test_prepare_keystone_exception(self, *args):
- """Testing prepare function with keystone exception"""
- with self.assertRaises(vnf.VnfPreparationException):
- self.ims_vnf.prepare()
- args[0].assert_called_once_with()
-
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
- return_value='test')
- @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
- side_effect=Exception)
- def test_prepare_tenant_exception(self, *args):
- """Testing prepare function with tenant exception"""
- with self.assertRaises(vnf.VnfPreparationException):
- self.ims_vnf.prepare()
- args[1].assert_called_once_with()
-
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
- return_value='test')
- @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf',
- side_effect=Exception)
- def test_prepare_user_exception(self, *args):
- """Testing prepare function with user exception"""
- with self.assertRaises(vnf.VnfPreparationException):
- self.ims_vnf.prepare()
- args[2].assert_called_once_with()
-
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
- return_value='test')
- @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_credentials',
- side_effect=Exception)
- def test_prepare_credentials_exception(self, *args):
- """Testing prepare function with credentials exception"""
- with self.assertRaises(vnf.VnfPreparationException):
- self.ims_vnf.prepare()
- args[0].assert_called_once_with()
-
- # # @mock.patch('functest.opnfv_tests.vnf.
- # ims.orchestra_clearwaterims.get_userdata')
- # def test_deploy_orchestrator(self, *args):
- # floating_ip = FloatingIp
- # floating_ip.ip = 'mocked_ip'
- # details = {'fip':floating_ip,'flavor':{'name':'mocked_name'}}
- # self.mano['details'] = details
- # with mock.patch.dict(self.mano, {'details':
- # {'fip':floating_ip,'flavor':{'name':'mocked_name'}}}):
- # # with mock.patch.dict(self.mano, details):
- # orchestra_clearwaterims.get_userdata(self.mano)
- # self.assertIsNone(self.ims_vnf.deploy_orchestrator())
- # args[4].assert_called_once_with()
-
if __name__ == "__main__":
logging.disable(logging.CRITICAL)
diff --git a/functest/tests/unit/vnf/ims/test_orchestra_openims.py b/functest/tests/unit/vnf/ims/test_orchestra_openims.py
index 5911cf77b..47a8d0338 100644
--- a/functest/tests/unit/vnf/ims/test_orchestra_openims.py
+++ b/functest/tests/unit/vnf/ims/test_orchestra_openims.py
@@ -11,7 +11,6 @@ import logging
import unittest
import mock
-from snaps.openstack.os_credentials import OSCreds
from functest.core import vnf
from functest.opnfv_tests.vnf.ims import orchestra_openims
@@ -112,117 +111,11 @@ class OrchestraOpenImsTesting(unittest.TestCase):
'vnf': {},
'test_vnf': {}}
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
- return_value='test')
- @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_credentials',
- return_value={'auth_url': 'test/v1'})
- @mock.patch(
- 'functest.utils.openstack_utils.get_tenant_id',
- return_value={'mocked_tenant_id'})
- @mock.patch(
- 'functest.utils.openstack_utils.get_floating_ips',
- return_value=[])
- @mock.patch('snaps.openstack.create_image.OpenStackImage.create')
- @mock.patch('snaps.openstack.create_flavor.OpenStackFlavor.create')
- @mock.patch(
- 'snaps.openstack.create_security_group.OpenStackSecurityGroup.create')
- @mock.patch('snaps.openstack.create_network.OpenStackNetwork.create')
- @mock.patch('snaps.openstack.create_router.OpenStackRouter.create')
- @mock.patch(
- 'functest.opnfv_tests.openstack.snaps.snaps_utils.get_ext_net_name')
- @mock.patch(
- 'functest.opnfv_tests.openstack.snaps.snaps_utils.'
- 'neutron_utils.create_floating_ip')
- def test_prepare_default(self, *args):
- """Testing prepare function without any exceptions expected"""
- self.assertIsNone(self.ims_vnf.prepare())
- args[4].assert_called_once_with()
-
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
- return_value='test')
- @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_credentials',
- return_value={'auth_url': 'test/no_v'})
- @mock.patch('snaps.openstack.create_image.OpenStackImage.create')
- def test_prepare_bad_auth_url(self, *args):
- """Testing prepare function with bad auth url"""
- with self.assertRaises(Exception):
- self.ims_vnf.image_creator(
- OSCreds(username='user', password='pass', auth_url='url',
- project_name='project', identity_api_version=3),
- mock.Mock())
- args[0].assert_not_called()
-
def test_prepare_missing_param(self):
"""Testing prepare function with missing param"""
with self.assertRaises(vnf.VnfPreparationException):
self.ims_vnf.prepare()
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
- side_effect=Exception)
- def test_prepare_keystone_exception(self, *args):
- """Testing prepare function with keystone exception"""
- with self.assertRaises(vnf.VnfPreparationException):
- self.ims_vnf.prepare()
- args[0].assert_called_once_with()
-
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
- return_value='test')
- @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
- side_effect=Exception)
- def test_prepare_tenant_exception(self, *args):
- """Testing prepare function with tenant exception"""
- with self.assertRaises(vnf.VnfPreparationException):
- self.ims_vnf.prepare()
- args[1].assert_called_once_with()
-
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
- return_value='test')
- @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf',
- side_effect=Exception)
- def test_prepare_user_exception(self, *args):
- """Testing prepare function with user exception"""
- with self.assertRaises(vnf.VnfPreparationException):
- self.ims_vnf.prepare()
- args[2].assert_called_once_with()
-
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
- return_value='test')
- @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_credentials',
- side_effect=Exception)
- def test_prepare_credentials_exception(self, *args):
- """Testing prepare function with credentials exception"""
- with self.assertRaises(vnf.VnfPreparationException):
- self.ims_vnf.prepare()
- args[0].assert_called_once_with()
-
- # # @mock.patch('functest.opnfv_tests.
- # vnf.ims.orchestra_openims.get_userdata')
- # def test_deploy_orchestrator(self, *args):
- # floating_ip = FloatingIp
- # floating_ip.ip = 'mocked_ip'
- # details = {'fip':floating_ip,'flavor':{'name':'mocked_name'}}
- # self.mano['details'] = details
- # with mock.patch.dict(self.mano, {'details':
- # {'fip':floating_ip,'flavor':{'name':'mocked_name'}}}):
- # # with mock.patch.dict(self.mano, details):
- # orchestra_openims.get_userdata(self.mano)
- # self.assertIsNone(self.ims_vnf.deploy_orchestrator())
- # args[4].assert_called_once_with()
-
if __name__ == "__main__":
logging.disable(logging.CRITICAL)
diff --git a/functest/tests/unit/vnf/router/test_cloudify_vrouter.py b/functest/tests/unit/vnf/router/test_cloudify_vrouter.py
index 7f2091be1..4f256234b 100644
--- a/functest/tests/unit/vnf/router/test_cloudify_vrouter.py
+++ b/functest/tests/unit/vnf/router/test_cloudify_vrouter.py
@@ -15,8 +15,6 @@ import mock
from functest.core import vnf
from functest.opnfv_tests.vnf.router import cloudify_vrouter
-from snaps.openstack.os_credentials import OSCreds
-
class CloudifyVrouterTesting(unittest.TestCase):
@@ -60,80 +58,10 @@ class CloudifyVrouterTesting(unittest.TestCase):
'vnf': {},
'test_vnf': {}}
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
- return_value='test')
- @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_credentials',
- return_value={'auth_url': 'test/v1'})
- @mock.patch('snaps.openstack.create_image.OpenStackImage.create')
- def test_prepare_default(self, *args):
- self.assertIsNone(self.router_vnf.prepare())
- args[4].assert_called_once_with()
-
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
- return_value='test')
- @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_credentials',
- return_value={'auth_url': 'test/no_v'})
- @mock.patch('snaps.openstack.create_image.OpenStackImage.create')
- def test_prepare_bad_auth_url(self, *args):
- with self.assertRaises(Exception):
- self.router_vnf.image_creator(
- OSCreds(username='user', password='pass', auth_url='url',
- project_name='project', identity_api_version=3),
- mock.Mock())
- args[0].assert_not_called()
-
def test_prepare_missing_param(self):
with self.assertRaises(vnf.VnfPreparationException):
self.router_vnf.prepare()
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
- side_effect=Exception)
- def test_prepare_keystone_exception(self, *args):
- with self.assertRaises(vnf.VnfPreparationException):
- self.router_vnf.prepare()
- args[0].assert_called_once_with()
-
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
- return_value='test')
- @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
- side_effect=Exception)
- def test_prepare_tenant_exception(self, *args):
- with self.assertRaises(vnf.VnfPreparationException):
- self.router_vnf.prepare()
- args[1].assert_called_once_with()
-
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
- return_value='test')
- @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf',
- side_effect=Exception)
- def test_prepare_user_exception(self, *args):
- with self.assertRaises(vnf.VnfPreparationException):
- self.router_vnf.prepare()
- args[2].assert_called_once_with()
-
- @mock.patch('functest.core.vnf.os_utils.get_keystone_client',
- return_value='test')
- @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf',
- return_value=True)
- @mock.patch('functest.core.vnf.os_utils.get_credentials',
- side_effect=Exception)
- def test_prepare_credentials_exception(self, *args):
- with self.assertRaises(vnf.VnfPreparationException):
- self.router_vnf.prepare()
- args[0].assert_called_once_with()
-
if __name__ == "__main__":
logging.disable(logging.CRITICAL)
diff --git a/functest/utils/openstack_tacker.py b/functest/utils/openstack_tacker.py
deleted file mode 100644
index 9fd9d5c47..000000000
--- a/functest/utils/openstack_tacker.py
+++ /dev/null
@@ -1,317 +0,0 @@
-###########################################################################
-# Copyright (c) 2016 Ericsson AB and others.
-# Author: George Paraskevopoulos <geopar@intracom-telecom.com>
-#
-# Wrappers for trozet's python-tackerclient v1.0
-# (https://github.com/trozet/python-tackerclient)
-#
-# 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
-
-from tackerclient.v1_0 import client as tackerclient
-import functest.utils.openstack_utils as os_utils
-import time
-
-logger = logging.getLogger(__name__)
-
-
-def get_tacker_client(other_creds={}):
- sess = os_utils.get_session(other_creds)
- return tackerclient.Client(session=sess)
-
-
-# *********************************************
-# TACKER
-# *********************************************
-def get_id_from_name(tacker_client, resource_type, resource_name):
- try:
- req_params = {'fields': 'id', 'name': resource_name}
- endpoint = '/{0}s'.format(resource_type)
- resp = tacker_client.get(endpoint, params=req_params)
- endpoint = endpoint.replace('-', '_')
- return resp[endpoint[1:]][0]['id']
- except Exception, e:
- logger.error("Error [get_id_from_name(tacker_client, "
- "resource_type, resource_name)]: %s" % e)
- return None
-
-
-def get_vnfd_id(tacker_client, vnfd_name):
- return get_id_from_name(tacker_client, 'vnfd', vnfd_name)
-
-
-def get_vnf_id(tacker_client, vnf_name, timeout=5):
- vnf_id = None
- while vnf_id is None and timeout >= 0:
- vnf_id = get_id_from_name(tacker_client, 'vnf', vnf_name)
- if vnf_id is None:
- logger.info("Could not retrieve ID for vnf with name [%s]."
- " Retrying." % vnf_name)
- time.sleep(1)
- timeout -= 1
- return vnf_id
-
-
-def get_sfc_id(tacker_client, sfc_name):
- return get_id_from_name(tacker_client, 'sfc', sfc_name)
-
-
-def get_sfc_classifier_id(tacker_client, sfc_clf_name):
- return get_id_from_name(tacker_client, 'sfc-classifier', sfc_clf_name)
-
-
-def list_vnfds(tacker_client, verbose=False):
- try:
- vnfds = tacker_client.list_vnfds(retrieve_all=True)
- if not verbose:
- vnfds = [vnfd['id'] for vnfd in vnfds['vnfds']]
- return vnfds
- except Exception, e:
- logger.error("Error [list_vnfds(tacker_client)]: %s" % e)
- return None
-
-
-def create_vnfd(tacker_client, tosca_file=None):
- try:
- vnfd_body = {}
- if tosca_file is not None:
- with open(tosca_file) as tosca_fd:
- vnfd_body = tosca_fd.read()
- logger.info('VNFD template:\n{0}'.format(vnfd_body))
- return tacker_client.create_vnfd(
- body={"vnfd": {"attributes": {"vnfd": vnfd_body}}})
- except Exception, e:
- logger.error("Error [create_vnfd(tacker_client, '%s')]: %s"
- % (tosca_file, e))
- return None
-
-
-def delete_vnfd(tacker_client, vnfd_id=None, vnfd_name=None):
- try:
- vnfd = vnfd_id
- if vnfd is None:
- if vnfd_name is None:
- raise Exception('You need to provide VNFD id or VNFD name')
- vnfd = get_vnfd_id(tacker_client, vnfd_name)
- return tacker_client.delete_vnfd(vnfd)
- except Exception, e:
- logger.error("Error [delete_vnfd(tacker_client, '%s', '%s')]: %s"
- % (vnfd_id, vnfd_name, e))
- return None
-
-
-def list_vnfs(tacker_client, verbose=False):
- try:
- vnfs = tacker_client.list_vnfs(retrieve_all=True)
- if not verbose:
- vnfs = [vnf['id'] for vnf in vnfs['vnfs']]
- return vnfs
- except Exception, e:
- logger.error("Error [list_vnfs(tacker_client)]: %s" % e)
- return None
-
-
-def create_vnf(tacker_client, vnf_name, vnfd_id=None,
- vnfd_name=None, param_file=None):
- try:
- vnf_body = {
- 'vnf': {
- 'attributes': {},
- 'name': vnf_name
- }
- }
- if param_file is not None:
- params = None
- with open(param_file) as f:
- params = f.read()
- vnf_body['vnf']['attributes']['param_values'] = params
- if vnfd_id is not None:
- vnf_body['vnf']['vnfd_id'] = vnfd_id
- else:
- if vnfd_name is None:
- raise Exception('vnfd id or vnfd name is required')
- vnf_body['vnf']['vnfd_id'] = get_vnfd_id(tacker_client, vnfd_name)
- return tacker_client.create_vnf(body=vnf_body)
- except Exception, e:
- logger.error("error [create_vnf(tacker_client,"
- " '%s', '%s', '%s')]: %s"
- % (vnf_name, vnfd_id, vnfd_name, e))
- return None
-
-
-def get_vnf(tacker_client, vnf_id=None, vnf_name=None):
- try:
- if vnf_id is None and vnf_name is None:
- raise Exception('You must specify vnf_id or vnf_name')
-
- _id = get_vnf_id(tacker_client, vnf_name) if vnf_id is None else vnf_id
-
- if _id is not None:
- all_vnfs = list_vnfs(tacker_client, verbose=True)['vnfs']
- return next((vnf for vnf in all_vnfs if vnf['id'] == _id), None)
- else:
- raise Exception('Could not retrieve ID from name [%s]' % vnf_name)
-
- except Exception, e:
- logger.error("Could not retrieve VNF [vnf_id=%s, vnf_name=%s] - %s"
- % (vnf_id, vnf_name, e))
- return None
-
-
-def wait_for_vnf(tacker_client, vnf_id=None, vnf_name=None, timeout=60):
- try:
- vnf = get_vnf(tacker_client, vnf_id, vnf_name)
- if vnf is None:
- raise Exception("Could not retrieve VNF - id='%s', name='%s'"
- % vnf_id, vnf_name)
- logger.info('Waiting for vnf {0}'.format(str(vnf)))
- while vnf['status'] != 'ACTIVE' and timeout >= 0:
- if vnf['status'] == 'ERROR':
- raise Exception('Error when booting vnf %s' % vnf['id'])
- elif vnf['status'] == 'PENDING_CREATE':
- time.sleep(3)
- timeout -= 3
- vnf = get_vnf(tacker_client, vnf_id, vnf_name)
-
- if (timeout < 0):
- raise Exception('Timeout when booting vnf %s' % vnf['id'])
-
- return vnf['id']
- except Exception, e:
- logger.error("error [wait_for_vnf(tacker_client, '%s', '%s')]: %s"
- % (vnf_id, vnf_name, e))
- return None
-
-
-def delete_vnf(tacker_client, vnf_id=None, vnf_name=None):
- try:
- vnf = vnf_id
- if vnf is None:
- if vnf_name is None:
- raise Exception('You need to provide a VNF id or name')
- vnf = get_vnf_id(tacker_client, vnf_name)
- return tacker_client.delete_vnf(vnf)
- except Exception, e:
- logger.error("Error [delete_vnf(tacker_client, '%s', '%s')]: %s"
- % (vnf_id, vnf_name, e))
- return None
-
-
-def list_sfcs(tacker_client, verbose=False):
- try:
- sfcs = tacker_client.list_sfcs(retrieve_all=True)
- if not verbose:
- sfcs = [sfc['id'] for sfc in sfcs['sfcs']]
- return sfcs
- except Exception, e:
- logger.error("Error [list_sfcs(tacker_client)]: %s" % e)
- return None
-
-
-def create_sfc(tacker_client, sfc_name,
- chain_vnf_ids=None,
- chain_vnf_names=None,
- symmetrical=False):
- try:
- sfc_body = {
- 'sfc': {
- 'attributes': {},
- 'name': sfc_name,
- 'chain': []
- }
- }
- if symmetrical:
- sfc_body['sfc']['symmetrical'] = True
- if chain_vnf_ids is not None:
- sfc_body['sfc']['chain'] = chain_vnf_ids
- else:
- if chain_vnf_names is None:
- raise Exception('You need to provide a chain of VNFs')
- sfc_body['sfc']['chain'] = [get_vnf_id(tacker_client, name)
- for name in chain_vnf_names]
- return tacker_client.create_sfc(body=sfc_body)
- except Exception, e:
- logger.error("error [create_sfc(tacker_client,"
- " '%s', '%s', '%s')]: %s"
- % (sfc_name, chain_vnf_ids, chain_vnf_names, e))
- return None
-
-
-def delete_sfc(tacker_client, sfc_id=None, sfc_name=None):
- try:
- sfc = sfc_id
- if sfc is None:
- if sfc_name is None:
- raise Exception('You need to provide an SFC id or name')
- sfc = get_sfc_id(tacker_client, sfc_name)
- return tacker_client.delete_sfc(sfc)
- except Exception, e:
- logger.error("Error [delete_sfc(tacker_client, '%s', '%s')]: %s"
- % (sfc_id, sfc_name, e))
- return None
-
-
-def list_sfc_classifiers(tacker_client, verbose=False):
- try:
- sfc_clfs = tacker_client.list_sfc_classifiers(retrieve_all=True)
- if not verbose:
- sfc_clfs = [sfc_clf['id']
- for sfc_clf in sfc_clfs['sfc_classifiers']]
- return sfc_clfs
- except Exception, e:
- logger.error("Error [list_sfc_classifiers(tacker_client)]: %s" % e)
- return None
-
-
-def create_sfc_classifier(tacker_client, sfc_clf_name, sfc_id=None,
- sfc_name=None, match={}):
- # Example match:
- # match: {
- # "source_port": "0",
- # "protocol": "6",
- # "dest_port": "80"
- # }
- try:
- sfc_clf_body = {
- 'sfc_classifier': {
- 'attributes': {},
- 'name': sfc_clf_name,
- 'match': match,
- 'chain': ''
- }
- }
- if sfc_id is not None:
- sfc_clf_body['sfc_classifier']['chain'] = sfc_id
- else:
- if sfc_name is None:
- raise Exception('You need to provide an SFC id or name')
- sfc_clf_body['sfc_classifier']['chain'] = get_sfc_id(
- tacker_client, sfc_name)
- return tacker_client.create_sfc_classifier(body=sfc_clf_body)
- except Exception, e:
- logger.error("error [create_sfc_classifier(tacker_client,"
- " '%s', '%s','%s', '%s')]: '%s'"
- % (sfc_clf_name, sfc_id, sfc_name, str(match), e))
- return None
-
-
-def delete_sfc_classifier(tacker_client,
- sfc_clf_id=None,
- sfc_clf_name=None):
- try:
- sfc_clf = sfc_clf_id
- if sfc_clf is None:
- if sfc_clf_name is None:
- raise Exception('You need to provide an SFC'
- 'classifier id or name')
- sfc_clf = get_sfc_classifier_id(tacker_client, sfc_clf_name)
- return tacker_client.delete_sfc_classifier(sfc_clf)
- except Exception, e:
- logger.error("Error [delete_sfc_classifier(tacker_client, '%s', "
- "'%s')]: %s" % (sfc_clf_id, sfc_clf_name, e))
- return None
diff --git a/requirements.txt b/requirements.txt
index ba5d91cb8..da1b5ad56 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -8,8 +8,6 @@ python-heatclient>=1.6.1 # Apache-2.0
python-keystoneclient>=3.8.0 # Apache-2.0
python-neutronclient>=5.1.0 # Apache-2.0
python-novaclient!=7.0.0,>=6.0.0 # Apache-2.0
-python-tackerclient>=0.8.0 # Apache-2.0
-pexpect!=3.3,>=3.1 # ISC License
requests!=2.12.2,>=2.10.0 # Apache-2.0
robotframework>=3.0
robotframework-httplibrary
@@ -20,7 +18,7 @@ dnspython>=1.14.0;python_version=='2.7' # http://www.dnspython.org/LICENSE
dnspython3!=1.13.0,!=1.14.0,>=1.12.0;python_version>='3.0' # http://www.dnspython.org/LICENSE
click
openbaton-cli
-cloudify_rest_client
+cloudify-rest-client
Flask!=0.11,<1.0,>=0.10 # BSD
Flask-RESTful>=0.3.5 # BSD
IPy
@@ -30,5 +28,11 @@ PrettyTable<0.8,>=0.7.1 # BSD
six>=1.9.0 # MIT
opnfv
snaps
-rally
tempest>=12.1.0 # Apache-2.0
+rally
+SQLAlchemy<1.1.0,>=1.0.10 # MIT
+paramiko>=2.0 # LGPLv2.1+
+Jinja2!=2.9.0,!=2.9.1,!=2.9.2,!=2.9.3,!=2.9.4,>=2.8 # BSD License (3 clause)
+urllib3>=1.15.1 # MIT
+oslo.serialization>=1.10.0 # Apache-2.0
+oslo.utils>=3.18.0 # Apache-2.0
diff --git a/upper-constraints.txt b/upper-constraints.txt
index fae94270c..e80ffffb2 100644
--- a/upper-constraints.txt
+++ b/upper-constraints.txt
@@ -6,12 +6,11 @@ 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
-e git+https://github.com/openstack/refstack-client#egg=refstack-client
-cloudify_rest_client===4.0
+cloudify-rest-client===4.0
iniparse===0.4
openbaton-cli===2.2.1b7
robotframework===3.0.2
robotframework-httplibrary===0.4.2
robotframework-requests===0.4.7
robotframework-sshlibrary===2.1.3;python_version=='2.7'
-rally===0.9.1
IPy===0.83