mkdir -p /var/lib/rally/database && rally db create && \
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"]
+CMD ["run_tests", "-t", "all"]
rm -r upper-constraints.txt thirdparty-requirements.txt /src/fds/.git && \
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"]
+CMD ["run_tests", "-t", "all"]
FROM opnfv/functest-core
COPY testcases.yaml /usr/lib/python2.7/site-packages/functest/ci/testcases.yaml
-CMD ["bash","-c","prepare_env start && run_tests -t all"]
+CMD ["run_tests", "-t", "all"]
rm upper-constraints.txt 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"]
+CMD ["run_tests", "-t", "all"]
mkdir -p /var/lib/rally/database && rally db create && \
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"]
+CMD ["run_tests", "-t", "all"]
cd /src/vims-test && bundle config build.nokogiri --use-system-libraries && bundle install --system && \
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"]
+CMD ["run_tests", "-t", "all"]
│ |-- config_patch.yaml
│ |-- download_images.sh
│ |-- logging.ini
- │ |-- prepare_env.py
│ |-- rally_aarch64_patch.conf
│ |-- run_tests.py
│ |-- testcases.yaml
$ source /home/opnfv/functest/conf/openstack.creds
-Prepare environment
--------------------
-Prior to commencing the Functest environment preparation, we can check
-the initial status of the environment. Issue the **functest env status**
-command at the prompt::
-
- # functest env status
- # Functest environment is not installed.
-
-To prepare the Functest docker container for test case execution, type::
-
- # functest env prepare
- # ...
- # Functest environment ready to run tests.
-
-you may also type prepare_env instead of functest env prepare.
-
-To list some basic information about an already prepared Functest
-docker container environment, issue the **functest env show** at the
-prompt::
- bash-4.3# functest env show
- +------------------------------+---------------------------------+
- | FUNCTEST ENVIRONMENT | VALUE |
- +------------------------------+---------------------------------+
- | STATUS | ready |
- | SCENARIO | os-nosdn-nofeature-noha |
- | DEBUG FLAG | false |
- | BUILD_TAG | None |
- | INSTALLER | compass |
- | POD | huawei-pod1 |
- +------------------------------+---------------------------------+
-
-See configuration guide for details on Functest environnement variables.
-
Tier
----
Each Alpine container provided on the docker hub matches with a tier.
-v $(pwd)/images:/home/opnfv/functest/images \
-v $(pwd)/my-custom-testcases.yaml:/usr/lib/python2.7/site-packages/functest/ci/testcases.yaml \
-v $(pwd)/my-custom-tempest-tests.txt:/usr/lib/python2.7/site-packages/functest/opnfv_tests/openstack/tempest/custom_tests/test_list.txt \
- opnfv/functest-components /bin/bash -c "prepare_env start && run_tests -t tempest_custom"
+ opnfv/functest-components run_tests -t tempest_custom
Rally
from functest.api.base import ApiResource
from functest.api.common import api_utils
from functest.cli.commands.cli_env import Env
-import functest.utils.functest_utils as ft_utils
ADDRESS = socket.gethostbyname(socket.gethostname())
ENDPOINT_ENVS = ('http://{}:5000/api/v1/functest/envs'.format(ADDRESS))
""" Used to handle post request """
return self._dispatch_post()
- def prepare(self, args): # pylint: disable=no-self-use, unused-argument
- """ Prepare environment """
-
- 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")
-
def update_hosts(self, hosts_info): # pylint: disable=no-self-use
""" Update hosts info """
case_name = args.get('testcase')
self._update_logging_ini(args.get('task_id'))
- if not os.path.isfile(CONST.__getattribute__('env_active')):
- raise Exception("Functest environment is not ready.")
+ 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:
- 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'),
- 'testcase': case_name,
- 'env_info': env_info,
- 'result': result
- }
-
- return {'result': result}
+ 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'),
+ 'testcase': case_name,
+ 'env_info': env_info,
+ 'result': result
+ }
+
+ return {'result': result}
def _update_logging_ini(self, task_id): # pylint: disable=no-self-use
""" Update the log file for each task"""
-Prepare environment or Update hosts info
+Update hosts info
-This api offers the interface to prepare environment or update hosts info.
+This api offers the interface to update hosts info.
-action: prepare
action: update_hosts
---
tags:
schema:
required:
- action
+ - args
properties:
action:
type: string
description: this is action for envs
- default: prepare
+ default: update_hosts
+ args:
+ type: string
+ description: Hosts info to be updated
+ default:
+ "identity.ac.dz.com": "8.20.11.22"
+ "image.ac.dz.com": "8.20.11.22"
definitions:
Environment:
type: object
type: dict
responses:
200:
- description: Prepare environment
+ description: Update hosts info
schema:
$ref: '#/definitions/Environment'
examples:
'status': 0
- 'result': 'Prapare env successfully'
+ 'result': 'Update hosts info successfully'
\ No newline at end of file
Url('/api/v1/functest/envs', 'v1_envs'),
# POST /api/v1/functest/envs/action
- # {"action":"prepare"} => Prepare environment
# {"action":"update_hosts", "args": {}} => Update hosts info
Url('/api/v1/functest/envs/action', 'v1_envs'),
def check_all(self):
"""
Calls all the class functions and returns 0 if all of them succeed.
- This is the method called by prepare_env or CLI
+ This is the method called by CLI
"""
self.check_rc()
try:
+++ /dev/null
-#!/usr/bin/env python
-#
-# All rights reserved. This program and the accompanying materials
-# are made available under the terms of the Apache License, Version 2.0
-# which accompanies this distribution, and is available at
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-
-import argparse
-import logging
-import logging.config
-import os
-import pkg_resources
-import re
-import sys
-
-import yaml
-
-from functest.ci import check_deployment
-import functest.utils.functest_utils as ft_utils
-import functest.utils.openstack_utils as os_utils
-from functest.utils.constants import CONST
-
-actions = ['start', 'check']
-
-""" logging configuration """
-logger = logging.getLogger('functest.ci.prepare_env')
-handler = None
-# set the architecture to default
-pod_arch = os.getenv("POD_ARCH", None)
-arch_filter = ['aarch64']
-
-CONFIG_FUNCTEST_PATH = pkg_resources.resource_filename(
- 'functest', 'ci/config_functest.yaml')
-CONFIG_PATCH_PATH = pkg_resources.resource_filename(
- 'functest', 'ci/config_patch.yaml')
-CONFIG_AARCH64_PATCH_PATH = pkg_resources.resource_filename(
- 'functest', 'ci/config_aarch64_patch.yaml')
-
-
-class PrepareEnvParser(object):
-
- def __init__(self):
- self.parser = argparse.ArgumentParser()
- self.parser.add_argument("action", help="Possible actions are: "
- "'{d[0]}|{d[1]}' ".format(d=actions),
- choices=actions)
- self.parser.add_argument("-d", "--debug", help="Debug mode",
- action="store_true")
-
- def parse_args(self, argv=[]):
- return vars(self.parser.parse_args(argv))
-
-
-def print_separator():
- logger.info("==============================================")
-
-
-def source_rc_file():
- print_separator()
-
- logger.info("Sourcing the OpenStack RC file...")
- os_utils.source_credentials(CONST.__getattribute__('openstack_creds'))
- for key, value in os.environ.iteritems():
- if re.search("OS_", key):
- if key == 'OS_AUTH_URL':
- CONST.__setattr__('OS_AUTH_URL', value)
- elif key == 'OS_USERNAME':
- CONST.__setattr__('OS_USERNAME', value)
- elif key == 'OS_TENANT_NAME':
- CONST.__setattr__('OS_TENANT_NAME', value)
- elif key == 'OS_PASSWORD':
- CONST.__setattr__('OS_PASSWORD', value)
-
-
-def update_config_file():
- patch_file(CONFIG_PATCH_PATH)
-
- if pod_arch and pod_arch in arch_filter:
- patch_file(CONFIG_AARCH64_PATCH_PATH)
-
- if "TEST_DB_URL" in os.environ:
- update_db_url()
-
-
-def patch_file(patch_file_path):
- logger.debug('Updating file: %s', patch_file_path)
- with open(patch_file_path) as f:
- patch_file = yaml.safe_load(f)
-
- updated = False
- for key in patch_file:
- if key in CONST.__getattribute__('DEPLOY_SCENARIO'):
- new_functest_yaml = dict(ft_utils.merge_dicts(
- ft_utils.get_functest_yaml(), patch_file[key]))
- updated = True
-
- if updated:
- os.remove(CONFIG_FUNCTEST_PATH)
- with open(CONFIG_FUNCTEST_PATH, "w") as f:
- f.write(yaml.dump(new_functest_yaml, default_style='"'))
-
-
-def update_db_url():
- with open(CONFIG_FUNCTEST_PATH) as f:
- functest_yaml = yaml.safe_load(f)
-
- with open(CONFIG_FUNCTEST_PATH, "w") as f:
- functest_yaml["results"]["test_db_url"] = os.environ.get('TEST_DB_URL')
- f.write(yaml.dump(functest_yaml, default_style='"'))
-
-
-def verify_deployment():
- print_separator()
- logger.info("Verifying OpenStack deployment...")
- deployment = check_deployment.CheckDeployment()
- deployment.check_all()
-
-
-def create_flavor():
- _, flavor_id = os_utils.get_or_create_flavor('m1.tiny',
- '512',
- '1',
- '1',
- public=True)
- if flavor_id is None:
- raise Exception('Failed to create flavor')
-
-
-def check_environment():
- msg_not_active = "The Functest environment is not installed."
- if not os.path.isfile(CONST.__getattribute__('env_active')):
- raise Exception(msg_not_active)
-
- with open(CONST.__getattribute__('env_active'), "r") as env_file:
- s = env_file.read()
- if not re.search("1", s):
- raise Exception(msg_not_active)
-
- logger.info("Functest environment is installed.")
-
-
-def prepare_env(**kwargs):
- try:
- if not (kwargs['action'] in actions):
- logger.error('Argument not valid.')
- return -1
- elif kwargs['action'] == "start":
- logger.info("######### Preparing Functest environment #########\n")
- verify_deployment()
- source_rc_file()
- update_config_file()
- create_flavor()
- with open(CONST.__getattribute__('env_active'), "w") as env_file:
- env_file.write("1")
- check_environment()
- elif kwargs['action'] == "check":
- check_environment()
- except Exception as e:
- logger.error(e)
- return -1
- return 0
-
-
-def main():
- logging.config.fileConfig(pkg_resources.resource_filename(
- 'functest', 'ci/logging.ini'))
- logging.captureWarnings(True)
- parser = PrepareEnvParser()
- args = parser.parse_args(sys.argv[1:])
- return prepare_env(**args)
import textwrap
import prettytable
+import yaml
import functest.ci.tier_builder as tb
import functest.core.testcase as testcase
# __name__ cannot be used here
logger = logging.getLogger('functest.ci.run_tests')
+CONFIG_FUNCTEST_PATH = pkg_resources.resource_filename(
+ 'functest', 'ci/config_functest.yaml')
+CONFIG_PATCH_PATH = pkg_resources.resource_filename(
+ 'functest', 'ci/config_patch.yaml')
+CONFIG_AARCH64_PATCH_PATH = pkg_resources.resource_filename(
+ 'functest', 'ci/config_aarch64_patch.yaml')
+# set the architecture to default
+pod_arch = os.getenv("POD_ARCH", None)
+arch_filter = ['aarch64']
+
class Result(enum.Enum):
EX_OK = os.EX_OK
CONST.__getattribute__('DEPLOY_SCENARIO'),
pkg_resources.resource_filename('functest', 'ci/testcases.yaml'))
+ @staticmethod
+ def update_config_file():
+ Runner.patch_file(CONFIG_PATCH_PATH)
+
+ if pod_arch and pod_arch in arch_filter:
+ Runner.patch_file(CONFIG_AARCH64_PATCH_PATH)
+
+ if "TEST_DB_URL" in os.environ:
+ Runner.update_db_url()
+
+ @staticmethod
+ def patch_file(patch_file_path):
+ logger.debug('Updating file: %s', patch_file_path)
+ with open(patch_file_path) as f:
+ patch_file = yaml.safe_load(f)
+
+ updated = False
+ for key in patch_file:
+ if key in CONST.__getattribute__('DEPLOY_SCENARIO'):
+ new_functest_yaml = dict(ft_utils.merge_dicts(
+ ft_utils.get_functest_yaml(), patch_file[key]))
+ updated = True
+
+ if updated:
+ os.remove(CONFIG_FUNCTEST_PATH)
+ with open(CONFIG_FUNCTEST_PATH, "w") as f:
+ f.write(yaml.dump(new_functest_yaml, default_style='"'))
+
+ @staticmethod
+ def update_db_url():
+ with open(CONFIG_FUNCTEST_PATH) as f:
+ functest_yaml = yaml.safe_load(f)
+
+ with open(CONFIG_FUNCTEST_PATH, "w") as f:
+ functest_yaml["results"]["test_db_url"] = os.environ.get(
+ 'TEST_DB_URL')
+ f.write(yaml.dump(functest_yaml, default_style='"'))
+
@staticmethod
def source_rc_file():
rc_file = CONST.__getattribute__('openstack_creds')
self.run_tier(tier)
def main(self, **kwargs):
+ Runner.update_config_file()
+
if 'noclean' in kwargs:
self.clean_flag = not kwargs['noclean']
if 'report' in kwargs:
_openstack.show_credentials()
-@env.command('prepare', help="Prepares the Functest environment. This step is "
- "needed run the tests.")
-def env_prepare():
- _env.prepare()
-
-
@env.command('show', help="Shows information about the current environment.")
def env_show():
_env.show()
-@env.command('status', help="Checks if the Functest environment is ready to "
- "run the tests.")
-def env_status():
- _env.status()
-
-
@testcase.command('list', help="Lists the available testcases.")
def testcase_list():
_testcase.list()
# http://www.apache.org/licenses/LICENSE-2.0
#
-import os
-
import click
import prettytable
from functest.utils.constants import CONST
-import functest.utils.functest_utils as ft_utils
class Env(object):
def __init__(self):
pass
- def prepare(self):
- if self.status(verbose=False) == 0:
- answer = raw_input("It seems that the environment has been "
- "already prepared. Do you want to do "
- "it again? [y|n]\n")
- while True:
- if answer.lower() in ["y", "yes"]:
- os.remove(CONST.__getattribute__('env_active'))
- break
- elif answer.lower() in ["n", "no"]:
- return
- else:
- answer = raw_input("Invalid answer. Please type [y|n]\n")
-
- ft_utils.execute_command("prepare_env start")
-
def show(self):
def _get_value(attr, default='Unknown'):
return attr if attr else default
build_tag = build_tag.lstrip(
"jenkins-").lstrip("functest").lstrip("-")
- STATUS = "not ready"
- if self.status(verbose=False) == 0:
- STATUS = "ready"
-
env_info = {'INSTALLER': installer_info,
'SCENARIO': scenario,
'POD': node,
'DEBUG FLAG': is_debug,
- 'BUILD_TAG': build_tag,
- 'STATUS': STATUS}
+ 'BUILD_TAG': build_tag}
return env_info
- def status(self, verbose=True):
- ret_val = 0
- if not os.path.isfile(CONST.__getattribute__('env_active')):
- if verbose:
- click.echo("Functest environment is not installed.\n")
- ret_val = 1
- elif verbose:
- click.echo("Functest environment ready to run tests.\n")
-
- return ret_val
-
class CliEnv(Env):
""" global variables """
-import os
import pkg_resources
import click
if testname == 'vacation':
vacation.main()
- elif not os.path.isfile(CONST.__getattribute__('env_active')):
- click.echo("Functest environment is not ready. "
- "Run first 'functest env prepare'")
else:
tests = testname.split(",")
for test in tests:
""" global variables """
-import os
import pkg_resources
import click
@staticmethod
def run(tiername, noclean=False, report=False):
-
flags = ""
if noclean:
flags += "-n "
if report:
flags += "-r "
- if not os.path.isfile(CONST.__getattribute__('env_active')):
- click.echo("Functest environment is not ready. "
- "Run first 'functest env prepare'")
- else:
- cmd = "run_tests {}-t {}".format(flags, tiername)
- ft_utils.execute_command(cmd)
+ cmd = "run_tests {}-t {}".format(flags, tiername)
+ ft_utils.execute_command(cmd)
class CliTier(Tier):
import functest.core.testcase as base
from functest.utils.constants import CONST
-from snaps.openstack.create_user import UserSettings, OpenStackUser
-from snaps.openstack.create_project import ProjectSettings, OpenStackProject
+from snaps.config.user import UserConfig
+from snaps.config.project import ProjectConfig
+from snaps.openstack.create_user import OpenStackUser
+from snaps.openstack.create_project import OpenStackProject
from snaps.openstack.tests import openstack_tests
__author__ = ("Morgan Richomme <morgan.richomme@orange.com>, "
project_creator = OpenStackProject(
snaps_creds,
- ProjectSettings(
+ ProjectConfig(
name=self.tenant_name,
description=tenant_description
))
user_creator = OpenStackUser(
snaps_creds,
- UserSettings(
+ UserConfig(
name=self.tenant_name,
password=self.tenant_name,
roles={'admin': self.tenant_name}))
from functest.opnfv_tests.openstack.tempest import conf_utils
from functest.utils.constants import CONST
-from snaps.openstack.create_flavor import FlavorSettings, OpenStackFlavor
-from snaps.openstack.create_image import ImageSettings
-from snaps.openstack.create_network import NetworkSettings, SubnetSettings
-from snaps.openstack.create_router import RouterSettings
+from snaps.config.flavor import FlavorConfig
+from snaps.config.image import ImageConfig
+from snaps.config.network import NetworkConfig, SubnetConfig
+from snaps.config.router import RouterConfig
+
+from snaps.openstack.create_flavor import OpenStackFlavor
from snaps.openstack.tests import openstack_tests
from snaps.openstack.utils import deploy_utils
LOGGER.info('No tests for scenario "%s"', test_name)
return
- cmd_line = ("rally task start --abort-on-sla-failure "
- "--task {0} "
- "--task-args \"{1}\""
- .format(task_file, self._build_task_args(test_name)))
- LOGGER.debug('running command line: %s', cmd_line)
+ cmd = (["rally", "task", "start", "--abort-on-sla-failure", "--task",
+ task_file, "--task-args",
+ str(self._build_task_args(test_name))])
+ LOGGER.debug('running command: %s', cmd)
- proc = subprocess.Popen(cmd_line, stdout=subprocess.PIPE,
- stderr=subprocess.STDOUT, shell=True)
+ proc = subprocess.Popen(cmd, stdout=subprocess.PIPE,
+ stderr=subprocess.STDOUT)
output = self._get_output(proc, test_name)
task_id = self.get_task_id(output)
LOGGER.debug('task_id : %s', task_id)
if task_id is None:
LOGGER.error('Failed to retrieve task_id, validating task...')
- cmd_line = ("rally task validate "
- "--task {0} "
- "--task-args \"{1}\""
- .format(task_file, self._build_task_args(test_name)))
- LOGGER.debug('running command line: %s', cmd_line)
- proc = subprocess.Popen(cmd_line, stdout=subprocess.PIPE,
- stderr=subprocess.STDOUT, shell=True)
+ cmd = (["rally", "task", "validate", "--task", task_file,
+ "--task-args", str(self._build_task_args(test_name))])
+ LOGGER.debug('running command: %s', cmd)
+ proc = subprocess.Popen(cmd, stdout=subprocess.PIPE,
+ stderr=subprocess.STDOUT)
output = self.get_cmd_output(proc)
LOGGER.error("Task validation result:" + "\n" + output)
return
# write html report file
report_html_name = 'opnfv-{}.html'.format(test_name)
report_html_dir = os.path.join(self.RESULTS_DIR, report_html_name)
- cmd_line = "rally task report {} --out {}".format(task_id,
- report_html_dir)
+ cmd = (["rally", "task", "report", task_id, "--out", report_html_dir])
- LOGGER.debug('running command line: %s', cmd_line)
- os.popen(cmd_line)
+ LOGGER.debug('running command: %s', cmd)
+ subprocess.Popen(cmd, stdout=subprocess.PIPE,
+ stderr=subprocess.STDOUT)
# get and save rally operation JSON result
- cmd_line = "rally task results %s" % task_id
- LOGGER.debug('running command line: %s', cmd_line)
- cmd = os.popen(cmd_line)
- json_results = cmd.read()
+ cmd = (["rally", "task", "results", task_id])
+ LOGGER.debug('running command: %s', cmd)
+ proc = subprocess.Popen(cmd, stdout=subprocess.PIPE,
+ stderr=subprocess.STDOUT)
+ json_results = self.get_cmd_output(proc)
report_json_name = 'opnfv-{}.json'.format(test_name)
report_json_dir = os.path.join(self.RESULTS_DIR, report_json_name)
with open(report_json_dir, 'w') as r_file:
LOGGER.debug("Creating image '%s'...", self.image_name)
image_creator = deploy_utils.create_image(
- self.os_creds, ImageSettings(
+ self.os_creds, ImageConfig(
name=self.image_name,
image_file=self.GLANCE_IMAGE_PATH,
img_format=self.GLANCE_IMAGE_FORMAT,
LOGGER.debug("Creating network '%s'...", network_name)
network_creator = deploy_utils.create_network(
- self.os_creds, NetworkSettings(
+ self.os_creds, NetworkConfig(
name=network_name,
shared=True,
- subnet_settings=[SubnetSettings(
+ subnet_settings=[SubnetConfig(
name=subnet_name,
cidr=self.RALLY_PRIVATE_SUBNET_CIDR)
]))
LOGGER.debug("Creating router '%s'...", router_name)
router_creator = deploy_utils.create_router(
- self.os_creds, RouterSettings(
+ self.os_creds, RouterConfig(
name=router_name,
external_gateway=self.ext_net_name,
internal_subnets=[subnet_name]))
LOGGER.debug("Creating flavor '%s'...", self.flavor_name)
flavor_creator = OpenStackFlavor(
- self.os_creds, FlavorSettings(
+ self.os_creds, FlavorConfig(
name=self.flavor_name, ram=512, disk=1, vcpus=1,
metadata=self.FLAVOR_EXTRA_SPECS))
if flavor_creator is None or flavor_creator.create() is None:
LOGGER.debug("Creating flavor '%s'...", self.flavor_alt_name)
flavor_alt_creator = OpenStackFlavor(
- self.os_creds, FlavorSettings(
+ self.os_creds, FlavorConfig(
name=self.flavor_alt_name, ram=1024, disk=1, vcpus=1,
metadata=self.FLAVOR_EXTRA_SPECS))
if flavor_alt_creator is None or flavor_alt_creator.create() is None:
from functest.utils.constants import CONST
import functest.utils.functest_utils as ft_utils
+from snaps.config.flavor import FlavorConfig
+from snaps.config.network import NetworkConfig, SubnetConfig
+from snaps.config.project import ProjectConfig
+from snaps.config.user import UserConfig
+
from snaps.openstack import create_flavor
-from snaps.openstack.create_flavor import FlavorSettings, OpenStackFlavor
-from snaps.openstack.create_project import ProjectSettings
-from snaps.openstack.create_network import NetworkSettings, SubnetSettings
-from snaps.openstack.create_user import UserSettings
+from snaps.openstack.create_flavor import OpenStackFlavor
from snaps.openstack.tests import openstack_tests
from snaps.openstack.utils import deploy_utils
project_name = CONST.__getattribute__(
'tempest_identity_tenant_name') + self.guid
project_creator = deploy_utils.create_project(
- self.os_creds, ProjectSettings(
+ self.os_creds, ProjectConfig(
name=project_name,
description=CONST.__getattribute__(
'tempest_identity_tenant_description')))
logger.debug("Creating user for Tempest suite")
user_creator = deploy_utils.create_user(
- self.os_creds, UserSettings(
+ self.os_creds, UserConfig(
name=CONST.__getattribute__(
'tempest_identity_user_name') + self.guid,
password=CONST.__getattribute__(
logger.debug("Creating private network for Tempest suite")
network_creator = deploy_utils.create_network(
- self.os_creds, NetworkSettings(
+ self.os_creds, NetworkConfig(
name=CONST.__getattribute__(
'tempest_private_net_name') + self.guid,
project_name=project_name,
- subnet_settings=[SubnetSettings(
+ subnet_settings=[SubnetConfig(
name=CONST.__getattribute__(
'tempest_private_subnet_name') + self.guid,
cidr=CONST.__getattribute__('tempest_private_subnet_cidr'))
if 'ovs' in scenario or 'fdio' in scenario:
flavor_metadata = create_flavor.MEM_PAGE_SIZE_LARGE
flavor_creator = OpenStackFlavor(
- self.os_creds, FlavorSettings(
+ self.os_creds, FlavorConfig(
name=CONST.__getattribute__(
'openstack_flavor_name') + self.guid,
ram=CONST.__getattribute__('openstack_flavor_ram'),
if 'ovs' in scenario or 'fdio' in scenario:
flavor_metadata_alt = create_flavor.MEM_PAGE_SIZE_LARGE
flavor_creator_alt = OpenStackFlavor(
- self.os_creds, FlavorSettings(
+ self.os_creds, FlavorConfig(
name=CONST.__getattribute__(
'openstack_flavor_name_alt') + self.guid,
ram=CONST.__getattribute__('openstack_flavor_ram'),
from functest.opnfv_tests.openstack.snaps import snaps_utils
from functest.utils.constants import CONST
+from snaps.config.flavor import FlavorConfig
+from snaps.config.network import NetworkConfig, SubnetConfig
+from snaps.config.router import RouterConfig
from snaps.openstack import create_flavor
-from snaps.openstack.create_flavor import FlavorSettings, OpenStackFlavor
-from snaps.openstack.create_network import NetworkSettings, SubnetSettings
-from snaps.openstack.create_router import RouterSettings
+from snaps.openstack.create_flavor import OpenStackFlavor
from snaps.openstack.tests import openstack_tests
from snaps.openstack.utils import deploy_utils
"Creating network with name: '%s'" % private_net_name)
self.network_creator = deploy_utils.create_network(
self.os_creds,
- NetworkSettings(
+ NetworkConfig(
name=private_net_name,
network_type=vping_network_type,
physical_network=vping_physical_network,
segmentation_id=vping_segmentation_id,
- subnet_settings=[SubnetSettings(
+ subnet_settings=[SubnetConfig(
name=private_subnet_name,
cidr=private_subnet_cidr)]))
self.creators.append(self.network_creator)
ext_net_name = snaps_utils.get_ext_net_name(self.os_creds)
self.router_creator = deploy_utils.create_router(
self.os_creds,
- RouterSettings(
+ RouterConfig(
name=self.router_name,
external_gateway=ext_net_name,
internal_subnets=[private_subnet_name]))
flavor_metadata = create_flavor.MEM_PAGE_SIZE_LARGE
flavor_creator = OpenStackFlavor(
self.os_creds,
- FlavorSettings(name=self.flavor_name, ram=512, disk=1, vcpus=1,
- metadata=flavor_metadata))
+ FlavorConfig(name=self.flavor_name, ram=512, disk=1, vcpus=1,
+ metadata=flavor_metadata))
flavor_creator.create()
self.creators.append(flavor_creator)
from functest.energy import energy
from functest.opnfv_tests.openstack.vping import vping_base
from functest.utils.constants import CONST
-from snaps.openstack.create_instance import FloatingIpSettings, \
- VmInstanceSettings
-from snaps.openstack.create_keypairs import KeypairSettings
-from snaps.openstack.create_network import PortSettings
-from snaps.openstack.create_security_group import Direction, Protocol, \
- SecurityGroupSettings, SecurityGroupRuleSettings
+from snaps.config.keypair import KeypairConfig
+from snaps.config.network import PortConfig
+from snaps.config.security_group import (
+ Direction, Protocol, SecurityGroupConfig, SecurityGroupRuleConfig)
+from snaps.config.vm_inst import FloatingIpConfig, VmInstanceConfig
+
from snaps.openstack.utils import deploy_utils
self.logger.info(log)
kp_creator = deploy_utils.create_keypair(
self.os_creds,
- KeypairSettings(name=self.kp_name,
- private_filepath=self.kp_priv_file,
- public_filepath=self.kp_pub_file))
+ KeypairConfig(
+ name=self.kp_name, private_filepath=self.kp_priv_file,
+ public_filepath=self.kp_pub_file))
self.creators.append(kp_creator)
# Creating Instance 1
- port1_settings = PortSettings(
+ port1_settings = PortConfig(
name=self.vm1_name + '-vPingPort',
network_name=self.network_creator.network_settings.name)
- instance1_settings = VmInstanceSettings(
+ instance1_settings = VmInstanceConfig(
name=self.vm1_name, flavor=self.flavor_name,
vm_boot_timeout=self.vm_boot_timeout,
vm_delete_timeout=self.vm_delete_timeout,
sg_creator = self.__create_security_group()
self.creators.append(sg_creator)
- port2_settings = PortSettings(
+ port2_settings = PortConfig(
name=self.vm2_name + '-vPingPort',
network_name=self.network_creator.network_settings.name)
- instance2_settings = VmInstanceSettings(
+ instance2_settings = VmInstanceConfig(
name=self.vm2_name, flavor=self.flavor_name,
vm_boot_timeout=self.vm_boot_timeout,
vm_delete_timeout=self.vm_delete_timeout,
ssh_connect_timeout=self.vm_ssh_connect_timeout,
port_settings=[port2_settings],
security_group_names=[sg_creator.sec_grp_settings.name],
- floating_ip_settings=[FloatingIpSettings(
+ floating_ip_settings=[FloatingIpConfig(
name=self.vm2_name + '-FIPName',
port_name=port2_settings.name,
router_name=self.router_creator.router_settings.name)])
"""
sg_rules = list()
sg_rules.append(
- SecurityGroupRuleSettings(sec_grp_name=self.sg_name,
- direction=Direction.ingress,
- protocol=Protocol.icmp))
+ SecurityGroupRuleConfig(
+ sec_grp_name=self.sg_name, direction=Direction.ingress,
+ protocol=Protocol.icmp))
sg_rules.append(
- SecurityGroupRuleSettings(sec_grp_name=self.sg_name,
- direction=Direction.ingress,
- protocol=Protocol.tcp, port_range_min=22,
- port_range_max=22))
+ SecurityGroupRuleConfig(
+ sec_grp_name=self.sg_name, direction=Direction.ingress,
+ protocol=Protocol.tcp, port_range_min=22, port_range_max=22))
sg_rules.append(
- SecurityGroupRuleSettings(sec_grp_name=self.sg_name,
- direction=Direction.egress,
- protocol=Protocol.tcp, port_range_min=22,
- port_range_max=22))
+ SecurityGroupRuleConfig(
+ sec_grp_name=self.sg_name, direction=Direction.egress,
+ protocol=Protocol.tcp, port_range_min=22, port_range_max=22))
log = "Security group with name: '%s'" % self.sg_name
self.logger.info(log)
return deploy_utils.create_security_group(self.os_creds,
- SecurityGroupSettings(
+ SecurityGroupConfig(
name=self.sg_name,
description=self.sg_desc,
rule_settings=sg_rules))
import time
+from snaps.config.network import PortConfig
+from snaps.config.vm_inst import VmInstanceConfig
from snaps.openstack.utils import deploy_utils
-from snaps.openstack.create_instance import VmInstanceSettings
-from snaps.openstack.create_network import PortSettings
from functest.core.testcase import TestCase
from functest.opnfv_tests.openstack.vping import vping_base
super(VPingUserdata, self).run()
# Creating Instance 1
- port1_settings = PortSettings(
+ port1_settings = PortConfig(
name=self.vm1_name + '-vPingPort',
network_name=self.network_creator.network_settings.name)
- instance1_settings = VmInstanceSettings(
+ instance1_settings = VmInstanceConfig(
name=self.vm1_name,
flavor=self.flavor_name,
vm_boot_timeout=self.vm_boot_timeout,
self.vm1_creator.get_port_ip(port1_settings.name))
if userdata:
# Creating Instance 2
- port2_settings = PortSettings(
+ port2_settings = PortConfig(
name=self.vm2_name + '-vPingPort',
network_name=self.network_creator.network_settings.name)
- instance2_settings = VmInstanceSettings(
+ instance2_settings = VmInstanceConfig(
name=self.vm2_name,
flavor=self.flavor_name,
vm_boot_timeout=self.vm_boot_timeout,
from functest.utils.constants import CONST
import functest.utils.openstack_utils as os_utils
-from snaps.openstack.create_network import (NetworkSettings, SubnetSettings,
- OpenStackNetwork)
-from snaps.openstack.create_security_group import (SecurityGroupSettings,
- SecurityGroupRuleSettings,
- Direction, Protocol,
- OpenStackSecurityGroup)
-from snaps.openstack.create_router import RouterSettings, OpenStackRouter
-from snaps.openstack.create_instance import (VmInstanceSettings,
- FloatingIpSettings,
- OpenStackVmInstance)
-from snaps.openstack.create_flavor import FlavorSettings, OpenStackFlavor
-from snaps.openstack.create_image import ImageSettings, OpenStackImage
-from snaps.openstack.create_keypairs import KeypairSettings, OpenStackKeypair
-from snaps.openstack.create_network import PortSettings
+from snaps.config.flavor import FlavorConfig
+from snaps.config.image import ImageConfig
+from snaps.config.keypair import KeypairConfig
+from snaps.config.network import NetworkConfig, PortConfig, SubnetConfig
+from snaps.config.router import RouterConfig
+from snaps.config.security_group import (
+ Direction, Protocol, SecurityGroupConfig, SecurityGroupRuleConfig)
+from snaps.config.vm_inst import FloatingIpConfig, VmInstanceConfig
+
+from snaps.openstack.create_flavor import OpenStackFlavor
+from snaps.openstack.create_image import OpenStackImage
+from snaps.openstack.create_instance import OpenStackVmInstance
+from snaps.openstack.create_keypairs import OpenStackKeypair
+from snaps.openstack.create_network import OpenStackNetwork
+from snaps.openstack.create_router import OpenStackRouter
+from snaps.openstack.create_security_group import OpenStackSecurityGroup
__author__ = "Valentin Boucher <valentin.boucher@orange.com>"
if image_file and image_name:
image_creator = OpenStackImage(
self.snaps_creds,
- ImageSettings(name=image_name,
- image_user='cloud',
- img_format='qcow2',
- image_file=image_file))
+ ImageConfig(
+ name=image_name, image_user='cloud',
+ img_format='qcow2', image_file=image_file))
image_creator.create()
# self.created_object.append(image_creator)
start_time = time.time()
self.__logger.info("Creating keypair ...")
kp_file = os.path.join(self.data_dir, "cloudify_ims.pem")
- keypair_settings = KeypairSettings(name='cloudify_ims_kp',
- private_filepath=kp_file)
+ keypair_settings = KeypairConfig(name='cloudify_ims_kp',
+ private_filepath=kp_file)
keypair_creator = OpenStackKeypair(self.snaps_creds, keypair_settings)
keypair_creator.create()
self.created_object.append(keypair_creator)
self.__logger.info("Creating full network ...")
- subnet_settings = SubnetSettings(name='cloudify_ims_subnet',
- cidr='10.67.79.0/24')
- network_settings = NetworkSettings(name='cloudify_ims_network',
- subnet_settings=[subnet_settings])
+ subnet_settings = SubnetConfig(name='cloudify_ims_subnet',
+ cidr='10.67.79.0/24')
+ network_settings = NetworkConfig(name='cloudify_ims_network',
+ subnet_settings=[subnet_settings])
network_creator = OpenStackNetwork(self.snaps_creds, network_settings)
network_creator.create()
self.created_object.append(network_creator)
ext_net_name = snaps_utils.get_ext_net_name(self.snaps_creds)
router_creator = OpenStackRouter(
self.snaps_creds,
- RouterSettings(
+ RouterConfig(
name='cloudify_ims_router',
external_gateway=ext_net_name,
internal_subnets=[subnet_settings.name]))
self.__logger.info("Creating security group for cloudify manager vm")
sg_rules = list()
sg_rules.append(
- SecurityGroupRuleSettings(sec_grp_name="sg-cloudify-manager",
- direction=Direction.ingress,
- protocol=Protocol.tcp, port_range_min=1,
- port_range_max=65535))
+ SecurityGroupRuleConfig(
+ sec_grp_name="sg-cloudify-manager",
+ direction=Direction.ingress, protocol=Protocol.tcp,
+ port_range_min=1, port_range_max=65535))
sg_rules.append(
- SecurityGroupRuleSettings(sec_grp_name="sg-cloudify-manager",
- direction=Direction.ingress,
- protocol=Protocol.udp, port_range_min=1,
- port_range_max=65535))
+ SecurityGroupRuleConfig(
+ sec_grp_name="sg-cloudify-manager",
+ direction=Direction.ingress, protocol=Protocol.udp,
+ port_range_min=1, port_range_max=65535))
securit_group_creator = OpenStackSecurityGroup(
self.snaps_creds,
- SecurityGroupSettings(
+ SecurityGroupConfig(
name="sg-cloudify-manager",
rule_settings=sg_rules))
# orchestrator VM flavor
self.__logger.info("Get or create flavor for cloudify manager vm ...")
- flavor_settings = FlavorSettings(
+ flavor_settings = FlavorConfig(
name=self.orchestrator['requirements']['flavor']['name'],
ram=self.orchestrator['requirements']['flavor']['ram_min'],
disk=50,
flavor_creator = OpenStackFlavor(self.snaps_creds, flavor_settings)
flavor_creator.create()
self.created_object.append(flavor_creator)
- image_settings = ImageSettings(
+ image_settings = ImageConfig(
name=self.orchestrator['requirements']['os_image'],
image_user='centos',
exists=True)
- port_settings = PortSettings(name='cloudify_manager_port',
- network_name=network_settings.name)
+ port_settings = PortConfig(name='cloudify_manager_port',
+ network_name=network_settings.name)
- manager_settings = VmInstanceSettings(
+ manager_settings = VmInstanceConfig(
name='cloudify_manager',
flavor=flavor_settings.name,
port_settings=[port_settings],
security_group_names=[securit_group_creator.sec_grp_settings.name],
- floating_ip_settings=[FloatingIpSettings(
+ floating_ip_settings=[FloatingIpConfig(
name='cloudify_manager_fip',
port_name=port_settings.name,
router_name=router_creator.router_settings.name)])
descriptor.get('file_name'))
self.__logger.info("Get or create flavor for all clearwater vm")
- flavor_settings = FlavorSettings(
+ flavor_settings = FlavorConfig(
name=self.vnf['requirements']['flavor']['name'],
ram=self.vnf['requirements']['flavor']['ram_min'],
disk=25,
import functest.opnfv_tests.vnf.ims.cloudify_ims as cloudify_ims
from functest.utils.constants import CONST
-from snaps.openstack.create_network import (NetworkSettings, SubnetSettings,
- OpenStackNetwork, PortSettings)
-from snaps.openstack.create_security_group import (SecurityGroupSettings,
- SecurityGroupRuleSettings,
- Direction, Protocol,
- OpenStackSecurityGroup)
-from snaps.openstack.create_router import RouterSettings, OpenStackRouter
-from snaps.openstack.create_instance import (VmInstanceSettings,
- FloatingIpSettings,
- OpenStackVmInstance)
-from snaps.openstack.create_flavor import FlavorSettings, OpenStackFlavor
-from snaps.openstack.create_image import ImageSettings
+from snaps.config.flavor import FlavorConfig
+from snaps.config.image import ImageConfig
+from snaps.config.network import NetworkConfig, PortConfig, SubnetConfig
+from snaps.config.router import RouterConfig
+from snaps.config.security_group import (
+ Direction, Protocol, SecurityGroupConfig, SecurityGroupRuleConfig)
+from snaps.config.vm_inst import FloatingIpConfig, VmInstanceConfig
+
+from snaps.openstack.create_flavor import OpenStackFlavor
+from snaps.openstack.create_instance import OpenStackVmInstance
+from snaps.openstack.create_network import OpenStackNetwork
+from snaps.openstack.create_router import OpenStackRouter
+from snaps.openstack.create_security_group import OpenStackSecurityGroup
from ixia.utils.IxChassisUtils import ChassisRestAPI
import ixia.utils.IxLoadUtils as IxLoadUtils
ellis_ip = outputs['ellis_ip']
self.__logger.info("Creating full IXIA network ...")
- subnet_settings = SubnetSettings(name='ixia_management_subnet',
- cidr='10.10.10.0/24')
- network_settings = NetworkSettings(name='ixia_management_network',
- subnet_settings=[subnet_settings])
+ subnet_settings = SubnetConfig(name='ixia_management_subnet',
+ cidr='10.10.10.0/24')
+ network_settings = NetworkConfig(name='ixia_management_network',
+ subnet_settings=[subnet_settings])
network_creator = OpenStackNetwork(self.snaps_creds, network_settings)
network_creator.create()
self.created_object.append(network_creator)
ext_net_name = snaps_utils.get_ext_net_name(self.snaps_creds)
router_creator = OpenStackRouter(
self.snaps_creds,
- RouterSettings(
+ RouterConfig(
name='ixia_management_router',
external_gateway=ext_net_name,
internal_subnets=[subnet_settings.name]))
self.__logger.info("Creating security groups for IXIA VMs")
sg_rules = list()
sg_rules.append(
- SecurityGroupRuleSettings(sec_grp_name="ixia_management",
- direction=Direction.ingress,
- protocol=Protocol.tcp, port_range_min=1,
- port_range_max=65535))
+ SecurityGroupRuleConfig(sec_grp_name="ixia_management",
+ direction=Direction.ingress,
+ protocol=Protocol.tcp, port_range_min=1,
+ port_range_max=65535))
sg_rules.append(
- SecurityGroupRuleSettings(sec_grp_name="ixia_management",
- direction=Direction.ingress,
- protocol=Protocol.udp, port_range_min=1,
- port_range_max=65535))
+ SecurityGroupRuleConfig(sec_grp_name="ixia_management",
+ direction=Direction.ingress,
+ protocol=Protocol.udp, port_range_min=1,
+ port_range_max=65535))
sg_rules.append(
- SecurityGroupRuleSettings(sec_grp_name="ixia_management",
- direction=Direction.ingress,
- protocol=Protocol.icmp))
+ SecurityGroupRuleConfig(sec_grp_name="ixia_management",
+ direction=Direction.ingress,
+ protocol=Protocol.icmp))
- ixia_managment_sg_settings = SecurityGroupSettings(
+ ixia_managment_sg_settings = SecurityGroupConfig(
name="ixia_management",
rule_settings=sg_rules)
securit_group_creator = OpenStackSecurityGroup(
sg_rules = list()
sg_rules.append(
- SecurityGroupRuleSettings(sec_grp_name="ixia_ssh_http",
- direction=Direction.ingress,
- protocol=Protocol.tcp, port_range_min=1,
- port_range_max=65535))
+ SecurityGroupRuleConfig(sec_grp_name="ixia_ssh_http",
+ direction=Direction.ingress,
+ protocol=Protocol.tcp, port_range_min=1,
+ port_range_max=65535))
- ixia_ssh_http_sg_settings = SecurityGroupSettings(
+ ixia_ssh_http_sg_settings = SecurityGroupConfig(
name="ixia_ssh_http",
rule_settings=sg_rules)
securit_group_creator = OpenStackSecurityGroup(
securit_group_creator.create()
self.created_object.append(securit_group_creator)
- chassis_flavor_settings = FlavorSettings(
+ chassis_flavor_settings = FlavorConfig(
name="ixia_vChassis",
ram=4096,
disk=40,
flavor_creator.create()
self.created_object.append(flavor_creator)
- card_flavor_settings = FlavorSettings(
+ card_flavor_settings = FlavorConfig(
name="ixia_vCard",
ram=4096,
disk=4,
flavor_creator.create()
self.created_object.append(flavor_creator)
- load_flavor_settings = FlavorSettings(
+ load_flavor_settings = FlavorConfig(
name="ixia_vLoad",
ram=8192,
disk=100,
flavor_creator.create()
self.created_object.append(flavor_creator)
- chassis_image_settings = ImageSettings(
+ chassis_image_settings = ImageConfig(
name=self.test['requirements']['chassis']['image'],
image_user='admin',
exists=True)
- card_image_settings = ImageSettings(
+ card_image_settings = ImageConfig(
name=self.test['requirements']['card']['image'],
image_user='admin',
exists=True)
- load_image_settings = ImageSettings(
+ load_image_settings = ImageConfig(
name=self.test['requirements']['load']['image'],
image_user='admin',
exists=True)
- chassis_port_settings = PortSettings(
+ chassis_port_settings = PortConfig(
name='ixia_chassis_port',
network_name=network_settings.name)
- card1_port1_settings = PortSettings(
+ card1_port1_settings = PortConfig(
name='ixia_card1_port1',
network_name=network_settings.name)
- card2_port1_settings = PortSettings(
+ card2_port1_settings = PortConfig(
name='ixia_card2_port1',
network_name=network_settings.name)
- card1_port2_settings = PortSettings(
+ card1_port2_settings = PortConfig(
name='ixia_card1_port2',
network_name="cloudify_ims_network")
- card2_port2_settings = PortSettings(
+ card2_port2_settings = PortConfig(
name='ixia_card2_port2',
network_name="cloudify_ims_network")
- load_port_settings = PortSettings(
+ load_port_settings = PortConfig(
name='ixia_load_port',
network_name=network_settings.name)
- chassis_settings = VmInstanceSettings(
+ chassis_settings = VmInstanceConfig(
name='ixia_vChassis',
flavor=chassis_flavor_settings.name,
port_settings=[chassis_port_settings],
security_group_names=[ixia_ssh_http_sg_settings.name,
ixia_managment_sg_settings.name],
- floating_ip_settings=[FloatingIpSettings(
+ floating_ip_settings=[FloatingIpConfig(
name='ixia_vChassis_fip',
port_name=chassis_port_settings.name,
router_name=router_creator.router_settings.name)])
fip_chassis = vm_creator.get_floating_ip().ip
self.created_object.append(vm_creator)
- card1_settings = VmInstanceSettings(
+ card1_settings = VmInstanceConfig(
name='ixia_vCard1',
flavor=card_flavor_settings.name,
port_settings=[card1_port1_settings, card1_port2_settings],
vcard_ips_p2.append(vm_creator.get_port_ip('ixia_card1_port2'))
self.created_object.append(vm_creator)
- card2_settings = VmInstanceSettings(
+ card2_settings = VmInstanceConfig(
name='ixia_vCard2',
flavor=card_flavor_settings.name,
port_settings=[card2_port1_settings, card2_port2_settings],
vcard_ips_p2.append(vm_creator.get_port_ip('ixia_card2_port2'))
self.created_object.append(vm_creator)
- load_settings = VmInstanceSettings(
+ load_settings = VmInstanceConfig(
name='ixia_vLoad',
flavor=load_flavor_settings.name,
port_settings=[load_port_settings],
security_group_names=[ixia_ssh_http_sg_settings.name,
ixia_managment_sg_settings.name],
- floating_ip_settings=[FloatingIpSettings(
+ floating_ip_settings=[FloatingIpConfig(
name='ixia_vLoad_fip',
port_name=load_port_settings.name,
router_name=router_creator.router_settings.name)])
import pkg_resources
import yaml
-from snaps.openstack.create_image import OpenStackImage, ImageSettings
-from snaps.openstack.create_flavor import OpenStackFlavor, FlavorSettings
-from snaps.openstack.create_security_group import (
- OpenStackSecurityGroup,
- SecurityGroupSettings,
- SecurityGroupRuleSettings,
- Direction,
- Protocol)
-from snaps.openstack.create_network import (
- OpenStackNetwork,
- NetworkSettings,
- SubnetSettings,
- PortSettings)
-from snaps.openstack.create_router import OpenStackRouter, RouterSettings
-from snaps.openstack.create_instance import (
- VmInstanceSettings,
- OpenStackVmInstance)
+from snaps.config.flavor import FlavorConfig
+from snaps.config.image import ImageConfig
+from snaps.config.network import NetworkConfig, PortConfig, SubnetConfig
+from snaps.config.router import RouterConfig
+from snaps.config.security_group import (
+ Direction, Protocol, SecurityGroupConfig, SecurityGroupRuleConfig)
+from snaps.config.vm_inst import VmInstanceConfig
+
+from snaps.openstack.create_flavor import OpenStackFlavor
+from snaps.openstack.create_image import OpenStackImage
+from snaps.openstack.create_instance import OpenStackVmInstance
+from snaps.openstack.create_network import OpenStackNetwork
+from snaps.openstack.create_router import OpenStackRouter
+from snaps.openstack.create_security_group import OpenStackSecurityGroup
+
from functest.opnfv_tests.openstack.snaps import snaps_utils
import functest.core.vnf as vnf
'vnf_{}_config'.format(self.case_name))
except BaseException:
raise Exception("Orchestra VNF config file not found")
+
config_file = self.case_dir + self.config
self.mano = dict(
if image_file and image_name:
image = OpenStackImage(
self.snaps_creds,
- ImageSettings(name=image_name,
- image_user='cloud',
- img_format='qcow2',
- image_file=image_file,
- public=True))
+ ImageConfig(name=image_name,
+ image_user='cloud',
+ img_format='qcow2',
+ image_file=image_file,
+ public=True))
image.create()
# self.created_resources.append(image);
"Creating security group for Open Baton if not yet existing...")
sg_rules = list()
sg_rules.append(
- SecurityGroupRuleSettings(
+ SecurityGroupRuleConfig(
sec_grp_name="orchestra-sec-group-allowall",
direction=Direction.ingress,
protocol=Protocol.tcp,
port_range_min=1,
port_range_max=65535))
sg_rules.append(
- SecurityGroupRuleSettings(
+ SecurityGroupRuleConfig(
sec_grp_name="orchestra-sec-group-allowall",
direction=Direction.egress,
protocol=Protocol.tcp,
port_range_min=1,
port_range_max=65535))
sg_rules.append(
- SecurityGroupRuleSettings(
+ SecurityGroupRuleConfig(
sec_grp_name="orchestra-sec-group-allowall",
direction=Direction.ingress,
protocol=Protocol.udp,
port_range_min=1,
port_range_max=65535))
sg_rules.append(
- SecurityGroupRuleSettings(
+ SecurityGroupRuleConfig(
sec_grp_name="orchestra-sec-group-allowall",
direction=Direction.egress,
protocol=Protocol.udp,
port_range_min=1,
port_range_max=65535))
sg_rules.append(
- SecurityGroupRuleSettings(
+ SecurityGroupRuleConfig(
sec_grp_name="orchestra-sec-group-allowall",
direction=Direction.ingress,
protocol=Protocol.icmp))
sg_rules.append(
- SecurityGroupRuleSettings(
+ SecurityGroupRuleConfig(
sec_grp_name="orchestra-sec-group-allowall",
direction=Direction.egress,
protocol=Protocol.icmp))
security_group = OpenStackSecurityGroup(
self.snaps_creds,
- SecurityGroupSettings(
+ SecurityGroupConfig(
name="orchestra-sec-group-allowall",
rule_settings=sg_rules))
self.logger.info(
"Create Flavor for Open Baton NFVO if not yet existing")
- flavor_settings = FlavorSettings(
+ flavor_settings = FlavorConfig(
name=self.mano['requirements']['flavor']['name'],
ram=self.mano['requirements']['flavor']['ram_min'],
disk=self.mano['requirements']['flavor']['disk'],
"""Create network/subnet/router if they doen't exist yet"""
self.logger.info(
"Creating network/subnet/router if they doen't exist yet...")
- subnet_settings = SubnetSettings(
+ subnet_settings = SubnetConfig(
name='%s_subnet' %
self.case_name,
cidr="192.168.100.0/24")
- network_settings = NetworkSettings(
+ network_settings = NetworkConfig(
name='%s_net' %
self.case_name,
subnet_settings=[subnet_settings])
self.created_resources.append(orchestra_network)
orchestra_router = OpenStackRouter(
self.snaps_creds,
- RouterSettings(
+ RouterConfig(
name='%s_router' %
self.case_name,
external_gateway=self.mano['details']['external_net_name'],
self.mano['details']['network']['id'])
self.logger.debug("userdata: %s\n", userdata)
# setting up image
- image_settings = ImageSettings(
+ image_settings = ImageConfig(
name=self.mano['requirements']['image'],
image_user='ubuntu',
exists=True)
# setting up port
- port_settings = PortSettings(
+ port_settings = PortConfig(
name='%s_port' % self.case_name,
network_name=self.mano['details']['network']['name'])
# build configuration of vm
- orchestra_settings = VmInstanceSettings(
+ orchestra_settings = VmInstanceConfig(
name=self.case_name,
flavor=self.mano['details']['flavor']['name'],
port_settings=[port_settings],
self.logger.info(
"Create %s Flavor if not existing", self.vnf['name'])
- flavor_settings = FlavorSettings(
+ flavor_settings = FlavorConfig(
name=self.vnf['requirements']['flavor']['name'],
ram=self.vnf['requirements']['flavor']['ram_min'],
disk=self.vnf['requirements']['flavor']['disk'],
import pkg_resources
import yaml
+from snaps.config.flavor import FlavorConfig
+from snaps.config.image import ImageConfig
+from snaps.config.network import NetworkConfig, PortConfig, SubnetConfig
+from snaps.config.router import RouterConfig
+from snaps.config.security_group import (
+ Direction, Protocol, SecurityGroupConfig, SecurityGroupRuleConfig)
+from snaps.config.vm_inst import VmInstanceConfig
+
+from snaps.openstack.create_image import OpenStackImage
+from snaps.openstack.create_flavor import OpenStackFlavor
+from snaps.openstack.create_security_group import OpenStackSecurityGroup
+from snaps.openstack.create_network import OpenStackNetwork
+from snaps.openstack.create_router import OpenStackRouter
+from snaps.openstack.create_instance import OpenStackVmInstance
-from snaps.openstack.create_image import OpenStackImage, ImageSettings
-from snaps.openstack.create_flavor import OpenStackFlavor, FlavorSettings
-from snaps.openstack.create_security_group import (
- OpenStackSecurityGroup,
- SecurityGroupSettings,
- SecurityGroupRuleSettings,
- Direction,
- Protocol)
-from snaps.openstack.create_network import (
- OpenStackNetwork,
- NetworkSettings,
- SubnetSettings,
- PortSettings)
-from snaps.openstack.create_router import OpenStackRouter, RouterSettings
-from snaps.openstack.create_instance import (
- VmInstanceSettings, OpenStackVmInstance)
from functest.opnfv_tests.openstack.snaps import snaps_utils
import functest.core.vnf as vnf
if image_file and image_name:
image = OpenStackImage(
self.snaps_creds,
- ImageSettings(name=image_name,
- image_user='cloud',
- img_format='qcow2',
- image_file=image_file,
- public=True))
+ ImageConfig(name=image_name,
+ image_user='cloud',
+ img_format='qcow2',
+ image_file=image_file,
+ public=True))
image.create()
# self.created_resources.append(image);
"Creating security group for Open Baton if not yet existing...")
sg_rules = list()
sg_rules.append(
- SecurityGroupRuleSettings(
+ SecurityGroupRuleConfig(
sec_grp_name="orchestra-sec-group-allowall",
direction=Direction.ingress,
protocol=Protocol.tcp,
port_range_min=1,
port_range_max=65535))
sg_rules.append(
- SecurityGroupRuleSettings(
+ SecurityGroupRuleConfig(
sec_grp_name="orchestra-sec-group-allowall",
direction=Direction.egress,
protocol=Protocol.tcp,
port_range_min=1,
port_range_max=65535))
sg_rules.append(
- SecurityGroupRuleSettings(
+ SecurityGroupRuleConfig(
sec_grp_name="orchestra-sec-group-allowall",
direction=Direction.ingress,
protocol=Protocol.udp,
port_range_min=1,
port_range_max=65535))
sg_rules.append(
- SecurityGroupRuleSettings(
+ SecurityGroupRuleConfig(
sec_grp_name="orchestra-sec-group-allowall",
direction=Direction.egress,
protocol=Protocol.udp,
port_range_max=65535))
security_group = OpenStackSecurityGroup(
self.snaps_creds,
- SecurityGroupSettings(
+ SecurityGroupConfig(
name="orchestra-sec-group-allowall",
rule_settings=sg_rules))
self.logger.info(
"Create Flavor for Open Baton NFVO if not yet existing")
- flavor_settings = FlavorSettings(
+ flavor_settings = FlavorConfig(
name=self.mano['requirements']['flavor']['name'],
ram=self.mano['requirements']['flavor']['ram_min'],
disk=self.mano['requirements']['flavor']['disk'],
"""Create network/subnet/router if they doen't exist yet"""
self.logger.info(
"Creating network/subnet/router if they doen't exist yet...")
- subnet_settings = SubnetSettings(
+ subnet_settings = SubnetConfig(
name='%s_subnet' %
self.case_name,
cidr="192.168.100.0/24")
- network_settings = NetworkSettings(
+ network_settings = NetworkConfig(
name='%s_net' %
self.case_name,
subnet_settings=[subnet_settings])
self.created_resources.append(orchestra_network)
orchestra_router = OpenStackRouter(
self.snaps_creds,
- RouterSettings(
+ RouterConfig(
name='%s_router' %
self.case_name,
external_gateway=self.mano['details']['external_net_name'],
self.mano['details']['network']['id'])
self.logger.debug("userdata: %s\n", userdata)
# setting up image
- image_settings = ImageSettings(
+ image_settings = ImageConfig(
name=self.mano['requirements']['image'],
image_user='ubuntu',
exists=True)
# setting up port
- port_settings = PortSettings(
+ port_settings = PortConfig(
name='%s_port' % self.case_name,
network_name=self.mano['details']['network']['name'])
# build configuration of vm
- orchestra_settings = VmInstanceSettings(
+ orchestra_settings = VmInstanceConfig(
name=self.case_name,
flavor=self.mano['details']['flavor']['name'],
port_settings=[port_settings],
self.logger.info(
"Create %s Flavor if not existing", self.vnf['name'])
- flavor_settings = FlavorSettings(
+ flavor_settings = FlavorConfig(
name=self.vnf['requirements']['flavor']['name'],
ram=self.vnf['requirements']['flavor']['ram_min'],
disk=self.vnf['requirements']['flavor']['disk'],
from git import Repo
-from snaps.openstack.create_network import (NetworkSettings, SubnetSettings,
- OpenStackNetwork)
-from snaps.openstack.create_security_group import (SecurityGroupSettings,
- SecurityGroupRuleSettings,
- Direction, Protocol,
- OpenStackSecurityGroup)
-from snaps.openstack.create_router import RouterSettings, OpenStackRouter
-from snaps.openstack.create_instance import (VmInstanceSettings,
- FloatingIpSettings,
- OpenStackVmInstance)
-from snaps.openstack.create_flavor import FlavorSettings, OpenStackFlavor
-from snaps.openstack.create_image import ImageSettings, OpenStackImage
-from snaps.openstack.create_keypairs import KeypairSettings, OpenStackKeypair
-from snaps.openstack.create_network import PortSettings
+from snaps.config.flavor import FlavorConfig
+from snaps.config.image import ImageConfig
+from snaps.config.keypair import KeypairConfig
+from snaps.config.network import NetworkConfig, PortConfig, SubnetConfig
+from snaps.config.router import RouterConfig
+from snaps.config.security_group import (
+ Direction, Protocol, SecurityGroupConfig, SecurityGroupRuleConfig)
+from snaps.config.vm_inst import FloatingIpConfig, VmInstanceConfig
+
+from snaps.openstack.create_flavor import OpenStackFlavor
+from snaps.openstack.create_image import OpenStackImage
+from snaps.openstack.create_instance import OpenStackVmInstance
+from snaps.openstack.create_keypairs import OpenStackKeypair
+from snaps.openstack.create_network import OpenStackNetwork
+from snaps.openstack.create_security_group import OpenStackSecurityGroup
+from snaps.openstack.create_router import OpenStackRouter
+
import snaps.openstack.utils.glance_utils as glance_utils
from functest.opnfv_tests.vnf.router.utilvnf import Utilvnf
if image_file and image_name:
image_creator = OpenStackImage(
self.snaps_creds,
- ImageSettings(name=image_name,
- image_user='cloud',
- img_format='qcow2',
- image_file=image_file))
+ ImageConfig(name=image_name,
+ image_user='cloud',
+ img_format='qcow2',
+ image_file=image_file))
image_creator.create()
self.created_object.append(image_creator)
start_time = time.time()
self.__logger.info("Creating keypair ...")
kp_file = os.path.join(self.data_dir, "cloudify_vrouter.pem")
- keypair_settings = KeypairSettings(name='cloudify_vrouter_kp',
- private_filepath=kp_file)
+ keypair_settings = KeypairConfig(name='cloudify_vrouter_kp',
+ private_filepath=kp_file)
keypair_creator = OpenStackKeypair(self.snaps_creds, keypair_settings)
keypair_creator.create()
self.created_object.append(keypair_creator)
self.__logger.info("Creating full network ...")
- subnet_settings = SubnetSettings(name='cloudify_vrouter_subnet',
- cidr='10.67.79.0/24')
- network_settings = NetworkSettings(name='cloudify_vrouter_network',
- subnet_settings=[subnet_settings])
+ subnet_settings = SubnetConfig(name='cloudify_vrouter_subnet',
+ cidr='10.67.79.0/24')
+ network_settings = NetworkConfig(name='cloudify_vrouter_network',
+ subnet_settings=[subnet_settings])
network_creator = OpenStackNetwork(self.snaps_creds, network_settings)
network_creator.create()
self.created_object.append(network_creator)
ext_net_name = snaps_utils.get_ext_net_name(self.snaps_creds)
router_creator = OpenStackRouter(
self.snaps_creds,
- RouterSettings(
+ RouterConfig(
name='cloudify_vrouter_router',
external_gateway=ext_net_name,
internal_subnets=[subnet_settings.name]))
self.__logger.info("Creating security group for cloudify manager vm")
sg_rules = list()
sg_rules.append(
- SecurityGroupRuleSettings(sec_grp_name="sg-cloudify-manager",
- direction=Direction.ingress,
- protocol=Protocol.tcp, port_range_min=1,
- port_range_max=65535))
+ SecurityGroupRuleConfig(sec_grp_name="sg-cloudify-manager",
+ direction=Direction.ingress,
+ protocol=Protocol.tcp, port_range_min=1,
+ port_range_max=65535))
sg_rules.append(
- SecurityGroupRuleSettings(sec_grp_name="sg-cloudify-manager",
- direction=Direction.ingress,
- protocol=Protocol.udp, port_range_min=1,
- port_range_max=65535))
+ SecurityGroupRuleConfig(sec_grp_name="sg-cloudify-manager",
+ direction=Direction.ingress,
+ protocol=Protocol.udp, port_range_min=1,
+ port_range_max=65535))
security_group_creator = OpenStackSecurityGroup(
self.snaps_creds,
- SecurityGroupSettings(
+ SecurityGroupConfig(
name="sg-cloudify-manager",
rule_settings=sg_rules))
# orchestrator VM flavor
self.__logger.info("Get or create flavor for cloudify manager vm ...")
- flavor_settings = FlavorSettings(
+ flavor_settings = FlavorConfig(
name=self.orchestrator['requirements']['flavor']['name'],
ram=self.orchestrator['requirements']['flavor']['ram_min'],
disk=50,
flavor_creator = OpenStackFlavor(self.snaps_creds, flavor_settings)
flavor_creator.create()
self.created_object.append(flavor_creator)
- image_settings = ImageSettings(
+ image_settings = ImageConfig(
name=self.orchestrator['requirements']['os_image'],
image_user='centos',
exists=True)
- port_settings = PortSettings(name='cloudify_manager_port',
- network_name=network_settings.name)
+ port_settings = PortConfig(name='cloudify_manager_port',
+ network_name=network_settings.name)
- manager_settings = VmInstanceSettings(
+ manager_settings = VmInstanceConfig(
name='cloudify_manager',
flavor=flavor_settings.name,
port_settings=[port_settings],
security_group_names=[
security_group_creator.sec_grp_settings.name],
- floating_ip_settings=[FloatingIpSettings(
+ floating_ip_settings=[FloatingIpConfig(
name='cloudify_manager_fip',
port_name=port_settings.name,
router_name=router_creator.router_settings.name)])
descriptor.get('name'))
self.__logger.info("Get or create flavor for vrouter")
- flavor_settings = FlavorSettings(
+ flavor_settings = FlavorConfig(
name=self.vnf['requirements']['flavor']['name'],
ram=self.vnf['requirements']['flavor']['ram_min'],
disk=25,
import logging
import unittest
+import os
import mock
self.run_tests_parser = run_tests.RunTestsParser()
+ self.config_file_yaml = {'general': {
+ 'openstack': {
+ 'image_name': 'test_image_name'}},
+ 'results': {
+ 'test_db_url': 'url1'}}
+ self.config_file_patch_yaml = {'fdio': {'general': {
+ 'openstack': {
+ 'image_name':
+ 'test_image_name_2'}}}}
+ self.config_file_aarcg64_patch_yaml = {'os': {'general': {
+ 'openstack': {'image_name': 'test_image_name_3'}}}}
+
+ @mock.patch('functest.ci.run_tests.Runner.patch_file')
+ @mock.patch('functest.ci.run_tests.Runner.update_db_url')
+ def test_update_config_file_default(self, *mock_methods):
+ self.runner.update_config_file()
+ mock_methods[1].assert_called()
+ mock_methods[0].assert_not_called()
+
+ @mock.patch('functest.ci.run_tests.Runner.patch_file')
+ @mock.patch('functest.ci.run_tests.Runner.update_db_url')
+ @mock.patch.dict(os.environ, {'TEST_DB_URL': 'somevalue'})
+ def test_update_config_file_update_db(self, *mock_methods):
+ self.runner.update_config_file()
+ mock_methods[1].assert_called()
+ mock_methods[0].assert_called()
+
+ def test_patch_file_missing_file(self):
+ patch_file_path = "unexisting_file"
+ with self.assertRaises(IOError):
+ self.runner.patch_file(patch_file_path)
+
+ @mock.patch('functest.ci.run_tests.ft_utils.merge_dicts')
+ @mock.patch('functest.ci.run_tests.ft_utils.get_functest_yaml')
+ def test_patch_file_default(self, *mock_methods):
+ CONST.__setattr__('DEPLOY_SCENARIO', 'os-nosdn-nofeature-noha')
+ with mock.patch(
+ 'six.moves.builtins.open', mock.mock_open()), mock.patch(
+ 'functest.ci.run_tests.yaml.safe_load') as yaml1, mock.patch(
+ 'functest.ci.run_tests.ft_utils.get_functest_yaml') as yaml2:
+ yaml1.return_value = self.config_file_patch_yaml
+ yaml2.return_value = self.config_file_yaml
+ self.runner.patch_file(yaml1)
+ mock_methods[1].assert_not_called()
+ mock_methods[0].assert_not_called()
+
+ @mock.patch('functest.ci.run_tests.ft_utils.merge_dicts')
+ @mock.patch('functest.ci.run_tests.os.remove')
+ def test_patch_file_match_scenario(self, *mock_methods):
+ CONST.__setattr__('DEPLOY_SCENARIO', 'os-nosdn-fdio-noha')
+ with mock.patch(
+ 'six.moves.builtins.open', mock.mock_open()), mock.patch(
+ 'functest.ci.run_tests.yaml.safe_load') as yaml1, mock.patch(
+ 'functest.ci.run_tests.ft_utils.get_functest_yaml') as yaml2:
+ yaml1.return_value = self.config_file_patch_yaml
+ yaml2.return_value = self.config_file_yaml
+ self.runner.patch_file(yaml2)
+ mock_methods[1].assert_called()
+ mock_methods[0].assert_called()
+
+ def test_update_db_url_missing_file(self):
+ run_tests.CONFIG_FUNCTEST_PATH = "unexisting_file"
+ with self.assertRaises(IOError):
+ self.runner.update_db_url()
+
+ @mock.patch('functest.ci.run_tests.yaml.safe_load')
+ @mock.patch('functest.ci.run_tests.ft_utils.get_functest_yaml')
+ @mock.patch.dict(os.environ, {'TEST_DB_URL': 'url2'})
+ def test_update_db_url_default(self, *mock_methods):
+ with mock.patch(
+ 'six.moves.builtins.open', mock.mock_open()), mock.patch(
+ 'functest.ci.run_tests.yaml.safe_load') as yaml1:
+ yaml1.return_value = self.config_file_yaml
+ self.runner.update_db_url()
+ self.assertEqual(
+ yaml1.return_value['results']['test_db_url'], 'url2')
+ mock_methods[0].assert_not_called()
+ mock_methods[1].assert_not_called()
+
@mock.patch('functest.ci.run_tests.logger.error')
def test_source_rc_file_missing_file(self, mock_logger_error):
with mock.patch('functest.ci.run_tests.os.path.isfile',
def setUp(self):
self.cli_environ = cli_env.CliEnv()
- @mock.patch('functest.cli.commands.cli_testcase.os.path.isfile',
- return_value=False)
- @mock.patch('functest.cli.commands.cli_testcase.ft_utils.execute_command')
- def test_prepare_default(self, mock_ft_utils, mock_os):
- cmd = "prepare_env start"
- self.cli_environ.prepare()
- mock_ft_utils.assert_called_with(cmd)
-
- @mock.patch('functest.cli.commands.cli_testcase.os.path.isfile',
- return_value=True)
- @mock.patch('functest.cli.commands.cli_testcase.ft_utils.execute_command')
- def test_prepare_missing_status(self, mock_ft_utils, mock_os):
- with mock.patch('__builtin__.raw_input', return_value="y"), \
- mock.patch('functest.cli.commands.cli_testcase.os.remove') \
- as mock_os_remove:
- cmd = "prepare_env start"
- self.cli_environ.prepare()
- mock_os_remove.assert_called_once_with(
- CONST.__getattribute__('env_active'))
- mock_ft_utils.assert_called_with(cmd)
-
def _test_show_missing_env_var(self, var, *args):
if var == 'INSTALLER_TYPE':
CONST.__setattr__('INSTALLER_TYPE', None)
elif var == 'DEBUG':
CONST.__setattr__('CI_DEBUG', None)
reg_string = "| DEBUG FLAG: false\s*|"
- elif var == 'STATUS':
- reg_string = "| STATUS: not ready\s*|"
with mock.patch('functest.cli.commands.cli_env.click.echo') \
as mock_click_echo:
def test_show_missing_ci_debug(self, *args):
self._test_show_missing_env_var('DEBUG', *args)
- @mock.patch('functest.cli.commands.cli_env.os.path.isfile',
- return_value=False)
- def test_show_missing_environment(self, *args):
- self._test_show_missing_env_var('STATUS', *args)
-
- @mock.patch('functest.cli.commands.cli_env.click.echo')
- @mock.patch('functest.cli.commands.cli_env.os.path.isfile',
- return_value=True)
- def test_status_environment_present(self, mock_path, mock_click_echo):
- self.assertEqual(self.cli_environ.status(), 0)
- mock_click_echo.assert_called_with("Functest environment"
- " ready to run tests.\n")
-
- @mock.patch('functest.cli.commands.cli_env.click.echo')
- @mock.patch('functest.cli.commands.cli_env.os.path.isfile',
- return_value=False)
- def test_status_environment_absent(self, mock_path, mock_click_echo):
- self.assertEqual(self.cli_environ.status(), 1)
- mock_click_echo.assert_called_with("Functest environment"
- " is not installed.\n")
-
if __name__ == "__main__":
logging.disable(logging.CRITICAL)
self.cli_tests.run('vacation')
self.assertTrue(mock_method.called)
- @mock.patch('functest.cli.commands.cli_testcase.os.path.isfile',
- return_value=False)
- @mock.patch('functest.cli.commands.cli_testcase.click.echo')
- def test_run_missing_env_file(self, mock_click_echo, mock_os):
- self.cli_tests.run(self.testname)
- mock_click_echo.assert_called_with("Functest environment is not ready."
- " Run first 'functest env prepare'")
-
- @mock.patch('functest.cli.commands.cli_testcase.os.path.isfile',
- return_value=True)
@mock.patch('functest.cli.commands.cli_testcase.ft_utils.execute_command')
- def test_run_default(self, mock_ft_utils, mock_os):
+ def test_run_default(self, mock_ft_utils):
cmd = "run_tests -n -r -t {}".format(self.testname)
self.cli_tests.run(self.testname, noclean=True, report=True)
mock_ft_utils.assert_called_with(cmd)
- @mock.patch('functest.cli.commands.cli_testcase.os.path.isfile',
- return_value=True)
@mock.patch('functest.cli.commands.cli_testcase.ft_utils.execute_command')
- def test_run_noclean_missing_report(self, mock_ft_utils, mock_os):
+ def test_run_noclean_missing_report(self, mock_ft_utils):
cmd = "run_tests -n -t {}".format(self.testname)
self.cli_tests.run(self.testname, noclean=True, report=False)
mock_ft_utils.assert_called_with(cmd)
- @mock.patch('functest.cli.commands.cli_testcase.os.path.isfile',
- return_value=True)
@mock.patch('functest.cli.commands.cli_testcase.ft_utils.execute_command')
- def test_run_report_missing_noclean(self, mock_ft_utils, mock_os):
+ def test_run_report_missing_noclean(self, mock_ft_utils):
cmd = "run_tests -r -t {}".format(self.testname)
self.cli_tests.run(self.testname, noclean=False, report=True)
mock_ft_utils.assert_called_with(cmd)
- @mock.patch('functest.cli.commands.cli_testcase.os.path.isfile',
- return_value=True)
@mock.patch('functest.cli.commands.cli_testcase.ft_utils.execute_command')
- def test_run_missing_noclean_report(self, mock_ft_utils, mock_os):
+ def test_run_missing_noclean_report(self, mock_ft_utils):
cmd = "run_tests -t {}".format(self.testname)
self.cli_tests.run(self.testname, noclean=False, report=False)
mock_ft_utils.assert_called_with(cmd)
":\n %s\n" % (self.tiername,
'tiernames'))
- @mock.patch('functest.cli.commands.cli_tier.os.path.isfile',
- return_value=False)
- @mock.patch('functest.cli.commands.cli_tier.click.echo')
- def test_run_missing_env_file(self, mock_click_echo, mock_os):
- self.cli_tier.run(self.tiername)
- mock_click_echo.assert_called_with("Functest environment is not ready."
- " Run first 'functest env prepare'")
-
- @mock.patch('functest.cli.commands.cli_tier.os.path.isfile',
- return_value=True)
@mock.patch('functest.cli.commands.cli_tier.ft_utils.execute_command')
- def test_run_default(self, mock_ft_utils, mock_os):
+ def test_run_default(self, mock_ft_utils):
cmd = "run_tests -n -r -t {}".format(self.tiername)
self.cli_tier.run(self.tiername, noclean=True, report=True)
mock_ft_utils.assert_called_with(cmd)
- @mock.patch('functest.cli.commands.cli_tier.os.path.isfile',
- return_value=True)
@mock.patch('functest.cli.commands.cli_tier.ft_utils.execute_command')
- def test_run_report_missing_noclean(self, mock_ft_utils, mock_os):
+ def test_run_report_missing_noclean(self, mock_ft_utils):
cmd = "run_tests -r -t {}".format(self.tiername)
self.cli_tier.run(self.tiername, noclean=False, report=True)
mock_ft_utils.assert_called_with(cmd)
- @mock.patch('functest.cli.commands.cli_tier.os.path.isfile',
- return_value=True)
@mock.patch('functest.cli.commands.cli_tier.ft_utils.execute_command')
- def test_run_noclean_missing_report(self, mock_ft_utils, mock_os):
+ def test_run_noclean_missing_report(self, mock_ft_utils):
cmd = "run_tests -n -t {}".format(self.tiername)
self.cli_tier.run(self.tiername, noclean=True, report=False)
mock_ft_utils.assert_called_with(cmd)
- @mock.patch('functest.cli.commands.cli_tier.os.path.isfile',
- return_value=True)
@mock.patch('functest.cli.commands.cli_tier.ft_utils.execute_command')
- def test_run_missing_noclean_report(self, mock_ft_utils, mock_os):
+ def test_run_missing_noclean_report(self, mock_ft_utils):
cmd = "run_tests -t {}".format(self.tiername)
self.cli_tier.run(self.tiername, noclean=False, report=False)
mock_ft_utils.assert_called_with(cmd)
self.assertEqual(result.exit_code, 0)
self.assertTrue(mock_method.called)
- def test_env_prepare(self):
- with mock.patch.object(self._env, 'prepare') as mock_method:
- result = self.runner.invoke(cli_base.env_prepare)
- self.assertEqual(result.exit_code, 0)
- self.assertTrue(mock_method.called)
-
def test_env_show(self):
with mock.patch.object(self._env, 'show') as mock_method:
result = self.runner.invoke(cli_base.env_show)
self.assertEqual(result.exit_code, 0)
self.assertTrue(mock_method.called)
- def test_env_status(self):
- with mock.patch.object(self._env, 'status') as mock_method:
- result = self.runner.invoke(cli_base.env_status)
- self.assertEqual(result.exit_code, 0)
- self.assertTrue(mock_method.called)
-
def test_testcase_list(self):
with mock.patch.object(self._testcase, 'list') as mock_method:
result = self.runner.invoke(cli_base.testcase_list)
return_value=True)
@mock.patch('functest.opnfv_tests.openstack.rally.rally.subprocess.Popen')
@mock.patch('functest.opnfv_tests.openstack.rally.rally.os.makedirs')
- @mock.patch('functest.opnfv_tests.openstack.rally.rally.os.popen')
@mock.patch('functest.opnfv_tests.openstack.rally.rally.LOGGER.info')
@mock.patch('functest.opnfv_tests.openstack.rally.rally.LOGGER.error')
def test_run_task_default(self, mock_logger_error, mock_logger_info,
- mock_popen, *args):
- attrs = {'read.return_value': 'json_result'}
- mock_popen.return_value.configure_mock(**attrs)
+ *args):
self.rally_base._run_task('test_name')
text = 'Test scenario: "test_name" OK.\n'
mock_logger_info.assert_any_call(text)
import mock
+from snaps.config.keypair import KeypairConfig
+from snaps.config.network import NetworkConfig, PortConfig, SubnetConfig
+from snaps.config.router import RouterConfig
+from snaps.config.security_group import SecurityGroupConfig
+from snaps.config.vm_inst import VmInstanceConfig
+
from snaps.openstack.create_image import OpenStackImage
-from snaps.openstack.create_instance import OpenStackVmInstance, \
- VmInstanceSettings
-from snaps.openstack.create_keypairs import OpenStackKeypair, KeypairSettings
-from snaps.openstack.create_network import OpenStackNetwork, NetworkSettings, \
- SubnetSettings, PortSettings
-from snaps.openstack.create_router import OpenStackRouter, RouterSettings
-from snaps.openstack.create_security_group import OpenStackSecurityGroup, \
- SecurityGroupSettings
+from snaps.openstack.create_instance import OpenStackVmInstance
+from snaps.openstack.create_keypairs import OpenStackKeypair
+from snaps.openstack.create_network import OpenStackNetwork
+from snaps.openstack.create_router import OpenStackRouter
+from snaps.openstack.create_security_group import OpenStackSecurityGroup
+
from snaps.openstack.os_credentials import OSCreds
from functest.core.testcase import TestCase
return_value=OpenStackImage(self.os_creds, None)), \
mock.patch('snaps.openstack.utils.deploy_utils.create_network',
return_value=OpenStackNetwork(
- self.os_creds, NetworkSettings(name='foo'))), \
+ self.os_creds, NetworkConfig(name='foo'))), \
mock.patch('snaps.openstack.utils.deploy_utils.'
'create_vm_instance',
return_value=OpenStackVmInstance(
self.os_creds,
- VmInstanceSettings(
+ VmInstanceConfig(
name='foo', flavor='bar',
- port_settings=[PortSettings(
+ port_settings=[PortConfig(
name='foo', network_name='bar')]),
None)), \
mock.patch('snaps.openstack.create_instance.'
mock.patch('snaps.openstack.utils.deploy_utils.create_network',
return_value=OpenStackNetwork(
self.os_creds,
- NetworkSettings(
+ NetworkConfig(
name='foo',
subnet_settings=[
- SubnetSettings(
+ SubnetConfig(
name='bar',
cidr='10.0.0.1/24')]))), \
mock.patch('snaps.openstack.utils.deploy_utils.'
'create_vm_instance',
return_value=OpenStackVmInstance(
self.os_creds,
- VmInstanceSettings(
+ VmInstanceConfig(
name='foo', flavor='bar',
- port_settings=[PortSettings(
+ port_settings=[PortConfig(
name='foo', network_name='bar')]),
None)), \
mock.patch('snaps.openstack.utils.deploy_utils.create_keypair',
return_value=OpenStackKeypair(
- self.os_creds, KeypairSettings(name='foo'))), \
+ self.os_creds, KeypairConfig(name='foo'))), \
mock.patch('snaps.openstack.utils.deploy_utils.create_router',
return_value=OpenStackRouter(
- self.os_creds, RouterSettings(name='foo'))), \
+ self.os_creds, RouterConfig(name='foo'))), \
mock.patch('snaps.openstack.utils.deploy_utils.'
'create_security_group',
return_value=OpenStackSecurityGroup(
self.os_creds,
- SecurityGroupSettings(name='foo'))), \
+ SecurityGroupConfig(name='foo'))), \
mock.patch('snaps.openstack.create_instance.'
'OpenStackVmInstance.'
'get_vm_inst', return_value=os_vm_inst), \
#!/usr/bin/env python
-import os
import yaml
import six
'{}_{}'.format(attr_now, next) if attr_now else next)
def _set_others(self):
- self.env_active = os.path.join(self.dir_functest_conf, "env_active")
-
+ pass
CONF = Config()
functest_tempest_conf = functest.opnfv_tests.openstack.refstack_client.tempest_conf:main
openstack_snapshot = functest.utils.openstack_snapshot:main
openstack_clean = functest.utils.openstack_clean:main
- prepare_env = functest.ci.prepare_env:main
run_tests = functest.ci.run_tests:main
check_deployment = functest.ci.check_deployment:main
functest_restapi = functest.api.server:main
Code Review / functest-xtesting.git / commitdiff