import uuid
from copy import deepcopy
from urlparse import urljoin
+import pkg_resources
+import yaml
from functest.core import 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 functest.utils import config
-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.config.security_group import (
+ Direction, Protocol, SecurityGroupConfig, SecurityGroupRuleConfig)
from snaps.config.user import UserConfig
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.create_security_group import OpenStackSecurityGroup
from snaps.openstack.create_user import OpenStackUser
from snaps.openstack.utils import keystone_utils
-from snaps.openstack.utils import neutron_utils
from snaps.openstack.utils import nova_utils
-import yaml
__author__ = "Amarendra Meher <amarendra@rebaca.com>"
__author__ = "Soumaya K Nayek <soumaya.nayek@rebaca.com>"
__logger = logging.getLogger(__name__)
- default_region_name = "RegionOne"
-
def __init__(self, **kwargs):
if "case_name" not in kwargs:
kwargs["case_name"] = "juju_epc"
'functest', 'opnfv_tests/vnf/epc')
try:
self.config = getattr(
- CONST, 'vnf_{}_config'.format(self.case_name))
+ config.CONF, 'vnf_{}_config'.format(self.case_name))
except Exception:
raise Exception("VNF config file not found")
self.config_file = os.path.join(self.case_dir, self.config)
self.public_auth_url = None
self.res_dir = os.path.join(
- getattr(CONST, 'dir_results'), self.case_name)
+ getattr(config.CONF, 'dir_results'), self.case_name)
def _bypass_juju_network_discovery_bug(self, name):
user_creator = OpenStackUser(
clouds_yaml = os.path.join(self.res_dir, "clouds.yaml")
cloud_data = {
'url': self.public_auth_url,
- 'region': os.environ.get(
- "OS_REGION_NAME", self.default_region_name)}
+ 'region': self.snaps_creds.region_name}
with open(clouds_yaml, 'w') as yfile:
yfile.write(CLOUD_TEMPLATE.format(**cloud_data))
if os.system(
credentials_yaml)):
raise vnf.VnfPreparationException
+ def _add_custom_rule(self, sec_grp_name):
+ """ To add custom rule for SCTP Traffic """
+ sec_grp_rules = list()
+ sec_grp_rules.append(
+ SecurityGroupRuleConfig(
+ sec_grp_name=sec_grp_name, direction=Direction.ingress,
+ protocol=Protocol.sctp))
+ security_group = OpenStackSecurityGroup(
+ self.snaps_creds,
+ SecurityGroupConfig(
+ name=sec_grp_name,
+ rule_settings=sec_grp_rules))
+ security_group.create()
+ self.created_object.append(security_group)
+
def prepare(self):
"""Prepare testcase (Additional pre-configuration steps)."""
self.__logger.info("Additional pre-configuration steps")
"""
self.__logger.info("Deployed Orchestrator")
private_net_name = getattr(
- CONST, 'vnf_{}_private_net_name'.format(self.case_name))
+ config.CONF, 'vnf_{}_private_net_name'.format(self.case_name))
private_subnet_name = getattr(
- CONST, 'vnf_{}_private_subnet_name'.format(self.case_name))
+ config.CONF, 'vnf_{}_private_subnet_name'.format(self.case_name))
private_subnet_cidr = getattr(
- CONST, 'vnf_{}_private_subnet_cidr'.format(self.case_name))
+ config.CONF, 'vnf_{}_private_subnet_cidr'.format(self.case_name))
abot_router = getattr(
- CONST, 'vnf_{}_external_router'.format(self.case_name))
+ config.CONF, 'vnf_{}_external_router'.format(self.case_name))
self.__logger.info("Creating full network ...")
subnet_settings = SubnetConfig(
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),
+ image_id, image_name, self.snaps_creds.region_name,
self.public_auth_url))
self.created_object.append(image_creator)
- self.__logger.info("Credential information : %s", net_id)
+ self.__logger.info("Network ID : %s", net_id)
juju_bootstrap_command = (
'juju bootstrap abot-epc abot-controller --config network={} '
'--metadata-source ~ --config ssl-hostname-verification=false '
self.__logger.info("juju wait completed: %s", status)
self.__logger.info("Deployed Abot-epc on Openstack")
nova_client = nova_utils.nova_client(self.snaps_creds)
- neutron_client = neutron_utils.neutron_client(self.snaps_creds)
if status == 0:
- instances = os_utils.get_instances(nova_client)
+ instances = get_instances(nova_client)
+ self.__logger.info("List of Instance: %s", instances)
for items in instances:
metadata = get_instance_metadata(nova_client, items)
if 'juju-units-deployed' in metadata:
- sec_group = ('juju-' + metadata['juju-controller-uuid'] +
+ sec_group = ('juju-' +
+ metadata['juju-controller-uuid'] +
'-' + metadata['juju-model-uuid'])
- self.sec_group_id = os_utils.get_security_group_id(
- neutron_client, sec_group)
+ self.__logger.info("Instance: %s", sec_group)
break
self.__logger.info("Adding Security group rule....")
- os_utils.create_secgroup_rule(
- neutron_client, self.sec_group_id, 'ingress', 132)
+ # This will add sctp rule to a common Security Group Created
+ # by juju and shared to all deployed units.
+ self._add_custom_rule(sec_group)
self.__logger.info("Copying the feature files to Abot_node ")
os.system('juju scp -- -r {}/featureFiles abot-'
'epc-basic/0:~/'.format(self.case_dir))
return obj
+def get_instances(nova_client):
+ """ To get all vm info of a project """
+ try:
+ instances = nova_client.servers.list()
+ return instances
+ except Exception as exc: # pylint: disable=broad-except
+ logging.error("Error [get_instances(nova_client)]: %s", exc)
+ return None
+
+
def get_instance_metadata(nova_client, instance):
""" Get instance Metadata - Instance ID """
try:
Code Review / functest.git / blobdiff