diff options
| -rw-r--r-- | functest/ci/config_aarch64_patch.yaml | 14 | ||||
| -rw-r--r-- | functest/ci/testcases.yaml | 4 | ||||
| -rw-r--r-- | functest/opnfv_tests/openstack/vping/ping.sh | 19 | ||||
| -rw-r--r-- | functest/opnfv_tests/vnf/ims/orchestra_ims.py | 873 | ||||
| -rw-r--r-- | functest/opnfv_tests/vnf/ims/orchestra_ims.yaml | 38 | ||||
| -rw-r--r-- | functest/tests/unit/vnf/ims/test_orchestra_ims.py | 224 |
6 files changed, 814 insertions, 358 deletions
diff --git a/functest/ci/config_aarch64_patch.yaml b/functest/ci/config_aarch64_patch.yaml index 27b270626..de82dbc70 100644 --- a/functest/ci/config_aarch64_patch.yaml +++ b/functest/ci/config_aarch64_patch.yaml @@ -8,18 +8,16 @@ os: images: glance_tests: disk_file: /home/opnfv/functest/images/cirros-d161201-aarch64-disk.img - kernel_file: /home/opnfv/functest/images/cirros-d161201-aarch64-kernel - ramdisk_file: /home/opnfv/functest/images/cirros-d161201-aarch64-initramfs extra_properties: - os_command_line: root=/dev/vdb1 rw rootwait console=tty0 console=ttyS0 console=ttyAMA0 - hw_video_model: vga + hw_firmware_type: 'uefi' + short_id: 'ubuntu16.04' + hw_video_model: 'vga' cirros: disk_file: /home/opnfv/functest/images/cirros-d161201-aarch64-disk.img - kernel_file: /home/opnfv/functest/images/cirros-d161201-aarch64-kernel - ramdisk_file: /home/opnfv/functest/images/cirros-d161201-aarch64-initramfs extra_properties: - os_command_line: root=/dev/vdb1 rw rootwait console=tty0 console=ttyS0 console=ttyAMA0 - hw_video_model: vga + hw_firmware_type: 'uefi' + short_id: 'ubuntu16.04' + hw_video_model: 'vga' vping: image_name: TestVM diff --git a/functest/ci/testcases.yaml b/functest/ci/testcases.yaml index c6aa63623..3dcf9b13e 100644 --- a/functest/ci/testcases.yaml +++ b/functest/ci/testcases.yaml @@ -543,7 +543,7 @@ tiers: - case_name: orchestra_ims - enabled: false + enabled: true project_name: functest criteria: 100 blocking: false @@ -551,7 +551,7 @@ tiers: VNF deployment with OpenBaton (Orchestra) dependencies: installer: '' - scenario: '' + scenario: 'os-nosdn-nofeature-ha' run: module: 'functest.opnfv_tests.vnf.ims.orchestra_ims' class: 'ImsVnf' diff --git a/functest/opnfv_tests/openstack/vping/ping.sh b/functest/opnfv_tests/openstack/vping/ping.sh index 693b86825..15f5e84e1 100644 --- a/functest/opnfv_tests/openstack/vping/ping.sh +++ b/functest/opnfv_tests/openstack/vping/ping.sh @@ -1,13 +1,10 @@ #!/bin/sh -while true; do - ping -c 1 $1 2>&1 >/dev/null - RES=$? - if [ "Z$RES" = "Z0" ] ; then - echo 'vPing OK' - break - else - echo 'vPing KO' - fi - sleep 1 -done
\ No newline at end of file + +ping -c 1 $1 2>&1 >/dev/null +RES=$? +if [ "Z$RES" = "Z0" ] ; then + echo 'vPing OK' +else + echo 'vPing KO' +fi diff --git a/functest/opnfv_tests/vnf/ims/orchestra_ims.py b/functest/opnfv_tests/vnf/ims/orchestra_ims.py index 7b1ea9ad8..0a0a766a3 100644 --- a/functest/opnfv_tests/vnf/ims/orchestra_ims.py +++ b/functest/opnfv_tests/vnf/ims/orchestra_ims.py @@ -7,21 +7,42 @@ # which accompanies this distribution, and is available at # http://www.apache.org/licenses/LICENSE-2.0 +"""Orchestra OpenIMS testcase implementation.""" + import json import logging import os -import pkg_resources import socket -import sys import time +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.os_credentials import OSCreds +from snaps.openstack.create_instance import ( + VmInstanceSettings, OpenStackVmInstance) +from functest.opnfv_tests.openstack.snaps import snaps_utils + import functest.core.vnf as vnf import functest.utils.openstack_utils as os_utils from functest.utils.constants import CONST -from org.openbaton.cli.agents.agents import MainAgent from org.openbaton.cli.errors.errors import NfvoException +from org.openbaton.cli.agents.agents import MainAgent __author__ = "Pauls, Michael <michael.pauls@fokus.fraunhofer.de>" @@ -32,404 +53,570 @@ __author__ = "Pauls, Michael <michael.pauls@fokus.fraunhofer.de>" # ----------------------------------------------------------- -def get_config(parameter, my_file): +def get_config(parameter, file_path): """ + Get config parameter. + 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) as f: - file_yaml = yaml.safe_load(f) - f.close() + 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" - " %s" % (parameter, my_file)) + " reporting.yaml", parameter) return value def servertest(host, port): + """Method to test that a server is reachable at IP:port""" args = socket.getaddrinfo(host, port, socket.AF_INET, socket.SOCK_STREAM) for family, socktype, proto, canonname, sockaddr in args: - s = socket.socket(family, socktype, proto) + sock = socket.socket(family, socktype, proto) try: - s.connect(sockaddr) + sock.connect(sockaddr) except socket.error: return False else: - s.close() + sock.close() return True +def get_userdata(orchestrator=dict): + """Build userdata for Open Baton machine""" + userdata = "#!/bin/bash\n" + userdata += "echo \"Executing userdata...\"\n" + userdata += "set -x\n" + userdata += "set -e\n" + userdata += "echo \"Set nameserver to '8.8.8.8'...\"\n" + userdata += "echo \"nameserver 8.8.8.8\" >> /etc/resolv.conf\n" + userdata += "echo \"Install curl...\"\n" + userdata += "apt-get install curl\n" + userdata += "echo \"Inject public key...\"\n" + userdata += ("echo \"ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCuPXrV3" + "geeHc6QUdyUr/1Z+yQiqLcOskiEGBiXr4z76MK4abiFmDZ18OMQlc" + "fl0p3kS0WynVgyaOHwZkgy/DIoIplONVr2CKBKHtPK+Qcme2PVnCtv" + "EqItl/FcD+1h5XSQGoa+A1TSGgCod/DPo+pes0piLVXP8Ph6QS1k7S" + "ic7JDeRQ4oT1bXYpJ2eWBDMfxIWKZqcZRiGPgMIbJ1iEkxbpeaAd9O" + "4MiM9nGCPESmed+p54uYFjwEDlAJZShcAZziiZYAvMZhvAhe6USljc" + "7YAdalAnyD/jwCHuwIrUw/lxo7UdNCmaUxeobEYyyFA1YVXzpNFZya" + "XPGAAYIJwEq/ openbaton@opnfv\" >> /home/ubuntu/.ssh/aut" + "horized_keys\n") + userdata += "echo \"Download bootstrap...\"\n" + userdata += ("curl -s %s " + "> ./bootstrap\n" % orchestrator['bootstrap']['url']) + userdata += ("curl -s %s" "> ./config_file\n" % + orchestrator['bootstrap']['config']['url']) + userdata += ("echo \"Disable usage of mysql...\"\n") + userdata += "sed -i s/mysql=.*/mysql=no/g /config_file\n" + userdata += ("echo \"Setting 'rabbitmq_broker_ip' to '%s'\"\n" + % orchestrator['details']['fip'].ip) + userdata += ("sed -i s/rabbitmq_broker_ip=localhost/rabbitmq_broker_ip" + "=%s/g /config_file\n" % orchestrator['details']['fip'].ip) + userdata += "echo \"Set autostart of components to 'false'\"\n" + userdata += "export OPENBATON_COMPONENT_AUTOSTART=false\n" + userdata += "echo \"Execute bootstrap...\"\n" + bootstrap = "sh ./bootstrap release -configFile=./config_file" + userdata += bootstrap + "\n" + userdata += "echo \"Setting 'nfvo.plugin.timeout' to '300000'\"\n" + userdata += ("echo \"nfvo.plugin.timeout=600000\" >> " + "/etc/openbaton/openbaton-nfvo.properties\n") + userdata += ( + "wget %s -O /etc/openbaton/openbaton-vnfm-generic-user-data.sh\n" % + orchestrator['gvnfm']['userdata']['url']) + userdata += "sed -i '113i"'\ \ \ \ '"sleep 60' " \ + "/etc/openbaton/openbaton-vnfm-generic-user-data.sh\n" + userdata += "echo \"Starting NFVO\"\n" + userdata += "service openbaton-nfvo restart\n" + userdata += "echo \"Starting Generic VNFM\"\n" + userdata += "service openbaton-vnfm-generic restart\n" + userdata += "echo \"...end of userdata...\"\n" + return userdata + + class ImsVnf(vnf.VnfOnBoarding): """OpenIMS VNF deployed with openBaton orchestrator""" - def __init__(self, project='functest', case_name='orchestra_ims', - repo='', cmd=''): - super(ImsVnf, self).__init__(project, case_name, repo, cmd) - self.logger = logging.getLogger(__name__) - self.logger.info("Orchestra IMS VNF onboarding test starting") - self.ob_password = "openbaton" - self.ob_username = "admin" - self.ob_https = False - self.ob_port = "8080" - self.ob_ip = "localhost" - self.ob_instance_id = "" + # logger = logging.getLogger(__name__) + + def __init__(self, **kwargs): + if "case_name" not in kwargs: + kwargs["case_name"] = "orchestra_ims" + super(ImsVnf, self).__init__(**kwargs) + self.logger = logging.getLogger("functest.ci.run_tests.orchestra") + self.logger.info("kwargs %s", (kwargs)) + self.case_dir = pkg_resources.resource_filename( 'functest', 'opnfv_tests/vnf/ims/') self.data_dir = CONST.__getattribute__('dir_ims_data') self.test_dir = CONST.__getattribute__('dir_repo_vims_test') - self.ob_projectid = "" - self.keystone_client = os_utils.get_keystone_client() - self.ob_nsr_id = "" - self.nsr = None - self.main_agent = None - # vIMS Data directory creation - if not os.path.exists(self.data_dir): - os.makedirs(self.data_dir) - # Retrieve the configuration + self.created_resources = [] + self.logger.info("Orchestra IMS VNF onboarding test starting") + try: self.config = CONST.__getattribute__( 'vnf_{}_config'.format(self.case_name)) except BaseException: raise Exception("Orchestra VNF config file not found") config_file = self.case_dir + self.config - self.imagename = get_config("openbaton.imagename", config_file) - self.bootstrap_link = get_config("openbaton.bootstrap_link", - config_file) - self.bootstrap_config_link = get_config( - "openbaton.bootstrap_config_link", config_file) - self.market_link = get_config("openbaton.marketplace_link", - config_file) + + self.baton = dict( + requirements=get_config("orchestrator.requirements", config_file), + credentials=get_config("orchestrator.credentials", config_file), + bootstrap=get_config("orchestrator.bootstrap", config_file), + gvnfm=get_config("orchestrator.gvnfm", config_file), + ) + self.logger.debug("Orchestrator configuration %s", self.baton) + + self.details['orchestrator'] = dict( + name=get_config("orchestrator.name", config_file), + version=get_config("orchestrator.version", config_file), + status='ERROR', + result='' + ) + self.baton['details'] = {} + self.baton['details']['image'] = self.baton['requirements']['os_image'] + self.baton['details']['name'] = self.details['orchestrator']['name'] + + self.vnf = dict( + descriptor=get_config("vnf.descriptor", config_file), + requirements=get_config("vnf.requirements", config_file) + ) + self.details['vnf'] = dict( + name=get_config("vnf.name", config_file), + ) + self.logger.debug("VNF configuration: %s", self.vnf) + + self.details['test_vnf'] = dict( + name="openims-test", + ) + + # vIMS Data directory creation + if not os.path.exists(self.data_dir): + os.makedirs(self.data_dir) + self.images = get_config("tenant_images", config_file) self.ims_conf = get_config("vIMS", config_file) - self.userdata_file = get_config("openbaton.userdata.file", - config_file) + self.snaps_creds = None - def deploy_orchestrator(self): - self.logger.info("Additional pre-configuration steps") - nova_client = os_utils.get_nova_client() - neutron_client = os_utils.get_neutron_client() - glance_client = os_utils.get_glance_client() + def prepare(self): + """Prepare testscase (Additional pre-configuration steps).""" + super(ImsVnf, self).prepare() - # Import images if needed - # needs some images - self.logger.info("Upload some OS images if it doesn't exist") - temp_dir = os.path.join(self.data_dir, "tmp/") + self.logger.info("Additional pre-configuration steps") + self.logger.info("creds %s", (self.creds)) + + self.snaps_creds = OSCreds( + username=self.creds['username'], + password=self.creds['password'], + auth_url=self.creds['auth_url'], + project_name=self.creds['tenant'], + identity_api_version=int(os_utils.get_keystone_client_version())) + + self.prepare_images() + self.prepare_flavor() + self.prepare_security_groups() + self.prepare_network() + self.prepare_floating_ip() + + def prepare_images(self): + """Upload images if they doen't exist yet""" + self.logger.info("Upload images if they doen't exist yet") for image_name, image_url in self.images.iteritems(): self.logger.info("image: %s, url: %s", image_name, image_url) - try: - image_id = os_utils.get_image_id(glance_client, - image_name) - self.logger.info("image_id: %s", image_id) - except BaseException: - self.logger.error("Unexpected error: %s", sys.exc_info()[0]) - - if image_id == '': - self.logger.info("""%s image doesn't exist on glance - repository. Try downloading this image - and upload on glance !""" % image_name) - image_id = os_utils.download_and_add_image_on_glance( - glance_client, - image_name, - image_url, - temp_dir) - if image_id == '': - self.logger.error("Failed to find or upload required OS " - "image for this deployment") - return False + if image_url and image_name: + image = OpenStackImage( + self.snaps_creds, + ImageSettings(name=image_name, + image_user='cloud', + img_format='qcow2', + url=image_url)) + image.create() + # self.created_resources.append(image); + + def prepare_security_groups(self): + """Create Open Baton security group if it doesn't exist yet""" + self.logger.info( + "Creating security group for Open Baton if not yet existing...") + sg_rules = list() + sg_rules.append( + SecurityGroupRuleSettings( + 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( + 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( + 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( + 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( + # sec_grp_name="orchestra-sec-group-allowall", + # direction=Direction.ingress, + # protocol=Protocol.icmp, + # port_range_min=-1, + # port_range_max=-1)) + # sg_rules.append( + # SecurityGroupRuleSettings( + # sec_grp_name="orchestra-sec-group-allowall", + # direction=Direction.egress, + # protocol=Protocol.icmp, + # port_range_min=-1, + # port_range_max=-1)) + + security_group = OpenStackSecurityGroup( + self.snaps_creds, + SecurityGroupSettings( + name="orchestra-sec-group-allowall", + rule_settings=sg_rules)) + + security_group_info = security_group.create() + self.created_resources.append(security_group) + self.baton['details']['sec_group'] = security_group_info.name + self.logger.info( + "Security group orchestra-sec-group-allowall prepared") - network_dic = os_utils.create_network_full(neutron_client, - "openbaton_mgmt", - "openbaton_mgmt_subnet", - "openbaton_router", - "192.168.100.0/24") + def prepare_flavor(self): + """Create Open Baton flavor if it doesn't exist yet""" + self.logger.info( + "Create Flavor for Open Baton NFVO if not yet existing") + + flavor_settings = FlavorSettings( + name=self.baton['requirements']['flavor']['name'], + ram=self.baton['requirements']['flavor']['ram_min'], + disk=self.baton['requirements']['flavor']['disk'], + vcpus=self.baton['requirements']['flavor']['vcpus']) + flavor = OpenStackFlavor(self.snaps_creds, flavor_settings) + flavor_info = flavor.create() + self.created_resources.append(flavor) + self.baton['details']['flavor'] = {} + self.baton['details']['flavor']['name'] = flavor_settings.name + self.baton['details']['flavor']['id'] = flavor_info.id + + def prepare_network(self): + """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( + name='%s_subnet' % + self.baton['details']['name'], + cidr="192.168.100.0/24") + network_settings = NetworkSettings( + name='%s_net' % + self.baton['details']['name'], + subnet_settings=[subnet_settings]) + orchestra_network = OpenStackNetwork( + self.snaps_creds, network_settings) + orchestra_network_info = orchestra_network.create() + self.baton['details']['network'] = {} + self.baton['details']['network']['id'] = orchestra_network_info.id + self.baton['details']['network']['name'] = orchestra_network_info.name + self.baton['details']['external_net_name'] = \ + snaps_utils.get_ext_net_name(self.snaps_creds) + self.created_resources.append(orchestra_network) + orchestra_router = OpenStackRouter( + self.snaps_creds, + RouterSettings( + name='%s_router' % + self.baton['details']['name'], + external_gateway=self.baton['details']['external_net_name'], + internal_subnets=[ + subnet_settings.name])) + orchestra_router.create() + self.created_resources.append(orchestra_router) + self.logger.info("Created network and router for Open Baton NFVO...") + + def prepare_floating_ip(self): + """Select/Create Floating IP if it doesn't exist yet""" + self.logger.info("Retrieving floating IP for Open Baton NFVO") + neutron_client = snaps_utils.neutron_utils.neutron_client( + self.snaps_creds) + # Finding Tenant ID to check to which tenant the Floating IP belongs + tenant_id = os_utils.get_tenant_id( + os_utils.get_keystone_client(self.creds), + self.tenant_name) + # Use os_utils to retrieve complete information of Floating IPs + floating_ips = os_utils.get_floating_ips(neutron_client) + my_floating_ips = [] + # Filter Floating IPs with tenant id + for floating_ip in floating_ips: + # self.logger.info("Floating IP: %s", floating_ip) + if floating_ip.get('tenant_id') == tenant_id: + my_floating_ips.append(floating_ip.get('floating_ip_address')) + # Select if Floating IP exist else create new one + if len(my_floating_ips) >= 1: + # Get Floating IP object from snaps for clean up + snaps_floating_ips = snaps_utils.neutron_utils.get_floating_ips( + neutron_client) + for my_floating_ip in my_floating_ips: + for snaps_floating_ip in snaps_floating_ips: + if snaps_floating_ip.ip == my_floating_ip: + self.baton['details']['fip'] = snaps_floating_ip + self.logger.info( + "Selected floating IP for Open Baton NFVO %s", + (self.baton['details']['fip'].ip)) + break + if self.baton['details']['fip'] is not None: + break + else: + self.logger.info("Creating floating IP for Open Baton NFVO") + self.baton['details']['fip'] = \ + snaps_utils.neutron_utils.create_floating_ip( + neutron_client, self.baton['details']['external_net_name']) + self.logger.info( + "Created floating IP for Open Baton NFVO %s", + (self.baton['details']['fip'].ip)) - # orchestrator VM flavor + def get_vim_descriptor(self): + """"Create VIM descriptor to be used for onboarding""" self.logger.info( - "Check if orchestra Flavor is available, if not create one") - flavor_exist, flavor_id = os_utils.get_or_create_flavor( - "orchestra", - "4096", - '20', - '2', - public=True) - self.logger.debug("Flavor id: %s" % flavor_id) - - if not network_dic: - self.logger.error("There has been a problem when creating the " - "neutron network") - - network_id = network_dic["net_id"] - - self.logger.info("Creating floating IP for VM in advance...") - floatip_dic = os_utils.create_floating_ip(neutron_client) - floatip = floatip_dic['fip_addr'] - - if floatip is None: - self.logger.error("Cannot create floating IP.") - return False + "Building VIM descriptor with PoP creds: %s", + self.creds) + # Depending on API version either tenant ID or project name must be + # used + if os_utils.is_keystone_v3(): + self.logger.info( + "Using v3 API of OpenStack... -> Using OS_PROJECT_ID") + project_id = os_utils.get_tenant_id( + os_utils.get_keystone_client(), + self.creds.get("project_name")) + else: + self.logger.info( + "Using v2 API of OpenStack... -> Using OS_TENANT_NAME") + project_id = self.creds.get("tenant_name") + self.logger.debug("VIM project/tenant id: %s", project_id) + vim_json = { + "name": "vim-instance", + "authUrl": self.creds.get("auth_url"), + "tenant": project_id, + "username": self.creds.get("username"), + "password": self.creds.get("password"), + "securityGroups": [ + self.baton['details']['sec_group'] + ], + "type": "openstack", + "location": { + "name": "opnfv", + "latitude": "52.525876", + "longitude": "13.314400" + } + } + self.logger.info("Built VIM descriptor: %s", vim_json) + return vim_json - userdata = "#!/bin/bash\n" - userdata += "echo \"Executing userdata...\"\n" - userdata += "set -x\n" - userdata += "set -e\n" - userdata += "echo \"Set nameserver to '8.8.8.8'...\"\n" - userdata += "echo \"nameserver 8.8.8.8\" >> /etc/resolv.conf\n" - userdata += "echo \"Install curl...\"\n" - userdata += "apt-get install curl\n" - userdata += "echo \"Inject public key...\"\n" - userdata += ("echo \"ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCuPXrV3" - "geeHc6QUdyUr/1Z+yQiqLcOskiEGBiXr4z76MK4abiFmDZ18OMQlc" - "fl0p3kS0WynVgyaOHwZkgy/DIoIplONVr2CKBKHtPK+Qcme2PVnCtv" - "EqItl/FcD+1h5XSQGoa+A1TSGgCod/DPo+pes0piLVXP8Ph6QS1k7S" - "ic7JDeRQ4oT1bXYpJ2eWBDMfxIWKZqcZRiGPgMIbJ1iEkxbpeaAd9O" - "4MiM9nGCPESmed+p54uYFjwEDlAJZShcAZziiZYAvMZhvAhe6USljc" - "7YAdalAnyD/jwCHuwIrUw/lxo7UdNCmaUxeobEYyyFA1YVXzpNFZya" - "XPGAAYIJwEq/ openbaton@opnfv\" >> /home/ubuntu/.ssh/aut" - "horized_keys\n") - userdata += "echo \"Download bootstrap...\"\n" - userdata += ("curl -s %s " - "> ./bootstrap\n" % self.bootstrap_link) - userdata += ("curl -s %s" - "> ./config_file\n" % self.bootstrap_config_link) - userdata += ("echo \"Disable usage of mysql...\"\n") - userdata += "sed -i s/mysql=.*/mysql=no/g /config_file\n" - userdata += ("echo \"Setting 'rabbitmq_broker_ip' to '%s'\"\n" - % floatip) - userdata += ("sed -i s/rabbitmq_broker_ip=localhost/rabbitmq_broker_ip" - "=%s/g /config_file\n" % floatip) - userdata += "echo \"Set autostart of components to 'false'\"\n" - userdata += "export OPENBATON_COMPONENT_AUTOSTART=false\n" - userdata += "echo \"Execute bootstrap...\"\n" - bootstrap = "sh ./bootstrap release -configFile=./config_file" - userdata += bootstrap + "\n" - userdata += "echo \"Setting 'nfvo.plugin.timeout' to '300000'\"\n" - userdata += ("echo \"nfvo.plugin.timeout=600000\" >> " - "/etc/openbaton/openbaton-nfvo.properties\n") - userdata += ( - "wget %s -O /etc/openbaton/openbaton-vnfm-generic-user-data.sh\n" % - self.userdata_file) - userdata += "sed -i '113i\ \ \ \ sleep 60' " \ - "/etc/openbaton/openbaton-vnfm-generic-user-data.sh\n" - userdata += "echo \"Starting NFVO\"\n" - userdata += "service openbaton-nfvo restart\n" - userdata += "echo \"Starting Generic VNFM\"\n" - userdata += "service openbaton-vnfm-generic restart\n" - userdata += "echo \"...end of userdata...\"\n" - - sg_id = os_utils.create_security_group_full(neutron_client, - "orchestra-sec-group", - "allowall") - - os_utils.create_secgroup_rule(neutron_client, sg_id, "ingress", - "icmp", 0, 255) - os_utils.create_secgroup_rule(neutron_client, sg_id, "egress", - "icmp", 0, 255) - os_utils.create_secgroup_rule(neutron_client, sg_id, "ingress", - "tcp", 1, 65535) - os_utils.create_secgroup_rule(neutron_client, sg_id, "ingress", - "udp", 1, 65535) - os_utils.create_secgroup_rule(neutron_client, sg_id, "egress", - "tcp", 1, 65535) - os_utils.create_secgroup_rule(neutron_client, sg_id, "egress", - "udp", 1, 65535) - - self.logger.info("Security group set") - - self.logger.info("Create instance....") - self.logger.info("flavor: m1.medium\n" + def deploy_orchestrator(self): + self.logger.info("Deploying orchestrator Open Baton ...") + self.logger.info("Details: %s", self.baton['details']) + start_time = time.time() + + self.logger.info("Creating orchestra instance...") + userdata = get_userdata(self.baton) + self.logger.info("flavor: %s\n" "image: %s\n" - "network_id: %s\n" - "userdata: %s\n", - self.imagename, - network_id, - userdata) - - instance = os_utils.create_instance_and_wait_for_active( - "orchestra", - os_utils.get_image_id(glance_client, self.imagename), - network_id, - "orchestra-openbaton", - config_drive=False, + "network_id: %s\n", + self.baton['details']['flavor']['name'], + self.baton['details']['image'], + self.baton['details']['network']['id']) + self.logger.debug("userdata: %s\n", userdata) + # setting up image + image_settings = ImageSettings( + name=self.baton['details']['image'], + image_user='ubuntu', + exists=True) + # setting up port + port_settings = PortSettings( + name='%s_port' % self.baton['details']['name'], + network_name=self.baton['details']['network']['name']) + # build configuration of vm + orchestra_settings = VmInstanceSettings( + name=self.baton['details']['name'], + flavor=self.baton['details']['flavor']['name'], + port_settings=[port_settings], + security_group_names=[self.baton['details']['sec_group']], userdata=userdata) + orchestra_vm = OpenStackVmInstance(self.snaps_creds, + orchestra_settings, + image_settings) - self.ob_instance_id = instance.id - - self.logger.info("Adding sec group to orchestra instance") - os_utils.add_secgroup_to_instance(nova_client, - self.ob_instance_id, sg_id) + orchestra_vm.create() + self.created_resources.append(orchestra_vm) + self.baton['details']['id'] = orchestra_vm.get_vm_info()['id'] + self.logger.info( + "Created orchestra instance: %s", + self.baton['details']['id']) self.logger.info("Associating floating ip: '%s' to VM '%s' ", - floatip, - "orchestra-openbaton") - if not os_utils.add_floating_ip(nova_client, instance.id, floatip): + self.baton['details']['fip'].ip, + self.baton['details']['name']) + nova_client = os_utils.get_nova_client() + if not os_utils.add_floating_ip( + nova_client, + self.baton['details']['id'], + self.baton['details']['fip'].ip): + duration = time.time() - start_time + self.details["orchestrator"].update( + status='FAIL', duration=duration) self.logger.error("Cannot associate floating IP to VM.") return False self.logger.info("Waiting for Open Baton NFVO to be up and running...") - x = 0 - while x < 200: - if servertest(floatip, "8080"): + timeout = 0 + while timeout < 200: + if servertest( + self.baton['details']['fip'].ip, + "8080"): break else: - self.logger.debug( - "Open Baton NFVO is not started yet (%ss)" % - (x * 5)) + self.logger.info("Open Baton NFVO is not started yet (%ss)", + (timeout * 5)) time.sleep(5) - x += 1 + timeout += 1 - if x == 200: + if timeout >= 200: + duration = time.time() - start_time + self.details["orchestrator"].update( + status='FAIL', duration=duration) self.logger.error("Open Baton is not started correctly") + return False - self.ob_ip = floatip - self.ob_password = "openbaton" - self.ob_username = "admin" - self.ob_https = False - self.ob_port = "8080" - self.logger.info("Waiting for all components up and running...") + self.logger.info("Waiting for all components to be up and running...") time.sleep(60) - self.details["orchestrator"] = { - 'status': "PASS", 'result': "Deploy Open Baton NFVO: OK"} + duration = time.time() - start_time + self.details["orchestrator"].update(status='PASS', duration=duration) self.logger.info("Deploy Open Baton NFVO: OK") return True def deploy_vnf(self): - self.logger.info("Starting vIMS Deployment...") + start_time = time.time() + self.logger.info("Deploying OpenIMS...") - self.main_agent = MainAgent(nfvo_ip=self.ob_ip, - nfvo_port=self.ob_port, - https=self.ob_https, - version=1, - username=self.ob_username, - password=self.ob_password) + main_agent = MainAgent( + nfvo_ip=self.baton['details']['fip'].ip, + nfvo_port=8080, + https=False, + version=1, + username=self.baton['credentials']['username'], + password=self.baton['credentials']['password']) self.logger.info( - "Check if openims Flavor is available, if not create one") - flavor_exist, flavor_id = os_utils.get_or_create_flavor( - "m1.small", - "2048", - '20', - '1', - public=True) - self.logger.debug("Flavor id: %s", flavor_id) + "Check if openims Flavor is available, if not, create one") + flavor_settings = FlavorSettings( + name=self.vnf['requirements']['flavor']['name'], + ram=self.vnf['requirements']['flavor']['ram_min'], + disk=self.vnf['requirements']['flavor']['disk'], + vcpus=self.vnf['requirements']['flavor']['vcpus']) + flavor = OpenStackFlavor(self.snaps_creds, flavor_settings) + flavor_info = flavor.create() + self.logger.debug("Flavor id: %s", flavor_info.id) self.logger.info("Getting project 'default'...") - project_agent = self.main_agent.get_agent("project", self.ob_projectid) - for p in json.loads(project_agent.find()): - if p.get("name") == "default": - self.ob_projectid = p.get("id") - self.logger.info("Found project 'default': %s", p) + project_agent = main_agent.get_agent("project", "") + for project in json.loads(project_agent.find()): + if project.get("name") == "default": + self.baton['details']['project_id'] = project.get("id") + self.logger.info("Found project 'default': %s", project) break - self.logger.debug("project id: %s", self.ob_projectid) - if self.ob_projectid == "": - self.logger.error("Default project id was not found!") + vim_json = self.get_vim_descriptor() + self.logger.info("Registering VIM: %s", vim_json) - creds = os_utils.get_credentials() - self.logger.info("PoP creds: %s", creds) + main_agent.get_agent( + "vim", project_id=self.baton['details']['project_id']).create( + entity=json.dumps(vim_json)) - if os_utils.is_keystone_v3(): - self.logger.info( - "Using v3 API of OpenStack... -> Using OS_PROJECT_ID") - project_id = os_utils.get_tenant_id( - os_utils.get_keystone_client(), - creds.get("project_name")) - else: - self.logger.info( - "Using v2 API of OpenStack... -> Using OS_TENANT_NAME") - project_id = creds.get("tenant_name") + market_agent = main_agent.get_agent( + "market", project_id=self.baton['details']['project_id']) - self.logger.debug("project id: %s", project_id) - - vim_json = { - "name": "vim-instance", - "authUrl": creds.get("auth_url"), - "tenant": project_id, - "username": creds.get("username"), - "password": creds.get("password"), - "securityGroups": [ - "default", - "orchestra-sec-group" - ], - "type": "openstack", - "location": { - "name": "opnfv", - "latitude": "52.525876", - "longitude": "13.314400" - } - } - - self.logger.debug("Registering VIM: %s", vim_json) - - self.main_agent.get_agent( - "vim", - project_id=self.ob_projectid).create(entity=json.dumps(vim_json)) - - market_agent = self.main_agent.get_agent("market", - project_id=self.ob_projectid) - - nsd = {} try: - self.logger.info("sending: %s", self.market_link) - nsd = market_agent.create(entity=self.market_link) + self.logger.info("sending: %s", self.vnf['descriptor']['url']) + nsd = market_agent.create(entity=self.vnf['descriptor']['url']) + if nsd.get('id') is None: + self.logger.error("NSD not onboarded correctly") + duration = time.time() - start_time + self.details["vnf"].update(status='FAIL', duration=duration) + return False + self.baton['details']['nsd_id'] = nsd.get('id') self.logger.info("Onboarded NSD: " + nsd.get("name")) - except NfvoException as e: - self.logger.error(e.message) - nsr_agent = self.main_agent.get_agent("nsr", - project_id=self.ob_projectid) - nsd_id = nsd.get('id') - if nsd_id is None: - self.logger.error("NSD not onboarded correctly") + nsr_agent = main_agent.get_agent( + "nsr", project_id=self.baton['details']['project_id']) - try: - self.nsr = nsr_agent.create(nsd_id) - except NfvoException as e: - self.logger.error(e.message) + self.baton['details']['nsr'] = nsr_agent.create( + self.baton['details']['nsd_id']) + except NfvoException as exc: + self.logger.error(exc.message) + duration = time.time() - start_time + self.details["vnf"].update(status='FAIL', duration=duration) + return False - if self.nsr.get('code') is not None: + if self.baton['details']['nsr'].get('code') is not None: self.logger.error( "vIMS cannot be deployed: %s -> %s", - self.nsr.get('code'), - self.nsr.get('message')) + self.baton['details']['nsr'].get('code'), + self.baton['details']['nsr'].get('message')) self.logger.error("vIMS cannot be deployed") + duration = time.time() - start_time + self.details["vnf"].update(status='FAIL', duration=duration) + return False - i = 0 + timeout = 0 self.logger.info("Waiting for NSR to go to ACTIVE...") - while self.nsr.get("status") != 'ACTIVE' and self.nsr.get( - "status") != 'ERROR': - i += 1 - if i == 150: - self.logger.error("INACTIVE NSR after %s sec..", 5 * i) - + while self.baton['details']['nsr'].get("status") != 'ACTIVE' \ + and self.baton['details']['nsr'].get("status") != 'ERROR': + timeout += 1 + self.logger.info("NSR is not yet ACTIVE... (%ss)", 5 * timeout) + if timeout == 150: + self.logger.error("INACTIVE NSR after %s sec..", 5 * timeout) + duration = time.time() - start_time + self.details["vnf"].update(status='FAIL', duration=duration) + return False time.sleep(5) - self.nsr = json.loads(nsr_agent.find(self.nsr.get('id'))) - - if self.nsr.get("status") == 'ACTIVE': - self.details["vnf"] = {'status': "PASS", 'result': self.nsr} - self.logger.info("Deploy VNF: OK") + self.baton['details']['nsr'] = json.loads( + nsr_agent.find(self.baton['details']['nsr'].get('id'))) + + duration = time.time() - start_time + if self.baton['details']['nsr'].get("status") == 'ACTIVE': + self.details["vnf"].update(status='PASS', duration=duration) + self.logger.info("Sleep for 60s to ensure that all " + "services are up and running...") + time.sleep(60) + result = True else: - self.details["vnf"] = {'status': "FAIL", 'result': self.nsr} - self.logger.error(self.nsr) - self.logger.error("Deploy VNF: ERROR") - return False - - self.ob_nsr_id = self.nsr.get("id") - self.logger.info( - "Sleep for 60s to ensure that all services are up and running...") - time.sleep(60) - return True + self.details["vnf"].update(status='FAIL', duration=duration) + self.logger.error("NSR: %s", self.baton['details'].get('nsr')) + result = False + return result def test_vnf(self): - # Adaptations probably needed - # code used for cloudify_ims - # ruby client on jumphost calling the vIMS on the SUT + self.logger.info("Testing VNF OpenIMS...") + start_time = time.time() self.logger.info( - "Testing if %s works properly...", self.nsr.get('name')) - for vnfr in self.nsr.get('vnfr'): + "Testing if %s works properly...", + self.baton['details']['nsr'].get('name')) + for vnfr in self.baton['details']['nsr'].get('vnfr'): self.logger.info( "Checking ports %s of VNF %s", self.ims_conf.get(vnfr.get('name')).get('ports'), @@ -439,49 +626,75 @@ class ImsVnf(vnf.VnfOnBoarding): self.logger.debug( "Checking ports of VNFC instance %s", vnfci.get('hostname')) - for floatingIp in vnfci.get('floatingIps'): + for floating_ip in vnfci.get('floatingIps'): self.logger.debug( "Testing %s:%s", vnfci.get('hostname'), - floatingIp.get('ip')) + floating_ip.get('ip')) for port in self.ims_conf.get( vnfr.get('name')).get('ports'): - if servertest(floatingIp.get('ip'), port): + if servertest(floating_ip.get('ip'), port): self.logger.info( "VNFC instance %s is reachable at %s:%s", vnfci.get('hostname'), - floatingIp.get('ip'), + floating_ip.get('ip'), port) else: self.logger.error( "VNFC instance %s is not reachable " "at %s:%s", vnfci.get('hostname'), - floatingIp.get('ip'), + floating_ip.get('ip'), port) - self.details["test_vnf"] = { - 'status': "FAIL", 'result': ( + duration = time.time() - start_time + self.details["test_vnf"].update( + status='FAIL', duration=duration, esult=( "Port %s of server %s -> %s is " "not reachable", port, vnfci.get('hostname'), - floatingIp.get('ip'))} + floating_ip.get('ip'))) self.logger.error("Test VNF: ERROR") return False - - self.details["test_vnf"] = { - 'status': "PASS", - 'result': "All tests have been executed successfully"} + duration = time.time() - start_time + self.details["test_vnf"].update(status='PASS', duration=duration) self.logger.info("Test VNF: OK") return True def clean(self): - self.main_agent.get_agent( - "nsr", - project_id=self.ob_projectid).delete(self.ob_nsr_id) - time.sleep(5) - os_utils.delete_instance(nova_client=os_utils.get_nova_client(), - instance_id=self.ob_instance_id) - # question is the clean removing also the VM? - # I think so since is goinf to remove the tenant... + self.logger.info("Cleaning...") + try: + main_agent = MainAgent( + nfvo_ip=self.baton['details']['fip'].ip, + nfvo_port=8080, + https=False, + version=1, + username=self.baton['credentials']['username'], + password=self.baton['credentials']['password']) + self.logger.info("Terminating OpenIMS VNF...") + if (self.baton['details'].get('nsr')): + main_agent.get_agent( + "nsr", + project_id=self.baton['details']['project_id'])\ + .delete(self.baton['details']['nsr'].get('id')) + self.logger.info("Waiting 60sec for terminating OpenIMS VNF..") + time.sleep(60) + # os_utils.delete_instance(nova_client=os_utils.get_nova_client(), + # instance_id=self.baton_instance_id) + except (NfvoException, KeyError) as exc: + self.logger.error('Unexpected error cleaning - %s', exc) + + try: + neutron_client = os_utils.get_neutron_client(self.creds) + snaps_utils.neutron_utils.delete_floating_ip( + neutron_client, self.baton['details']['fip']) + except Exception as exc: + self.logger.error('Unexpected error cleaning - %s', exc) + + for resource in reversed(self.created_resources): + try: + self.logger.info("Cleaning %s", str(resource)) + resource.clean() + except Exception as exc: + self.logger.error('Unexpected error cleaning - %s', exc) super(ImsVnf, self).clean() diff --git a/functest/opnfv_tests/vnf/ims/orchestra_ims.yaml b/functest/opnfv_tests/vnf/ims/orchestra_ims.yaml index 5b25d3c96..9deb11c7b 100644 --- a/functest/opnfv_tests/vnf/ims/orchestra_ims.yaml +++ b/functest/opnfv_tests/vnf/ims/orchestra_ims.yaml @@ -1,13 +1,37 @@ tenant_images: ubuntu_14.04: http://cloud-images.ubuntu.com/trusty/current/trusty-server-cloudimg-amd64-disk1.img openims: http://marketplace.openbaton.org:8082/api/v1/images/52e2ccc0-1dce-4663-894d-28aab49323aa/img -openbaton: - bootstrap_link: http://get.openbaton.org/bootstraps/bootstrap_3.2.0_opnfv/bootstrap - bootstrap_config_link: http://get.openbaton.org/bootstraps/bootstrap_3.2.0_opnfv/bootstrap-config-file - userdata: - file: https://raw.githubusercontent.com/openbaton/generic-vnfm/3.2.0/src/main/resources/user-data.sh - marketplace_link: http://marketplace.openbaton.org:8082/api/v1/nsds/fokus/OpenImsCore/3.2.0/json - imagename: ubuntu_14.04 +orchestrator: + name: openbaton + version: '3.2.0' + requirements: + flavor: + name: orchestra + ram_min: 4096 + disk: 5 + vcpus: 2 + os_image: 'ubuntu_14.04' + bootstrap: + url: http://get.openbaton.org/bootstraps/bootstrap_3.2.0_opnfv/bootstrap + config: + url: http://get.openbaton.org/bootstraps/bootstrap_3.2.0_opnfv/bootstrap-config-file + gvnfm: + userdata: + url: https://raw.githubusercontent.com/openbaton/generic-vnfm/3.2.0/src/main/resources/user-data.sh + credentials: + username: admin + password: openbaton +vnf: + name: openims + descriptor: + url: http://marketplace.openbaton.org:8082/api/v1/nsds/fokus/OpenImsCore/3.2.0/json + requirements: + flavor: + name: m1.small + ram_min: 2048 + disk: 5 + vcpus: 2 + vIMS: scscf: ports: [3870, 6060] diff --git a/functest/tests/unit/vnf/ims/test_orchestra_ims.py b/functest/tests/unit/vnf/ims/test_orchestra_ims.py new file mode 100644 index 000000000..219997301 --- /dev/null +++ b/functest/tests/unit/vnf/ims/test_orchestra_ims.py @@ -0,0 +1,224 @@ +#!/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 + +"""Test module for orchestra_ims""" + +import logging +import unittest + +import mock + +from functest.core import vnf +from functest.opnfv_tests.vnf.ims import orchestra_ims + + +class OrchestraImsTesting(unittest.TestCase): + """Test class for orchestra_ims""" + def setUp(self): + + self.tenant = 'orchestra_ims' + self.creds = {'username': 'mocked_username', + 'password': 'mocked_password'} + self.tenant_images = { + 'image1': 'mocked_image_url_1', + 'image2': 'mocked_image_url_2' + } + self.orchestrator = { + 'name': 'openbaton', + 'version': '3.2.0', + 'object': 'foo', + 'requirements': { + 'flavor': { + 'name': 'mocked_flavor', + 'ram_min': 4096, + 'disk': 5, + 'vcpus': 2 + }, + 'os_image': 'mocked_image' + }, + 'bootstrap': { + 'url': 'mocked_bootstrap_url', + 'config': { + 'url': 'mocked_config_url'} + }, + 'gvnfm': { + 'userdata': { + 'url': 'mocked_userdata_url' + } + }, + 'credentials': { + 'username': 'mocked_username', + 'password': 'mocked_password' + } + } + self.vnf = { + 'name': 'openims', + 'descriptor': { + 'url': 'mocked_descriptor_url' + }, + 'requirements': { + 'flavor': { + 'name': 'mocked_flavor', + 'ram_min': 2048, + 'disk': 5, + 'vcpus': 2} + } + } + self.vIMS = { + 'scscf': { + 'ports': [3870, 6060] + }, + 'pcscf': { + 'ports': [4060] + }, + 'icscf': { + 'ports': [3869, 5060] + }, + 'fhoss': { + 'ports': [3868] + }, + 'bind9': { + 'ports': [] + } + } + with mock.patch('functest.opnfv_tests.vnf.ims.orchestra_ims.' + 'os.makedirs'),\ + mock.patch('functest.opnfv_tests.vnf.ims.orchestra_ims.' + 'get_config', return_value={ + 'orchestrator': self.orchestrator, + 'name': self.orchestrator['name'], + 'version': self.orchestrator['version'], + 'requirements': self.orchestrator['requirements'], + 'credentials': self.orchestrator['credentials'], + 'bootstrap': self.orchestrator['bootstrap'], + 'gvnfm': self.orchestrator['gvnfm'], + 'os_image': + self.orchestrator['requirements']['os_image'], + 'flavor': + self.orchestrator['requirements']['flavor'], + 'url': self.orchestrator['bootstrap']['url'], + 'config': self.orchestrator['bootstrap']['config'], + 'tenant_images': self.tenant_images, + 'vnf': self.vnf, + 'vIMS': self.vIMS}): + self.ims_vnf = orchestra_ims.ImsVnf() + + self.details = {'orchestrator': {'status': 'PASS', 'duration': 120}, + 'vnf': {}, + 'test_vnf': {}} + + @mock.patch('functest.core.vnf.os_utils.get_keystone_client', + return_value='test') + @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf', + return_value=True) + @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf', + return_value=True) + @mock.patch('functest.core.vnf.os_utils.get_credentials', + return_value={'auth_url': 'test/v1'}) + @mock.patch( + 'functest.utils.openstack_utils.get_tenant_id', + return_value={'mocked_tenant_id'}) + @mock.patch( + 'functest.utils.openstack_utils.get_floating_ips', + return_value=[]) + @mock.patch('snaps.openstack.create_image.OpenStackImage.create') + @mock.patch('snaps.openstack.create_flavor.OpenStackFlavor.create') + @mock.patch( + 'snaps.openstack.create_security_group.OpenStackSecurityGroup.create') + @mock.patch('snaps.openstack.create_network.OpenStackNetwork.create') + @mock.patch('snaps.openstack.create_router.OpenStackRouter.create') + @mock.patch( + 'functest.opnfv_tests.openstack.snaps.snaps_utils.get_ext_net_name') + @mock.patch( + 'functest.opnfv_tests.openstack.snaps.snaps_utils.' + 'neutron_utils.create_floating_ip') + def test_prepare_default(self, *args): + """Testing prepare function without any exceptions expected""" + self.assertIsNone(self.ims_vnf.prepare()) + args[4].assert_called_once_with() + + @mock.patch('functest.core.vnf.os_utils.get_keystone_client', + return_value='test') + @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf', + return_value=True) + @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf', + return_value=True) + @mock.patch('functest.core.vnf.os_utils.get_credentials', + return_value={'auth_url': 'test/no_v'}) + @mock.patch('snaps.openstack.create_image.OpenStackImage.create') + def test_prepare_bad_auth_url(self, *args): + """Testing prepare function with bad auth url""" + with self.assertRaises(Exception): + self.ims_vnf.prepare() + args[0].assert_not_called() + + def test_prepare_missing_param(self): + """Testing prepare function with missing param""" + with self.assertRaises(vnf.VnfPreparationException): + self.ims_vnf.prepare() + + @mock.patch('functest.core.vnf.os_utils.get_keystone_client', + side_effect=Exception) + def test_prepare_keystone_exception(self, *args): + """Testing prepare function with keystone exception""" + with self.assertRaises(vnf.VnfPreparationException): + self.ims_vnf.prepare() + args[0].assert_called_once_with() + + @mock.patch('functest.core.vnf.os_utils.get_keystone_client', + return_value='test') + @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf', + side_effect=Exception) + def test_prepare_tenant_exception(self, *args): + """Testing prepare function with tenant exception""" + with self.assertRaises(vnf.VnfPreparationException): + self.ims_vnf.prepare() + args[1].assert_called_once_with() + + @mock.patch('functest.core.vnf.os_utils.get_keystone_client', + return_value='test') + @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf', + return_value=True) + @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf', + side_effect=Exception) + def test_prepare_user_exception(self, *args): + """Testing prepare function with user exception""" + with self.assertRaises(vnf.VnfPreparationException): + self.ims_vnf.prepare() + args[2].assert_called_once_with() + + @mock.patch('functest.core.vnf.os_utils.get_keystone_client', + return_value='test') + @mock.patch('functest.core.vnf.os_utils.get_or_create_tenant_for_vnf', + return_value=True) + @mock.patch('functest.core.vnf.os_utils.get_or_create_user_for_vnf', + return_value=True) + @mock.patch('functest.core.vnf.os_utils.get_credentials', + side_effect=Exception) + def test_prepare_credentials_exception(self, *args): + """Testing prepare function with credentials exception""" + with self.assertRaises(vnf.VnfPreparationException): + self.ims_vnf.prepare() + args[0].assert_called_once_with() + + # # @mock.patch('functest.opnfv_tests.vnf.ims.orchestra_ims.get_userdata') + # def test_deploy_orchestrator(self, *args): + # floating_ip = FloatingIp + # floating_ip.ip = 'mocked_ip' + # details = {'fip':floating_ip,'flavor':{'name':'mocked_name'}} + # self.orchestrator['details'] = details + # with mock.patch.dict(self.orchestrator, {'details': + # {'fip':floating_ip,'flavor':{'name':'mocked_name'}}}): + # # with mock.patch.dict(self.orchestrator, details): + # orchestra_ims.get_userdata(self.orchestrator) + # self.assertIsNone(self.ims_vnf.deploy_orchestrator()) + # args[4].assert_called_once_with() + + +if __name__ == "__main__": + logging.disable(logging.CRITICAL) + unittest.main(verbosity=2) |