# http://www.apache.org/licenses/LICENSE-2.0
"""Juju testcase implementation."""
+import errno
import logging
import os
import time
import json
+import re
import sys
-from copy import deepcopy
-from urlparse import urljoin
-
-import functest.core.vnf as vnf
-from functest.opnfv_tests.openstack.snaps import snaps_utils
-from functest.utils.constants import CONST
-import functest.utils.openstack_utils as os_utils
+from copy import deepcopy
import pkg_resources
-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.create_image import OpenStackImage
-from snaps.openstack.create_network import OpenStackNetwork
-from snaps.openstack.create_router import OpenStackRouter
-from snaps.openstack.utils import keystone_utils
-import yaml
+import scp
+
+from functest.core import singlevm
+from functest.utils import config
+from functest.utils import env
+from functest.utils import functest_utils
__author__ = "Amarendra Meher <amarendra@rebaca.com>"
__author__ = "Soumaya K Nayek <soumaya.nayek@rebaca.com>"
+CLOUD_TEMPLATE = """clouds:
+ abot-epc:
+ type: openstack
+ auth-types: [userpass]
+ endpoint: {url}
+ regions:
+ {region}:
+ endpoint: {url}"""
+
+CREDS_TEMPLATE2 = """credentials:
+ abot-epc:
+ default-credential: abot-epc
+ abot-epc:
+ auth-type: userpass
+ password: '{pass}'
+ project-domain-name: {project_domain_n}
+ tenant-name: {tenant_n}"""
+
+CREDS_TEMPLATE = """credentials:
+ abot-epc:
+ default-credential: abot-epc
+ abot-epc:
+ auth-type: userpass
+ password: '{pass}'
+ project-domain-name: {project_domain_n}
+ tenant-name: {tenant_n}
+ user-domain-name: {user_domain_n}
+ username: {user_n}"""
+
-class JujuEpc(vnf.VnfOnBoarding):
+class JujuEpc(singlevm.SingleVm2):
# pylint:disable=too-many-instance-attributes
"""Abot EPC deployed with JUJU Orchestrator Case"""
__logger = logging.getLogger(__name__)
- default_region_name = "RegionOne"
+ cidr = '192.168.120.0/24'
+
+ filename = ('/home/opnfv/functest/images/'
+ 'ubuntu-16.04-server-cloudimg-amd64-disk1.img')
+ filename_alt = ('/home/opnfv/functest/images/'
+ 'ubuntu-14.04-server-cloudimg-amd64-disk1.img')
+
+ flavor_ram = 2048
+ flavor_vcpus = 1
+ flavor_disk = 10
+ flavor_alt_ram = 4096
+ flavor_alt_vcpus = 1
+ flavor_alt_disk = 10
+ username = 'ubuntu'
+ juju_timeout = '4800'
def __init__(self, **kwargs):
if "case_name" not in kwargs:
kwargs["case_name"] = "juju_epc"
- super(JujuEpc, self).__init__(**kwargs)
+ super().__init__(**kwargs)
# Retrieve the configuration
self.case_dir = pkg_resources.resource_filename(
'functest', 'opnfv_tests/vnf/epc')
try:
- self.config = CONST.__getattribute__(
- 'vnf_{}_config'.format(self.case_name))
- except Exception:
- raise Exception("VNF config file not found")
- config_file = os.path.join(self.case_dir, self.config)
+ self.config = getattr(
+ config.CONF, f'vnf_{self.case_name}_config')
+ except Exception as exc:
+ raise Exception("VNF config file not found") from exc
+ self.config_file = os.path.join(self.case_dir, self.config)
self.orchestrator = dict(
- requirements=get_config("orchestrator.requirements", config_file),
- )
+ requirements=functest_utils.get_parameter_from_yaml(
+ "orchestrator.requirements", self.config_file))
self.created_object = []
- self.snaps_creds = snaps_utils.get_credentials()
self.details['orchestrator'] = dict(
- name=get_config("orchestrator.name", config_file),
- version=get_config("orchestrator.version", config_file),
+ name=functest_utils.get_parameter_from_yaml(
+ "orchestrator.name", self.config_file),
+ version=functest_utils.get_parameter_from_yaml(
+ "orchestrator.version", self.config_file),
status='ERROR',
result=''
)
self.vnf = dict(
- descriptor=get_config("vnf.descriptor", config_file),
- requirements=get_config("vnf.requirements", config_file)
+ descriptor=functest_utils.get_parameter_from_yaml(
+ "vnf.descriptor", self.config_file),
+ requirements=functest_utils.get_parameter_from_yaml(
+ "vnf.requirements", self.config_file)
)
self.details['vnf'] = dict(
descriptor_version=self.vnf['descriptor']['version'],
- name=get_config("vnf.name", config_file),
- version=get_config("vnf.version", config_file),
+ name=functest_utils.get_parameter_from_yaml(
+ "vnf.name", self.config_file),
+ version=functest_utils.get_parameter_from_yaml(
+ "vnf.version", self.config_file),
)
self.__logger.debug("VNF configuration: %s", self.vnf)
self.details['test_vnf'] = dict(
- name=get_config("vnf_test_suite.name", config_file),
- version=get_config("vnf_test_suite.version", config_file),
- tag_name=get_config("vnf_test_suite.tag_name", config_file)
+ name=functest_utils.get_parameter_from_yaml(
+ "vnf_test_suite.name", self.config_file),
+ version=functest_utils.get_parameter_from_yaml(
+ "vnf_test_suite.version", self.config_file),
+ tag_name=functest_utils.get_parameter_from_yaml(
+ "vnf_test_suite.tag_name", self.config_file)
)
- self.images = get_config("tenant_images", config_file)
- self.__logger.info("Images needed for vEPC: %s", self.images)
- self.keystone_client = os_utils.get_keystone_client()
- self.glance_client = os_utils.get_glance_client()
- self.neutron_client = os_utils.get_neutron_client()
- self.nova_client = os_utils.get_nova_client()
- self.sec_group_id = None
- self.public_auth_url = None
- self.creds = None
- self.filename = None
-
- def prepare(self):
- """Prepare testcase (Additional pre-configuration steps)."""
- self.__logger.debug("OS Credentials: %s", os_utils.get_credentials())
-
- super(JujuEpc, self).prepare()
- self.__logger.info("Additional pre-configuration steps")
- self.public_auth_url = keystone_utils.get_endpoint(
- self.snaps_creds, 'identity')
- # it enforces a versioned public identity endpoint as juju simply
- # adds /auth/tokens wich fails vs an unversioned endpoint.
- if not self.public_auth_url.endswith(('v3', 'v3/', 'v2.0', 'v2.0/')):
- self.public_auth_url = urljoin(self.public_auth_url, 'v3')
-
- self.creds = {
- "tenant": self.tenant_name,
- "username": self.tenant_name,
- "password": self.tenant_name,
- "auth_url": os_utils.get_credentials()['auth_url']
- }
+ self.res_dir = os.path.join(
+ getattr(config.CONF, 'dir_results'), self.case_name)
+ try:
+ self.public_auth_url = self.get_public_auth_url(self.orig_cloud)
+ if not self.public_auth_url.endswith(('v3', 'v3/')):
+ self.public_auth_url = f"{self.public_auth_url}/v3"
+ except Exception: # pylint: disable=broad-except
+ self.public_auth_url = None
+ self.sec = None
+ self.image_alt = None
+ self.flavor_alt = None
+
+ def _install_juju(self):
+ (_, stdout, stderr) = self.ssh.exec_command(
+ 'sudo snap install juju --channel=2.3/stable --classic')
+ self.__logger.debug("stdout:\n%s", stdout.read().decode("utf-8"))
+ self.__logger.debug("stderr:\n%s", stderr.read().decode("utf-8"))
+ return not stdout.channel.recv_exit_status()
+
+ def _install_juju_wait(self):
+ (_, stdout, stderr) = self.ssh.exec_command(
+ 'sudo apt-get update && sudo apt-get install python3-pip -y && '
+ 'sudo pip3 install juju_wait===2.6.4')
+ self.__logger.debug("stdout:\n%s", stdout.read().decode("utf-8"))
+ self.__logger.debug("stderr:\n%s", stderr.read().decode("utf-8"))
+ return not stdout.channel.recv_exit_status()
+
+ def _register_cloud(self):
+ assert self.public_auth_url
+ self.__logger.info("Creating Cloud for Abot-epc .....")
+ clouds_yaml = os.path.join(self.res_dir, "clouds.yaml")
cloud_data = {
'url': self.public_auth_url,
- 'pass': self.tenant_name,
- 'tenant_n': self.tenant_name,
- 'user_n': self.tenant_name,
- 'region': os.environ.get(
- "OS_REGION_NAME", self.default_region_name)
- }
- self.__logger.info("Cloud DATA: %s", cloud_data)
- self.filename = os.path.join(self.case_dir, 'abot-epc.yaml')
- self.__logger.info("Create %s to add cloud info", self.filename)
- write_config(self.filename, CLOUD_TEMPLATE, **cloud_data)
-
- if self.snaps_creds.identity_api_version == 3:
- append_config(
- self.filename, '{}'.format(
- os_utils.get_credentials()['project_domain_name']),
- '{}'.format(os_utils.get_credentials()['user_domain_name']))
-
- self.__logger.info("Upload some OS images if it doesn't exist")
- for image_name, image_file in self.images.iteritems():
- self.__logger.info("image: %s, file: %s", image_name, image_file)
- if image_file and image_name:
- image_creator = OpenStackImage(
- self.snaps_creds,
- ImageConfig(name=image_name,
- image_user='cloud',
- img_format='qcow2',
- image_file=image_file))
- image_creator.create()
- self.created_object.append(image_creator)
+ 'region': self.cloud.region_name if self.cloud.region_name else (
+ 'RegionOne')}
+ with open(clouds_yaml, 'w', encoding='utf-8') as yfile:
+ yfile.write(CLOUD_TEMPLATE.format(**cloud_data))
+ scpc = scp.SCPClient(self.ssh.get_transport())
+ scpc.put(clouds_yaml, remote_path='~/')
+ (_, stdout, stderr) = self.ssh.exec_command(
+ '/snap/bin/juju add-cloud abot-epc -f clouds.yaml --replace')
+ self.__logger.debug("stdout:\n%s", stdout.read().decode("utf-8"))
+ self.__logger.debug("stderr:\n%s", stderr.read().decode("utf-8"))
+ return not stdout.channel.recv_exit_status()
+
+ def _register_credentials(self):
+ self.__logger.info("Creating Credentials for Abot-epc .....")
+ credentials_yaml = os.path.join(self.res_dir, "credentials.yaml")
+ creds_data = {
+ 'pass': self.project.password,
+ 'tenant_n': self.project.project.name,
+ 'user_n': self.project.user.name,
+ 'project_domain_n': self.cloud.auth.get(
+ "project_domain_name", "Default"),
+ 'user_domain_n': self.cloud.auth.get(
+ "user_domain_name", "Default")}
+ with open(credentials_yaml, 'w', encoding='utf-8') as yfile:
+ yfile.write(CREDS_TEMPLATE.format(**creds_data))
+ scpc = scp.SCPClient(self.ssh.get_transport())
+ scpc.put(credentials_yaml, remote_path='~/')
+ (_, stdout, stderr) = self.ssh.exec_command(
+ '/snap/bin/juju add-credential abot-epc -f credentials.yaml '
+ ' --replace --debug')
+ self.__logger.debug("stdout:\n%s", stdout.read().decode("utf-8"))
+ self.__logger.debug("stderr:\n%s", stderr.read().decode("utf-8"))
+ return not stdout.channel.recv_exit_status()
+
+ def _publish_image(self):
+ region_name = self.cloud.region_name if self.cloud.region_name else (
+ 'RegionOne')
+ (_, stdout, stderr) = self.ssh.exec_command(
+ '/snap/bin/juju metadata generate-image -d /home/ubuntu '
+ f'-i {self.image.id} -s xenial -r {region_name} '
+ f'-u {self.public_auth_url}')
+ self.__logger.debug("stdout:\n%s", stdout.read().decode("utf-8"))
+ self.__logger.debug("stderr:\n%s", stderr.read().decode("utf-8"))
+ return not stdout.channel.recv_exit_status()
+
+ def publish_image_alt(self, name=None):
+ image_alt = super().publish_image_alt(name)
+ region_name = self.cloud.region_name if self.cloud.region_name else (
+ 'RegionOne')
+ (_, stdout, stderr) = self.ssh.exec_command(
+ '/snap/bin/juju metadata generate-image -d /home/ubuntu '
+ f'-i {image_alt.id} -s trusty -r {region_name} '
+ f'-u {self.public_auth_url}')
+ self.__logger.debug("stdout:\n%s", stdout.read().decode("utf-8"))
+ self.__logger.debug("stderr:\n%s", stderr.read().decode("utf-8"))
+ return image_alt
def deploy_orchestrator(self): # pylint: disable=too-many-locals
"""
Bootstrap juju
"""
- self.__logger.info("Deployed Orchestrator")
- private_net_name = CONST.__getattribute__(
- 'vnf_{}_private_net_name'.format(self.case_name))
- private_subnet_name = CONST.__getattribute__(
- 'vnf_{}_private_subnet_name'.format(self.case_name))
- private_subnet_cidr = CONST.__getattribute__(
- 'vnf_{}_private_subnet_cidr'.format(self.case_name))
- abot_router = CONST.__getattribute__(
- 'vnf_{}_external_router'.format(self.case_name))
- dns_nameserver = CONST.__getattribute__(
- 'vnf_{}_dns_nameserver'.format(self.case_name))
- ext_net_name = CONST.__getattribute__(
- 'vnf_{}_external_network_name'.format(self.case_name))
-
- self.__logger.info("Creating full network ...")
- subnet_settings = SubnetConfig(name=private_subnet_name,
- cidr=private_subnet_cidr,
- dns_nameservers=dns_nameserver)
- network_settings = NetworkConfig(name=private_net_name,
- 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)
- self.__logger.info("Creating network Router ....")
- router_creator = OpenStackRouter(
- self.snaps_creds,
- RouterConfig(
- name=abot_router,
- external_gateway=ext_net_name,
- internal_subnets=[subnet_settings.name]))
- router_creator.create()
- self.created_object.append(router_creator)
- self.__logger.info("Creating Flavor ....")
- flavor_settings = FlavorConfig(
- name=self.orchestrator['requirements']['flavor']['name'],
- ram=self.orchestrator['requirements']['flavor']['ram_min'],
- disk=10,
- vcpus=1)
- flavor_creator = OpenStackFlavor(self.snaps_creds, flavor_settings)
- self.__logger.info("Juju Bootstrap: Skip creation of flavors")
- flavor_creator.create()
- self.created_object.append(flavor_creator)
- self.__logger.info("Creating Cloud for Abot-epc .....")
- os.system('juju add-cloud abot-epc -f {}'.format(self.filename))
- os.system('juju add-credential abot-epc -f {}'.format(self.filename))
- for image_name in self.images.keys():
- self.__logger.info("Generating Metadata for %s", image_name)
- image_id = os_utils.get_image_id(self.glance_client, image_name)
- os.system(
- 'juju metadata generate-image -d ~ -i {} -s {} -r '
- '{} -u {}'.format(
- image_id, image_name,
- os.environ.get("OS_REGION_NAME", self.default_region_name),
- self.public_auth_url))
- net_id = os_utils.get_network_id(self.neutron_client, private_net_name)
- self.__logger.info("Credential information : %s", net_id)
- juju_bootstrap_command = ('juju bootstrap abot-epc abot-controller '
- '--config network={} --metadata-source ~ '
- '--config ssl-hostname-verification=false '
- '--constraints mem=2G --bootstrap-series '
- 'xenial '
- '--config use-floating-ip=true --debug'.
- format(net_id))
- os.system(juju_bootstrap_command)
+ self._publish_image()
+ self.image_alt = self.publish_image_alt()
+ self.flavor_alt = self.create_flavor_alt()
+ self.__logger.info("Starting Juju Bootstrap process...")
+ region_name = self.cloud.region_name if self.cloud.region_name else (
+ 'RegionOne')
+ (_, stdout, stderr) = self.ssh.exec_command(
+ f'timeout {JujuEpc.juju_timeout} '
+ f'/snap/bin/juju bootstrap abot-epc/{region_name} abot-controller '
+ '--agent-version 2.3.9 --metadata-source /home/ubuntu '
+ '--constraints mem=2G --bootstrap-series xenial '
+ f'--config network={self.network.id} '
+ '--config ssl-hostname-verification=false '
+ f'--config external-network={self.ext_net.id} '
+ '--config use-floating-ip=true '
+ '--config use-default-secgroup=true '
+ '--debug')
+ self.__logger.debug("stdout:\n%s", stdout.read().decode("utf-8"))
+ self.__logger.debug("stderr:\n%s", stderr.read().decode("utf-8"))
+ return not stdout.channel.recv_exit_status()
+
+ def check_app(self, name='abot-epc-basic', status='active'):
+ """Check application status."""
+ for i in range(10):
+ (_, stdout, stderr) = self.ssh.exec_command(
+ f'/snap/bin/juju status --format short {name}')
+ output = stdout.read().decode("utf-8")
+ self.__logger.debug("stdout:\n%s", output)
+ self.__logger.debug("stderr:\n%s", stderr.read().decode("utf-8"))
+ if stdout.channel.recv_exit_status():
+ continue
+ ret = re.search(
+ rf'(?=workload:({status})\))', output)
+ if ret:
+ self.__logger.info("%s workload is %s", name, status)
+ break
+ self.__logger.info(
+ "loop %d: %s workload differs from %s", i + 1, name, status)
+ time.sleep(60)
+ else:
+ self.__logger.error("%s workload differs from %s", name, status)
+ return False
return True
def deploy_vnf(self):
"""Deploy ABOT-OAI-EPC."""
self.__logger.info("Upload VNFD")
- descriptor = self.vnf['descriptor']
- self.__logger.info("Get or create flavor for all Abot-EPC")
- flavor_settings = FlavorConfig(
- name=self.vnf['requirements']['flavor']['name'],
- ram=self.vnf['requirements']['flavor']['ram_min'],
- disk=10,
- vcpus=1)
- flavor_creator = OpenStackFlavor(self.snaps_creds, flavor_settings)
- flavor_creator.create()
- self.created_object.append(flavor_creator)
+ scpc = scp.SCPClient(self.ssh.get_transport())
+ scpc.put(
+ '/src/epc-requirements/abot_charm', remote_path='~/',
+ recursive=True)
self.__logger.info("Deploying Abot-epc bundle file ...")
- os.system('juju deploy {}'.format('/' + descriptor.get('file_name')))
- self.__logger.info("Waiting for instances .....")
- status = os.system('juju-wait')
- self.__logger.info("juju wait completed: %s", status)
- self.__logger.info("Deployed Abot-epc on Openstack")
- if status == 0:
- instances = os_utils.get_instances(self.nova_client)
- for items in instances:
- metadata = get_instance_metadata(self.nova_client, items)
- if 'juju-units-deployed' in metadata:
- sec_group = ('juju-' + metadata['juju-controller-uuid'] +
- '-' + metadata['juju-model-uuid'])
- self.sec_group_id = os_utils.get_security_group_id(
- self.neutron_client, sec_group)
- break
- self.__logger.info("Adding Security group rule....")
- os_utils.create_secgroup_rule(self.neutron_client,
- self.sec_group_id, 'ingress', 132)
- self.__logger.info("Copying the feature files to Abot_node ")
- os.system('juju scp -- -r {}/featureFiles abot-'
- 'epc-basic/0:~/'.format(self.case_dir))
- self.__logger.info("Copying the feature files in Abot_node ")
- os.system("juju ssh abot-epc-basic/0 'sudo rsync -azvv "
- "~/featureFiles /etc/rebaca-test-suite"
- "/featureFiles'")
- count = 0
- while count < 10:
- epcstatus = os.system('juju status oai-epc | '
- 'grep {} | grep {} | grep {}'
- .format('EPC', 'is', 'running'))
- if epcstatus == 0:
- break
- else:
- time.sleep(60)
- count = count + 1
- os.system('juju-wait')
- return True
- return False
+ (_, stdout, stderr) = self.ssh.exec_command(
+ 'sudo mkdir -p /src/epc-requirements && '
+ 'sudo mv abot_charm /src/epc-requirements/abot_charm && '
+ '/snap/bin/juju deploy '
+ '/src/epc-requirements/abot_charm/functest-abot-epc-bundle/'
+ 'bundle.yaml')
+ self.__logger.debug("stdout:\n%s", stdout.read().decode("utf-8"))
+ self.__logger.debug("stderr:\n%s", stderr.read().decode("utf-8"))
+ if stdout.channel.recv_exit_status():
+ return not stdout.channel.recv_exit_status()
+ (_, stdout, stderr) = self.ssh.exec_command(
+ 'PATH=/snap/bin/:$PATH '
+ f'timeout {JujuEpc.juju_timeout} juju-wait')
+ self.__logger.debug("stdout:\n%s", stdout.read().decode("utf-8"))
+ self.__logger.debug("stderr:\n%s", stderr.read().decode("utf-8"))
+ if stdout.channel.recv_exit_status():
+ return not stdout.channel.recv_exit_status()
+ self.__logger.info("Checking status of ABot and EPC units ...")
+ (_, stdout, stderr) = self.ssh.exec_command('/snap/bin/juju status')
+ output = stdout.read().decode("utf-8")
+ self.__logger.debug("stdout:\n%s", output)
+ self.__logger.debug("stderr:\n%s", stderr.read().decode("utf-8"))
+ if stdout.channel.recv_exit_status():
+ return not stdout.channel.recv_exit_status()
+ for app in ['abot-epc-basic', 'oai-epc', 'oai-hss']:
+ if not self.check_app(app):
+ return False
+ scpc = scp.SCPClient(self.ssh.get_transport())
+ scpc.put(
+ f'{self.case_dir}/featureFiles', remote_path='~/',
+ recursive=True)
+ (_, stdout, stderr) = self.ssh.exec_command(
+ f'timeout {JujuEpc.juju_timeout} /snap/bin/juju scp -- -r -v '
+ '~/featureFiles abot-epc-basic/0:/etc/rebaca-test-suite/')
+ output = stdout.read().decode("utf-8")
+ self.__logger.debug("stdout:\n%s", output)
+ self.__logger.debug("stderr:\n%s", stderr.read().decode("utf-8"))
+ return not stdout.channel.recv_exit_status()
def test_vnf(self):
"""Run test on ABoT."""
start_time = time.time()
- self.__logger.info("Running VNF Test cases....")
- os.system('juju run-action abot-epc-basic/0 run '
- 'tagnames={}'.format(self.details['test_vnf']['tag_name']))
- os.system('juju-wait')
+ (_, stdout, stderr) = self.ssh.exec_command(
+ "/snap/bin/juju run-action abot-epc-basic/0 "
+ f"run tagnames={self.details['test_vnf']['tag_name']}")
+ self.__logger.debug("stdout:\n%s", stdout.read().decode("utf-8"))
+ self.__logger.debug("stderr:\n%s", stderr.read().decode("utf-8"))
+ if stdout.channel.recv_exit_status():
+ return not stdout.channel.recv_exit_status()
+ (_, stdout, stderr) = self.ssh.exec_command(
+ 'PATH=/snap/bin/:$PATH '
+ f'timeout {JujuEpc.juju_timeout} juju-wait')
+ self.__logger.debug("stdout:\n%s", stdout.read().decode("utf-8"))
+ self.__logger.debug("stderr:\n%s", stderr.read().decode("utf-8"))
+ if stdout.channel.recv_exit_status():
+ return not stdout.channel.recv_exit_status()
duration = time.time() - start_time
self.__logger.info("Getting results from Abot node....")
- os.system('juju scp abot-epc-basic/0:/var/lib/abot-'
- 'epc-basic/artifacts/TestResults.json {}/.'
- .format(self.case_dir))
+ (_, stdout, stderr) = self.ssh.exec_command(
+ f'timeout {JujuEpc.juju_timeout} /snap/bin/juju scp '
+ '-- -v abot-epc-basic/0:'
+ '/var/lib/abot-epc-basic/artifacts/TestResults.json .')
+ self.__logger.debug("stdout:\n%s", stdout.read().decode("utf-8"))
+ self.__logger.debug("stderr:\n%s", stderr.read().decode("utf-8"))
+ if stdout.channel.recv_exit_status():
+ return not stdout.channel.recv_exit_status()
+ scpc = scp.SCPClient(self.ssh.get_transport())
+ scpc.get('TestResults.json', self.res_dir)
self.__logger.info("Parsing the Test results...")
- res = (process_abot_test_result('{}/TestResults.'
- 'json'.format(self.case_dir)))
+ res = process_abot_test_result(f'{self.res_dir}/TestResults.json')
short_result = sig_test_format(res)
self.__logger.info(short_result)
- self.details['test_vnf'].update(status='PASS',
- result=short_result,
- full_result=res,
- duration=duration)
-
- self.__logger.info("Test VNF result: Passed: %d, Failed:"
- "%d, Skipped: %d", short_result['passed'],
- short_result['failures'], short_result['skipped'])
+ self.details['test_vnf'].update(
+ status='PASS', result=short_result, full_result=res,
+ duration=duration)
+ self.__logger.info(
+ "Test VNF result: Passed: %d, Failed:%d, Skipped: %d",
+ short_result['passed'],
+ short_result['failures'], short_result['skipped'])
return True
- def clean(self):
- """Clean created objects/functions."""
+ def execute(self):
+ """Prepare testcase (Additional pre-configuration steps)."""
+ assert self.public_auth_url
+ self.__logger.info("Additional pre-configuration steps")
try:
- if not self.orchestrator['requirements']['preserve_setup']:
- self.__logger.info("Removing deployment files...")
- testresult = os.path.join(self.case_dir, 'TestResults.json')
- if os.path.exists(testresult):
- os.remove(testresult)
- self.__logger.info("Removing %s file ", self.filename)
- if os.path.exists(self.filename):
- os.remove(self.filename)
- self.__logger.info("Destroying Orchestrator...")
- os.system('juju destroy-controller -y abot-controller '
- '--destroy-all-models')
+ os.makedirs(self.res_dir)
+ except OSError as ex:
+ if ex.errno != errno.EEXIST:
+ self.__logger.exception("Cannot create %s", self.res_dir)
+ raise Exception from ex
+ self.__logger.info("ENV:\n%s", env.string())
+ try:
+ assert self._install_juju()
+ assert self._install_juju_wait()
+ assert self._register_cloud()
+ assert self._register_credentials()
+ assert self.deploy_orchestrator()
+ assert self.deploy_vnf()
+ assert self.test_vnf()
except Exception: # pylint: disable=broad-except
- self.__logger.warn("Some issue during the undeployment ..")
- self.__logger.warn("Tenant clean continue ..")
-
- if not self.orchestrator['requirements']['preserve_setup']:
- self.__logger.info('Remove the Abot_epc 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)
-
- self.__logger.info("Releasing all the floating IPs")
- floating_ips = os_utils.get_floating_ips(self.neutron_client)
- tenant_id = os_utils.get_tenant_id(self.keystone_client,
- self.tenant_name)
- self.__logger.info("TENANT ID : %s", tenant_id)
- for item in floating_ips:
- if item['tenant_id'] == tenant_id:
- os_utils.delete_floating_ip(self.neutron_client,
- item['id'])
- self.__logger.info("Cleaning Projects and Users")
- 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)
- return True
+ self.__logger.exception("juju_epc failed")
+ return 1
+ return 0
-
-# ----------------------------------------------------------
-#
-# YAML UTILS
-#
-# -----------------------------------------------------------
-def get_config(parameter, file_path):
- """
- Returns the value of a given parameter in file.yaml
- parameter must be given in string format with dots
- Example: general.openstack.image_name
- """
- with open(file_path) as config_file:
- file_yaml = yaml.safe_load(config_file)
- config_file.close()
- value = file_yaml
- for element in parameter.split("."):
- value = value.get(element)
- if value is None:
- raise ValueError("The parameter %s is not defined in"
- " reporting.yaml" % parameter)
- return value
+ def clean(self):
+ """Clean created objects/functions."""
+ (_, stdout, stderr) = self.ssh.exec_command(
+ '/snap/bin/juju debug-log --replay --no-tail')
+ self.__logger.debug("stdout:\n%s", stdout.read().decode("utf-8"))
+ self.__logger.debug("stderr:\n%s", stderr.read().decode("utf-8"))
+ (_, stdout, stderr) = self.ssh.exec_command(
+ '/snap/bin/juju destroy-controller -y abot-controller '
+ '--destroy-all-models')
+ self.__logger.debug("stdout:\n%s", stdout.read().decode("utf-8"))
+ self.__logger.debug("stderr:\n%s", stderr.read().decode("utf-8"))
+ for fip in self.cloud.list_floating_ips():
+ self.cloud.delete_floating_ip(fip.id)
+ if self.image_alt:
+ self.cloud.delete_image(self.image_alt)
+ if self.flavor_alt:
+ self.orig_cloud.delete_flavor(self.flavor_alt.id)
+ super().clean()
def sig_test_format(sig_test):
def process_abot_test_result(file_path):
""" Process ABoT Result """
- with open(file_path) as test_result:
+ with open(file_path, encoding='utf-8') as test_result:
data = json.load(test_result)
res = []
for tests in data:
for steps in flatten_steps:
steps['result'] = steps['step_status']
res.append(steps)
- except:
+ except Exception: # pylint: disable=broad-except
logging.error("Could not post data to ElasticSearch host")
raise
return res
raise
return obj
-
-
-def get_instance_metadata(nova_client, instance):
- """ Get instance Metadata - Instance ID """
- try:
- instance = nova_client.servers.get(instance.id)
- return instance.metadata
- except Exception as exc: # pylint: disable=broad-except
- logging.error("Error [get_instance_status(nova_client)]: %s", exc)
- return None
-
-
-CLOUD_TEMPLATE = """clouds:
- abot-epc:
- type: openstack
- auth-types: [userpass]
- endpoint: {url}
- regions:
- {region}:
- endpoint: {url}
-credentials:
- abot-epc:
- abot-epc:
- auth-type: userpass
- password: {pass}
- tenant-name: {tenant_n}
- username: {user_n}"""
-
-
-def write_config(fname, template, **kwargs):
- """ Generate yaml from template for addinh cloud in juju """
- with open(fname, 'w') as yfile:
- yfile.write(template.format(**kwargs))
-
-
-def append_config(file_name, p_domain, u_domain):
- """ Append values into a yaml file """
- with open(file_name) as yfile:
- doc = yaml.load(yfile)
- doc['credentials']['abot-epc']['abot-epc']['project-domain-name'] = (
- p_domain)
- doc['credentials']['abot-epc']['abot-epc']['user-domain-name'] = (
- u_domain)
-
- with open(file_name, 'w') as yfile:
- yaml.safe_dump(doc, yfile, default_flow_style=False)