import time
import pkg_resources
-import scp
from functest.core import cloudify
from functest.opnfv_tests.vnf.router import vrouter_base
from functest.opnfv_tests.vnf.router.utilvnf import Utilvnf
from functest.utils import config
+from functest.utils import env
from functest.utils import functest_utils
self.image_alt = None
self.flavor_alt = None
+ def check_requirements(self):
+ if env.get('NEW_USER_ROLE').lower() == "admin":
+ self.__logger.warning(
+ "Defining NEW_USER_ROLE=admin will easily break the testcase "
+ "because Cloudify doesn't manage tenancy (e.g. subnet "
+ "overlapping)")
+
def execute(self):
# pylint: disable=too-many-locals,too-many-statements
"""
# network creation
super(CloudifyVrouter, self).execute()
start_time = time.time()
- self.__logger.info("Put private keypair in manager")
- scpc = scp.SCPClient(self.ssh.get_transport())
- scpc.put(self.key_filename, remote_path='~/cloudify_ims.pem')
- (_, stdout, stderr) = self.ssh.exec_command(
- "sudo docker exec cfy_manager_local "
- "cfy plugins upload -y {} {} && "
- "sudo docker cp ~/cloudify_ims.pem "
- "cfy_manager_local:/etc/cloudify/ && "
- "sudo docker exec cfy_manager_local "
- "chmod 444 /etc/cloudify/cloudify_ims.pem && "
- "sudo docker exec cfy_manager_local cfy status".format(
- self.cop_yaml, self.cop_wgn))
- self.__logger.info("output:\n%s", stdout.read())
- self.__logger.info("error:\n%s", stderr.read())
+ self.put_private_key()
+ self.upload_cfy_plugins(self.cop_yaml, self.cop_wgn)
self.image_alt = self.publish_image_alt()
self.flavor_alt = self.create_flavor_alt()
def test_vnf(self):
start_time = time.time()
- testing = vrouter_base.VrouterOnBoardingBase(
- self.case_name, self.util, self.util_info)
+ testing = vrouter_base.VrouterOnBoardingBase(self.util, self.util_info)
result, test_result_data = testing.test_vnf()
duration = time.time() - start_time
if result:
return True
def clean(self):
- try:
- dep_name = self.vnf['descriptor'].get('name')
- # kill existing execution
- self.__logger.info('Deleting the current deployment')
- exec_list = self.cfy_client.executions.list()
- for execution in exec_list:
- if execution['status'] == "started":
- try:
- self.cfy_client.executions.cancel(
- execution['id'], force=True)
- except Exception: # pylint: disable=broad-except
- self.__logger.warn("Can't cancel the current exec")
-
- execution = self.cfy_client.executions.start(
- dep_name, 'uninstall', parameters=dict(ignore_failure=True))
-
- cloudify.wait_for_execution(
- self.cfy_client, execution, self.__logger)
- self.cfy_client.deployments.delete(
- self.vnf['descriptor'].get('name'))
- time.sleep(10)
- self.cfy_client.blueprints.delete(
- self.vnf['descriptor'].get('name'))
- except Exception: # pylint: disable=broad-except
- self.__logger.exception("Some issue during the undeployment ..")
+ self.kill_existing_execution(self.vnf['descriptor'].get('name'))
if self.image_alt:
self.cloud.delete_image(self.image_alt)
if self.flavor_alt: