from scp import SCPClient
import yaml
-from functest.energy import energy
-from functest.opnfv_tests.openstack.snaps import snaps_utils
-import functest.opnfv_tests.vnf.ims.clearwater_ims_base as clearwater_ims_base
-from functest.utils.constants import CONST
-
from snaps.config.flavor import FlavorConfig
from snaps.config.image import ImageConfig
from snaps.config.keypair import KeypairConfig
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_router import OpenStackRouter
from snaps.openstack.create_security_group import OpenStackSecurityGroup
from snaps.openstack.utils import keystone_utils
+from xtesting.energy import energy
+from functest.opnfv_tests.openstack.snaps import snaps_utils
+import functest.opnfv_tests.vnf.ims.clearwater_ims_base as clearwater_ims_base
+from functest.utils import config
+from functest.utils import env
__author__ = "Valentin Boucher <valentin.boucher@orange.com>"
# Retrieve the configuration
try:
- self.config = CONST.__getattribute__(
- 'vnf_{}_config'.format(self.case_name))
+ self.config = getattr(
+ config.CONF, 'vnf_{}_config'.format(self.case_name))
except Exception:
raise Exception("VNF config file not found")
self.__logger.info("Creating full network ...")
subnet_settings = SubnetConfig(
name='cloudify_ims_subnet-{}'.format(self.uuid),
- cidr='10.67.79.0/24')
+ cidr='10.67.79.0/24',
+ dns_nameservers=[env.get('NAMESERVER')])
network_settings = NetworkConfig(
name='cloudify_ims_network-{}'.format(self.uuid),
subnet_settings=[subnet_settings])
keystone_password=self.snaps_creds.password,
keystone_tenant_name=self.snaps_creds.project_name,
keystone_url=public_auth_url,
- region=self.snaps_creds.region_name)
+ region=self.snaps_creds.region_name,
+ user_domain_name=self.snaps_creds.user_domain_name,
+ project_domain_name=self.snaps_creds.project_domain_name)
self.__logger.info("Set creds for cloudify manager %s", cfy_creds)
cfy_client = CloudifyClient(
while str(cfy_status) != 'running' and retry:
try:
cfy_status = cfy_client.manager.get_status()['status']
- self.__logger.debug("The current manager status is %s",
- cfy_status)
+ self.__logger.info(
+ "The current manager status is %s", cfy_status)
except Exception: # pylint: disable=broad-except
- self.__logger.exception(
+ self.__logger.info(
"Cloudify Manager isn't up and running. Retrying ...")
retry = retry - 1
time.sleep(30)
scp = SCPClient(ssh.get_transport(), socket_timeout=15.0)
scp.put(kp_file, '~/')
cmd = "sudo cp ~/cloudify_ims.pem /etc/cloudify/"
- run_blocking_ssh_command(ssh, cmd)
+ self.run_blocking_ssh_command(ssh, cmd)
cmd = "sudo chmod 444 /etc/cloudify/cloudify_ims.pem"
- run_blocking_ssh_command(ssh, cmd)
+ self.run_blocking_ssh_command(ssh, cmd)
cmd = "sudo yum install -y gcc python-devel"
- run_blocking_ssh_command(ssh, cmd, "Unable to install packages \
- on manager")
+ self.run_blocking_ssh_command(
+ ssh, cmd, "Unable to install packages on manager")
self.details['orchestrator'].update(status='PASS', duration=duration)
try:
cfy_client.executions.cancel(execution['id'],
force=True)
- except: # pylint: disable=broad-except
+ except Exception: # pylint: disable=broad-except
self.__logger.warn("Can't cancel the current exec")
execution = cfy_client.executions.start(
wait_for_execution(cfy_client, execution, self.__logger)
cfy_client.deployments.delete(self.vnf['descriptor'].get('name'))
cfy_client.blueprints.delete(self.vnf['descriptor'].get('name'))
- except: # pylint: disable=broad-except
- self.__logger.warn("Some issue during the undeployment ..")
- self.__logger.warn("Tenant clean continue ..")
+ except Exception: # pylint: disable=broad-except
+ self.__logger.exception("Some issue during the undeployment ..")
super(CloudifyIms, self).clean()
+ @staticmethod
+ def run_blocking_ssh_command(ssh, cmd,
+ error_msg="Unable to run this command"):
+ """Command to run ssh command with the exit status."""
+ _, stdout, stderr = ssh.exec_command(cmd)
+ CloudifyIms.__logger.debug("SSH %s stdout: %s", cmd, stdout.read())
+ if stdout.channel.recv_exit_status() != 0:
+ CloudifyIms.__logger.error("SSH %s stderr: %s", cmd, stderr.read())
+ raise Exception(error_msg)
+
@energy.enable_recording
def run(self, **kwargs):
"""Execute CloudifyIms test case."""
short_sig_test_result['skipped'] = nb_skipped
nb_test = nb_passed + nb_skipped
return (short_sig_test_result, nb_test)
-
-
-def run_blocking_ssh_command(ssh, cmd, error_msg="Unable to run this command"):
- """Command to run ssh command with the exit status."""
- stdin, stdout, stderr = ssh.exec_command(cmd)
- if stdout.channel.recv_exit_status() != 0:
- raise Exception(error_msg)