aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--sfc/lib/cleanup.py39
-rw-r--r--sfc/lib/openstack_utils.py320
-rw-r--r--sfc/tests/functest/run_sfc_tests.py6
-rw-r--r--sfc/tests/functest/sfc_chain_deletion.py9
-rw-r--r--sfc/tests/functest/sfc_one_chain_two_service_functions.py4
-rw-r--r--sfc/tests/functest/sfc_parent_function.py96
-rw-r--r--sfc/tests/functest/sfc_symmetric_chain.py8
-rw-r--r--sfc/tests/functest/sfc_two_chains_SSH_and_HTTP.py9
-rw-r--r--sfc/unit_tests/unit/lib/test_cleanup.py152
-rw-r--r--sfc/unit_tests/unit/lib/test_openstack_utils.py925
10 files changed, 942 insertions, 626 deletions
diff --git a/sfc/lib/cleanup.py b/sfc/lib/cleanup.py
index ee3ad689..7ac7903a 100644
--- a/sfc/lib/cleanup.py
+++ b/sfc/lib/cleanup.py
@@ -3,6 +3,7 @@ import sys
import time
import sfc.lib.openstack_utils as os_sfc_utils
import sfc.lib.odl_utils as odl_utils
+from openstack import connection
logger = logging.getLogger(__name__)
@@ -73,16 +74,6 @@ def delete_vims():
os_sfc_utils.delete_vim(t, vim_id=vim)
-# Creators is a list full of SNAPs objects
-def delete_openstack_objects(creators):
- logger.info("Deleting the openstack objects...")
- for creator in reversed(creators):
- try:
- creator.clean()
- except Exception as e:
- logger.error('Unexpected error cleaning - %s', e)
-
-
# Networking-odl generates a new security group when creating a router
# which is not tracked by SNAPs
def delete_untracked_security_groups():
@@ -131,9 +122,33 @@ def cleanup_mano_objects(mano):
cleanup_nsfc_objects()
-def cleanup(creators, mano, odl_ip=None, odl_port=None):
+def delete_openstack_objects(testcase_config, creators):
+ conn = connection.from_config(verify=False)
+ for creator in creators:
+ if creator.name == testcase_config.subnet_name:
+ subnet_obj = creator
+
+ for creator in reversed(creators):
+ try:
+ logger.info("Deleting " + creator.name)
+ if creator.name == testcase_config.router_name:
+ logger.info("Removing subnet from router")
+ conn.network.remove_interface_from_router(
+ creator.id, subnet_obj.id)
+ time.sleep(2)
+ logger.info("Deleting router")
+ conn.network.delete_router(creator)
+ else:
+ creator.delete(conn.session)
+ time.sleep(2)
+ creators.remove(creator)
+ except Exception as e:
+ logger.error('Unexpected error cleaning - %s', e)
+
+
+def cleanup(testcase_config, creators, mano, odl_ip=None, odl_port=None):
cleanup_mano_objects(mano)
- delete_openstack_objects(creators)
+ delete_openstack_objects(testcase_config, creators)
delete_untracked_security_groups()
if odl_ip is not None and odl_port is not None:
cleanup_odl(odl_ip, odl_port)
diff --git a/sfc/lib/openstack_utils.py b/sfc/lib/openstack_utils.py
index 38ca61bb..0f5da884 100644
--- a/sfc/lib/openstack_utils.py
+++ b/sfc/lib/openstack_utils.py
@@ -3,35 +3,29 @@ import time
import json
import logging
import yaml
+import urllib2
from tackerclient.tacker import client as tackerclient
from functest.utils import constants
from functest.utils import env
from snaps.openstack.tests import openstack_tests
-from snaps.openstack.create_image import OpenStackImage
-from snaps.config.image import ImageConfig
-from snaps.config.flavor import FlavorConfig
-from snaps.openstack.create_flavor import OpenStackFlavor
-from snaps.config.network import NetworkConfig, SubnetConfig, PortConfig
-from snaps.openstack.create_network import OpenStackNetwork
-from snaps.config.router import RouterConfig
-from snaps.openstack.create_router import OpenStackRouter
-from snaps.config.security_group import (
- Protocol, SecurityGroupRuleConfig, Direction, SecurityGroupConfig)
-from snaps.openstack.create_security_group import OpenStackSecurityGroup
+from snaps.config.vm_inst import FloatingIpConfig
import snaps.openstack.create_instance as cr_inst
-from snaps.config.vm_inst import VmInstanceConfig, FloatingIpConfig
from snaps.openstack.utils import (
nova_utils, neutron_utils, heat_utils, keystone_utils)
+from openstack import connection
+from neutronclient.neutron import client as neutronclient
logger = logging.getLogger(__name__)
DEFAULT_TACKER_API_VERSION = '1.0'
+DEFAULT_API_VERSION = '2'
class OpenStackSFC:
def __init__(self):
+ self.conn = self.get_os_connection()
self.os_creds = openstack_tests.get_credentials(
os_env_file=constants.ENV_FILE)
self.creators = []
@@ -39,138 +33,184 @@ class OpenStackSFC:
self.neutron = neutron_utils.neutron_client(self.os_creds)
self.heat = heat_utils.heat_client(self.os_creds)
self.keystone = keystone_utils.keystone_client(self.os_creds)
+ self.neutron_client = neutronclient.\
+ Client(self.get_neutron_client_version(),
+ session=self.conn.session)
+
+ def get_os_connection(self):
+ return connection.from_config(verify=False)
+
+ def get_neutron_client_version(self):
+ api_version = os.getenv('OS_NETWORK_API_VERSION')
+ if api_version is not None:
+ logger.info("OS_NETWORK_API_VERSION is set in env as '%s'",
+ api_version)
+ return api_version
+ return DEFAULT_API_VERSION
def register_glance_image(self, name, url, img_format, public):
logger.info("Registering the image...")
- # Check whether the image is local or not
- if 'http' in url:
- image_settings = ImageConfig(name=name,
- img_format=img_format,
- url=url,
- public=public,
- image_user='admin')
+ image = self.conn.image.find_image(name)
+ if image:
+ logger.info("Image %s already exists." % image.name)
else:
- image_settings = ImageConfig(name=name,
- img_format=img_format,
- image_file=url,
- public=public,
- image_user='admin')
-
- # TODO Remove this when tacker is part of SNAPS
- self.image_settings = image_settings
+ if 'http' in url:
+ logger.info("Downloading image")
+ response = urllib2.urlopen(url)
+ image_data = response.read()
+ else:
+ with open(url) as f:
+ image_data = f.read()
- image_creator = OpenStackImage(self.os_creds, image_settings)
- image_creator.create()
+ image_settings = {'name': name,
+ 'disk_format': img_format,
+ 'data': image_data,
+ 'is_public': public,
+ 'container_format': 'bare'}
+ image = self.conn.image.upload_image(**image_settings)
+ self.creators.append(image)
+ logger.info("Image created")
- self.creators.append(image_creator)
- return image_creator
+ return image
def create_flavor(self, name, ram, disk, vcpus):
- logger.info("Creating the flavor...")
- flavor_settings = FlavorConfig(name=name, ram=ram, disk=disk,
- vcpus=vcpus)
- flavor_creator = OpenStackFlavor(self.os_creds, flavor_settings)
- flavor = flavor_creator.create()
+ logger.info("Creating flavor...")
+ flavor_settings = {"name": name, "ram": ram, "disk": disk,
+ "vcpus": vcpus}
- self.creators.append(flavor_creator)
+ flavor = self.conn.compute.create_flavor(**flavor_settings)
+
+ self.creators.append(flavor)
return flavor
def create_network_infrastructure(self, net_name, subnet_name, subnet_cidr,
router_name):
- logger.info("Creating networks...")
+ logger.info("Creating Networks...")
# Network and subnet
- subnet_settings = SubnetConfig(name=subnet_name, cidr=subnet_cidr)
- network_settings = NetworkConfig(name=net_name,
- subnet_settings=[subnet_settings])
- network_creator = OpenStackNetwork(self.os_creds, network_settings)
- network = network_creator.create()
+ network = self.conn.network.create_network(name=net_name)
+ self.creators.append(network)
- self.creators.append(network_creator)
+ subnet_settings = {"name": subnet_name, "cidr": subnet_cidr,
+ "network_id": network.id, 'ip_version': '4'}
+ subnet = self.conn.network.create_subnet(**subnet_settings)
+ self.creators.append(subnet)
# Router
- logger.info("Creating the router...")
ext_network_name = env.get('EXTERNAL_NETWORK')
+ ext_net = self.conn.network.find_network(ext_network_name)
+ router_dict = {'network_id': ext_net.id}
- router_settings = RouterConfig(name=router_name,
- external_gateway=ext_network_name,
- internal_subnets=[subnet_name])
+ logger.info("Creating Router...")
+ router = self.conn.network.create_router(name=router_name)
- router_creator = OpenStackRouter(self.os_creds, router_settings)
- router = router_creator.create()
+ self.conn.network.add_interface_to_router(router.id,
+ subnet_id=subnet.id)
- self.creators.append(router_creator)
+ self.conn.network.update_router(router.id,
+ external_gateway_info=router_dict)
+ router_obj = self.conn.network.get_router(router.id)
+ self.creators.append(router_obj)
- return network, router
+ return network, router_obj
def create_security_group(self, sec_grp_name):
logger.info("Creating the security groups...")
- rule_ping = SecurityGroupRuleConfig(sec_grp_name=sec_grp_name,
- direction=Direction.ingress,
- protocol=Protocol.icmp)
-
- rule_ssh = SecurityGroupRuleConfig(sec_grp_name=sec_grp_name,
- direction=Direction.ingress,
- protocol=Protocol.tcp,
- port_range_min=22,
- port_range_max=22)
-
- rule_http = SecurityGroupRuleConfig(sec_grp_name=sec_grp_name,
- direction=Direction.ingress,
- protocol=Protocol.tcp,
- port_range_min=80,
- port_range_max=80)
+ sec_group = self.conn.network.create_security_group(name=sec_grp_name)
- rules = [rule_ping, rule_ssh, rule_http]
+ rule_ping = {"security_group_id": sec_group.id,
+ "direction": "ingress",
+ "protocol": "icmp"}
- secgroup_settings = SecurityGroupConfig(name=sec_grp_name,
- rule_settings=rules)
+ rule_ssh = {"security_group_id": sec_group.id,
+ "direction": "ingress",
+ "protocol": "tcp",
+ "port_range_min": 22,
+ "port_range_max": 22}
- sec_group_creator = OpenStackSecurityGroup(self.os_creds,
- secgroup_settings)
- sec_group = sec_group_creator.create()
+ rule_http = {"security_group_id": sec_group.id,
+ "direction": "ingress",
+ "protocol": "tcp",
+ "port_range_min": 80,
+ "port_range_max": 80}
- self.creators.append(sec_group_creator)
+ rules = [rule_ping, rule_ssh, rule_http]
+
+ for rule in rules:
+ self.conn.network.create_security_group_rule(**rule)
+
+ self.creators.append(sec_group)
return sec_group
- def create_instance(self, vm_name, flavor_name, image_creator, network,
- secgrp, av_zone, ports, port_security=True):
- logger.info("Creating the instance {}...".format(vm_name))
- port_settings = []
+ def create_instance(self, vm_name, flavor, image, network,
+ sec_group, av_zone, ports, port_security=True):
+ logger.info("Creating Key Pair {}...".format(vm_name))
+
+ keypair = self.conn.compute.\
+ create_keypair(name="{}_keypair".format(vm_name))
+ self.creators.append(keypair)
+ flavor_obj = self.conn.compute.find_flavor(flavor)
+
+ logger.info("Creating Port {}...".format(ports))
+ port_list = []
for port in ports:
- port_settings.append(
- PortConfig(name=port,
- port_security_enabled=port_security,
- network_name=network.name))
- if port_security:
- instance_settings = VmInstanceConfig(
- name=vm_name, flavor=flavor_name,
- security_group_names=str(secgrp.name),
- port_settings=port_settings,
- availability_zone=av_zone)
+ if port_security:
+ port_obj = self.conn.network.create_port(
+ name=port, is_port_security_enabled=port_security,
+ network_id=network.id, security_group_ids=[sec_group.id])
+ else:
+ port_obj = self.conn.network.create_port(
+ name=port, is_port_security_enabled=port_security,
+ network_id=network.id)
+ port_list.append(port_obj)
+ self.creators.append(port_obj)
+ logger.info("Creating the instance {}...".format(vm_name))
+
+ if len(port_list) > 1:
+ network_list = [{"port": port_list[0].id},
+ {"port": port_list[1].id}]
else:
- instance_settings = VmInstanceConfig(
- name=vm_name, flavor=flavor_name,
- port_settings=port_settings,
- availability_zone=av_zone)
+ network_list = [{"port": port_obj.id}]
+
+ instance = self.conn.compute.create_server(name=vm_name,
+ image_id=image.id,
+ flavor_id=flavor_obj.id,
+ networks=network_list,
+ key_name=keypair.name,
+ availability_zone=av_zone)
- instance_creator = cr_inst.OpenStackVmInstance(
- self.os_creds,
- instance_settings,
- image_creator.image_settings)
+ logger.info("Waiting for {} to become Active".format(instance.name))
+ self.conn.compute.wait_for_server(instance)
+ logger.info("{} is active".format(instance.name))
- instance = instance_creator.create(block=True)
+ self.creators.append(instance)
- self.creators.append(instance_creator)
- return instance, instance_creator
+ return instance, port_list
def get_av_zones(self):
'''
Return the availability zone each host belongs to
'''
- hosts = nova_utils.get_hypervisor_hosts(self.nova)
+ hosts = self.get_hypervisor_hosts()
return ['nova::{0}'.format(host) for host in hosts]
+ def get_hypervisor_hosts(self):
+ """
+ Returns the host names of all nova nodes with active hypervisors
+ :param nova: the Nova client
+ :return: a list of hypervisor host names
+ """
+ try:
+ nodes = []
+ hypervisors = self.conn.compute.hypervisors()
+ for hypervisor in hypervisors:
+ if hypervisor.state == "up":
+ nodes.append(hypervisor.name)
+ return nodes
+ except Exception as e:
+ logger.error("Error [get_hypervisors(compute)]: %s" % e)
+ return None
+
def get_compute_client(self):
'''
Return the compute where the client sits
@@ -196,18 +236,26 @@ class OpenStackSFC:
raise Exception("There is no VM with name '{}'!!".format(vm_name))
- def assign_floating_ip(self, router, vm, vm_creator):
+ def assign_floating_ip(self, vm, vm_port):
'''
Assign floating ips to all the VMs
'''
- name = vm.name + "-float"
- port_name = vm.ports[0].name
- float_ip = FloatingIpConfig(name=name,
- port_name=port_name,
- router_name=router.name)
- ip = vm_creator.add_floating_ip(float_ip)
+ logger.info(" Creating floating ips ")
+
+ ext_network_name = env.get('EXTERNAL_NETWORK')
+ ext_net = self.conn.network.find_network(ext_network_name)
+
+ fip = self.conn.network.create_ip(floating_network_id=ext_net.id,
+ port_id=vm_port.id)
+ logger.info(
+ " FLoating IP address {} created".format(fip.floating_ip_address))
+
+ logger.info(" Adding Floating IPs to instances ")
+ self.conn.compute.add_floating_ip_to_server(
+ vm.id, fip.floating_ip_address)
- return ip.ip
+ self.creators.append(fip)
+ return fip.floating_ip_address
# We need this function because tacker VMs cannot be created through SNAPs
def assign_floating_ip_vnfs(self, router, ips=None):
@@ -222,7 +270,7 @@ class OpenStackSFC:
for stack in stacks:
servers = heat_utils.get_stack_servers(self.heat,
self.nova,
- self.neutron,
+ self.neutron_client,
self.keystone,
stack,
project_name)
@@ -242,10 +290,11 @@ class OpenStackSFC:
break
if port_name is None:
- err_msg = "The VNF {} does not have any suitable port {} " \
- "for floating IP assignment".format(
- name,
- 'with ip any of ' + str(ips) if ips else '')
+ err_msg = ("The VNF {} does not have any suitable port {} "
+ "for floating IP assignment"
+ .format(name,
+ 'with ip any of ' +
+ str(ips) if ips else ''))
logger.error(err_msg)
raise Exception(err_msg)
@@ -275,13 +324,13 @@ class OpenStackSFC:
def delete_all_security_groups(self):
'''
Deletes all the available security groups
-
Needed until this bug is fixed:
https://bugs.launchpad.net/networking-odl/+bug/1763705
'''
- sec_groups = neutron_utils.list_security_groups(self.neutron)
+ logger.info("Deleting remaining security groups...")
+ sec_groups = self.conn.network.security_groups()
for sg in sec_groups:
- neutron_utils.delete_security_group(self.neutron, sg)
+ self.conn.network.delete_security_group(sg)
def wait_for_vnf(self, vnf_creator):
'''
@@ -293,7 +342,7 @@ class OpenStackSFC:
'''
Creates a networking-sfc port pair and group
'''
- logger.info("Creating the port pairs for %s" % vm_instance.name)
+ logger.info("Creating the port pairs...")
port_pair = dict()
port_pair['name'] = vm_instance.name + '-connection-points'
port_pair['description'] = 'port pair for ' + vm_instance.name
@@ -309,7 +358,7 @@ class OpenStackSFC:
logger.error("Only SFs with one or two ports are supported")
raise Exception("Failed to create port pairs")
port_pair_info = \
- self.neutron.create_sfc_port_pair({'port_pair': port_pair})
+ self.neutron_client.create_sfc_port_pair({'port_pair': port_pair})
if not port_pair_info:
logger.warning("Chain creation failed due to port pair "
"creation failed for vnf %(vnf)s",
@@ -320,7 +369,7 @@ class OpenStackSFC:
iterations = 5
found_it = False
for i in range(iterations):
- pp_list = self.neutron.list_sfc_port_pairs()['port_pairs']
+ pp_list = self.neutron_client.list_sfc_port_pairs()['port_pairs']
for pp in pp_list:
if pp['id'] == port_pair_info['port_pair']['id']:
found_it = True
@@ -334,6 +383,7 @@ class OpenStackSFC:
raise Exception("Port pair was not committed in openstack")
logger.info("Creating the port pair groups for %s" % vm_instance.name)
+
port_pair_group = {}
port_pair_group['name'] = vm_instance.name + '-port-pair-group'
port_pair_group['description'] = \
@@ -342,7 +392,7 @@ class OpenStackSFC:
port_pair_group['port_pairs'].append(port_pair_info['port_pair']['id'])
ppg_config = {'port_pair_group': port_pair_group}
port_pair_group_info = \
- self.neutron.create_sfc_port_pair_group(ppg_config)
+ self.neutron_client.create_sfc_port_pair_group(ppg_config)
if not port_pair_group_info:
logger.warning("Chain creation failed due to port pair group "
"creation failed for vnf "
@@ -376,7 +426,7 @@ class OpenStackSFC:
fc_config = {'flow_classifier': sfc_classifier_params}
fc_info = \
- self.neutron.create_sfc_flow_classifier(fc_config)
+ self.neutron_client.create_sfc_flow_classifier(fc_config)
logger.info("Creating the chain...")
port_chain = {}
@@ -389,35 +439,37 @@ class OpenStackSFC:
port_chain['chain_parameters'] = {}
port_chain['chain_parameters']['symmetric'] = True
chain_config = {'port_chain': port_chain}
- return self.neutron.create_sfc_port_chain(chain_config)
+ return self.neutron_client.create_sfc_port_chain(chain_config)
def delete_port_groups(self):
'''
Delete all port groups and port pairs
'''
logger.info("Deleting the port groups...")
- ppg_list = self.neutron.list_sfc_port_pair_groups()['port_pair_groups']
+ ppg_list = self.neutron_client.\
+ list_sfc_port_pair_groups()['port_pair_groups']
for ppg in ppg_list:
- self.neutron.delete_sfc_port_pair_group(ppg['id'])
+ self.neutron_client.delete_sfc_port_pair_group(ppg['id'])
logger.info("Deleting the port pairs...")
- pp_list = self.neutron.list_sfc_port_pairs()['port_pairs']
+ pp_list = self.neutron_client.list_sfc_port_pairs()['port_pairs']
for pp in pp_list:
- self.neutron.delete_sfc_port_pair(pp['id'])
+ self.neutron_client.delete_sfc_port_pair(pp['id'])
def delete_chain(self):
'''
Delete the classifiers and the chains
'''
logger.info("Deleting the chain...")
- pc_list = self.neutron.list_sfc_port_chains()['port_chains']
+ pc_list = self.neutron_client.list_sfc_port_chains()['port_chains']
for pc in pc_list:
- self.neutron.delete_sfc_port_chain(pc['id'])
+ self.neutron_client.delete_sfc_port_chain(pc['id'])
logger.info("Deleting the classifiers...")
- fc_list = self.neutron.list_sfc_flow_classifiers()['flow_classifiers']
+ fc_list = self.neutron_client.\
+ list_sfc_flow_classifiers()['flow_classifiers']
for fc in fc_list:
- self.neutron.delete_sfc_flow_classifier(fc['id'])
+ self.neutron_client.delete_sfc_flow_classifier(fc['id'])
# TACKER SECTION #
@@ -602,7 +654,6 @@ def get_vnf_ip(tacker_client, vnf_id=None, vnf_name=None):
"""
Get the management ip of the first VNF component as obtained from the
tacker REST API:
-
{
"vnf": {
...
@@ -819,8 +870,7 @@ def register_vim(tacker_client, vim_file=None):
create_vim(tacker_client, vim_file=tmp_file)
-def create_vnf_in_av_zone(
- tacker_client,
+def create_vnf_in_av_zone(tacker_client,
vnf_name,
vnfd_name,
vim_name,
@@ -832,9 +882,7 @@ def create_vnf_in_av_zone(
param_file = os.path.join(
'/tmp',
'param_{0}.json'.format(av_zone.replace('::', '_')))
- data = {
- 'zone': av_zone
- }
+ data = {'zone': av_zone}
with open(param_file, 'w+') as f:
json.dump(data, f)
create_vnf(tacker_client,
diff --git a/sfc/tests/functest/run_sfc_tests.py b/sfc/tests/functest/run_sfc_tests.py
index 2f72acb5..c3d68811 100644
--- a/sfc/tests/functest/run_sfc_tests.py
+++ b/sfc/tests/functest/run_sfc_tests.py
@@ -176,7 +176,8 @@ class SfcFunctest(testcase.TestCase):
result = {'status': 'FAILED'}
creators = tc_instance.get_creators()
if self.cleanup_flag is True:
- sfc_cleanup.cleanup(creators,
+ sfc_cleanup.cleanup(testcase_config,
+ creators,
COMMON_CONFIG.mano_component,
odl_ip=odl_ip,
odl_port=odl_port)
@@ -200,7 +201,8 @@ class SfcFunctest(testcase.TestCase):
self.details.update({test_name: dic})
if cleanup_run_flag is not True and self.cleanup_flag is True:
- sfc_cleanup.cleanup(creators,
+ sfc_cleanup.cleanup(testcase_config,
+ creators,
COMMON_CONFIG.mano_component,
odl_ip=odl_ip,
odl_port=odl_port)
diff --git a/sfc/tests/functest/sfc_chain_deletion.py b/sfc/tests/functest/sfc_chain_deletion.py
index aa6a1933..dac32143 100644
--- a/sfc/tests/functest/sfc_chain_deletion.py
+++ b/sfc/tests/functest/sfc_chain_deletion.py
@@ -42,9 +42,8 @@ class SfcChainDeletion(sfc_parent_function.SfcCommonTestCase):
t1 = threading.Thread(target=odl_utils.wait_for_classification_rules,
args=(self.ovs_logger, self.compute_nodes,
self.odl_ip, self.odl_port,
- self.client_instance.compute_host,
- [self.neutron_client_port],))
-
+ self.client_instance.hypervisor_hostname,
+ [self.neutron_port],))
try:
t1.start()
except Exception as e:
@@ -71,8 +70,8 @@ class SfcChainDeletion(sfc_parent_function.SfcCommonTestCase):
t2 = threading.Thread(target=odl_utils.wait_for_classification_rules,
args=(self.ovs_logger, self.compute_nodes,
self.odl_ip, self.odl_port,
- self.client_instance.compute_host,
- [self.neutron_client_port],))
+ self.client_instance.hypervisor_hostname,
+ [self.neutron_port],))
try:
t2.start()
except Exception as e:
diff --git a/sfc/tests/functest/sfc_one_chain_two_service_functions.py b/sfc/tests/functest/sfc_one_chain_two_service_functions.py
index ccae6807..38fa3fef 100644
--- a/sfc/tests/functest/sfc_one_chain_two_service_functions.py
+++ b/sfc/tests/functest/sfc_one_chain_two_service_functions.py
@@ -45,8 +45,8 @@ class SfcOneChainTwoServiceTC(sfc_parent_function.SfcCommonTestCase):
t1 = threading.Thread(target=odl_utils.wait_for_classification_rules,
args=(self.ovs_logger, self.compute_nodes,
self.odl_ip, self.odl_port,
- self.client_instance.compute_host,
- [self.neutron_client_port],))
+ self.client_instance.hypervisor_hostname,
+ [self.neutron_port],))
try:
t1.start()
except Exception as e:
diff --git a/sfc/tests/functest/sfc_parent_function.py b/sfc/tests/functest/sfc_parent_function.py
index 06ea4982..dab6abc0 100644
--- a/sfc/tests/functest/sfc_parent_function.py
+++ b/sfc/tests/functest/sfc_parent_function.py
@@ -47,16 +47,16 @@ class SfcCommonTestCase(object):
self.server_creator = None
self.client_instance = None
self.client_creator = None
- self.server_ip = None
self.vnf_id = None
self.client_floating_ip = None
self.server_floating_ip = None
self.fips_sfs = []
- self.neutron_client_port = None
- self.neutron_server_port = None
self.vnf_objects = dict()
self.testcase_config = testcase_config
self.vnfs = vnfs
+ self.port_server = None
+ self.server_ip = None
+ self.port_client = None
# n-sfc variables
self.port_groups = []
@@ -172,25 +172,30 @@ class SfcCommonTestCase(object):
logger.info('Topology description: {0}'
.format(self.test_topology['description']))
- self.server_instance, self.server_creator = \
+ self.server_instance, port_server = \
openstack_sfc.create_instance(SERVER, COMMON_CONFIG.flavor,
self.image_creator, self.network,
self.sg,
self.test_topology['server'],
[SERVER + '-port'])
+ self.port_server = port_server[0]
- self.client_instance, self.client_creator = \
+ self.client_instance, port_client = \
openstack_sfc.create_instance(CLIENT, COMMON_CONFIG.flavor,
self.image_creator, self.network,
self.sg,
self.test_topology['client'],
[CLIENT + '-port'])
+ self.port_client = port_client[0]
+
logger.info('This test is run with the topology {0}'.format(
self.test_topology['id']))
logger.info('Topology description: {0}'.format(
self.test_topology['description']))
- self.server_ip = self.server_instance.ports[0].ips[0]['ip_address']
+ port_fixed_ips = self.port_server.fixed_ips
+ for ip in port_fixed_ips:
+ self.server_ip = ip.get('ip_address')
logger.info("Server instance received private ip [{}]".format(
self.server_ip))
@@ -255,7 +260,7 @@ class SfcCommonTestCase(object):
ports = [vnf_name + '-port1', vnf_name + '-port2']
else:
ports = [vnf_name + '-port']
- vnf_instance, vnf_creator = \
+ vnf_instance, vnf_port = \
openstack_sfc.create_instance(vnf_name, COMMON_CONFIG.flavor,
self.vnf_image_creator,
self.network,
@@ -264,25 +269,19 @@ class SfcCommonTestCase(object):
ports,
port_security=False)
- if not openstack_sfc.wait_for_vnf(vnf_creator):
- raise Exception('ERROR while booting vnf %s' % vnf_name)
-
- self.creators.append(vnf_creator)
- self.vnf_objects[vnf_name] = [vnf_creator, vnf_instance]
+ self.vnf_objects[vnf_name] = [vnf_instance, vnf_port]
def assign_floating_ip_client_server(self):
"""Assign floating IPs on the router about server and the client
instances
-
:return: Floating IPs for client and server
"""
-
logger.info("Assigning floating IPs to client and server instances")
self.client_floating_ip = openstack_sfc.assign_floating_ip(
- self.router, self.client_instance, self.client_creator)
+ self.client_instance, self.port_client)
self.server_floating_ip = openstack_sfc.assign_floating_ip(
- self.router, self.server_instance, self.server_creator)
+ self.server_instance, self.port_server)
def assign_floating_ip_sfs(self):
"""Assign floating IPs to service function
@@ -300,11 +299,11 @@ class SfcCommonTestCase(object):
vnf_ip)
elif COMMON_CONFIG.mano_component == 'no-mano':
for vnf in self.vnfs:
- # creator object is in [0] and instance in [1]
- vnf_instance = self.vnf_objects[vnf][1]
- vnf_creator = self.vnf_objects[vnf][0]
- sf_floating_ip = openstack_sfc.assign_floating_ip(
- self.router, vnf_instance, vnf_creator)
+ # instance object is in [0] and port in [1]
+ vnf_instance = self.vnf_objects[vnf][0]
+ vnf_port = self.vnf_objects[vnf][1]
+ sf_floating_ip = openstack_sfc.\
+ assign_floating_ip(vnf_instance, vnf_port[0])
self.fips_sfs.append(sf_floating_ip)
def check_floating_ips(self):
@@ -429,14 +428,9 @@ class SfcCommonTestCase(object):
tosca_file=tosca_file,
vnffgd_name=vnffgd_name)
- self.neutron_client_port = \
- openstack_sfc.get_instance_port(self.client_instance,
- self.client_creator)
+ self.neutron_port = self.port_client
if symmetric:
- self.neutron_server_port = openstack_sfc.get_instance_port(
- self.server_instance,
- self.server_creator)
server_ip_prefix = self.server_ip + '/32'
os_sfc_utils.create_vnffg_with_param_file(
@@ -444,8 +438,8 @@ class SfcCommonTestCase(object):
vnffgd_name,
vnffg_name,
self.default_param_file,
- self.neutron_client_port.id,
- server_port=self.neutron_server_port.id,
+ self.neutron_port.id,
+ server_port=self.port_server.id,
server_ip=server_ip_prefix)
else:
@@ -454,61 +448,43 @@ class SfcCommonTestCase(object):
vnffgd_name,
vnffg_name,
self.default_param_file,
- self.neutron_client_port.id)
+ self.neutron_port.id)
elif COMMON_CONFIG.mano_component == 'no-mano':
if not only_chain:
for vnf in self.vnfs:
- # creator object is in [0] and instance in [1]
- vnf_instance = self.vnf_objects[vnf][1]
- vnf_creator = self.vnf_objects[vnf][0]
+ # vnf_instance is in [0] and vnf_port in [1]
+ vnf_instance = self.vnf_objects[vnf][0]
+ vnf_port = self.vnf_objects[vnf][1]
if symmetric:
# VNFs have two ports
- p1 = vnf_instance.name + '-port1'
- neutron_port1 = \
- openstack_sfc.get_instance_port(vnf_instance,
- vnf_creator,
- port_name=p1)
- p2 = vnf_instance.name + '-port2'
- neutron_port2 = \
- openstack_sfc.get_instance_port(vnf_instance,
- vnf_creator,
- port_name=p2)
+ neutron_port1 = vnf_port[0]
+ neutron_port2 = vnf_port[1]
neutron_ports = [neutron_port1, neutron_port2]
-
else:
- neutron_port1 = \
- openstack_sfc.get_instance_port(vnf_instance,
- vnf_creator)
+ neutron_port1 = vnf_port[0]
neutron_ports = [neutron_port1]
port_group = \
openstack_sfc.create_port_groups(neutron_ports,
vnf_instance)
-
self.port_groups.append(port_group)
- self.neutron_client_port = \
- openstack_sfc.get_instance_port(self.client_instance,
- self.client_creator)
+ self.neutron_port = self.port_client
if symmetric:
# We must pass the server_port and server_ip in the symmetric
# case. Otherwise ODL does not work well
- self.neutron_server_port = openstack_sfc.get_instance_port(
- self.server_instance,
- self.server_creator)
server_ip_prefix = self.server_ip + '/32'
- server_port_id = self.neutron_server_port.id
openstack_sfc.create_chain(self.port_groups,
- self.neutron_client_port.id,
+ self.neutron_port.id,
port, protocol, vnffg_name,
symmetric,
- server_port=server_port_id,
+ server_port=self.port_server.id,
server_ip=server_ip_prefix)
else:
openstack_sfc.create_chain(self.port_groups,
- self.neutron_client_port.id,
+ self.neutron_port.id,
port, protocol, vnffg_name,
symmetric)
@@ -590,8 +566,8 @@ class SfcCommonTestCase(object):
if not odl_utils.\
check_vnffg_deletion(self.odl_ip, self.odl_port,
self.ovs_logger,
- [self.neutron_client_port],
- self.client_instance.compute_host,
+ [self.neutron_port],
+ self.client_instance.hypervisor_hostname,
self.compute_nodes):
logger.debug("The chains were not correctly removed")
raise Exception("Chains not correctly removed, test failed")
diff --git a/sfc/tests/functest/sfc_symmetric_chain.py b/sfc/tests/functest/sfc_symmetric_chain.py
index 7df938e3..cec45219 100644
--- a/sfc/tests/functest/sfc_symmetric_chain.py
+++ b/sfc/tests/functest/sfc_symmetric_chain.py
@@ -50,10 +50,10 @@ class SfcSymmetricChain(sfc_parent_function.SfcCommonTestCase):
# rules
t1 = threading.Thread(target=symmetric_wait_for_classification_rules,
args=(self.ovs_logger, self.compute_nodes,
- self.server_instance.compute_host,
- self.neutron_server_port,
- self.client_instance.compute_host,
- self.neutron_client_port,
+ self.server_instance.hypervisor_hostname,
+ self.port_server,
+ self.client_instance.hypervisor_hostname,
+ self.port_client,
self.odl_ip, self.odl_port,))
try:
t1.start()
diff --git a/sfc/tests/functest/sfc_two_chains_SSH_and_HTTP.py b/sfc/tests/functest/sfc_two_chains_SSH_and_HTTP.py
index e2037a41..0cfbea22 100644
--- a/sfc/tests/functest/sfc_two_chains_SSH_and_HTTP.py
+++ b/sfc/tests/functest/sfc_two_chains_SSH_and_HTTP.py
@@ -46,9 +46,8 @@ class SfcTwoChainsSSHandHTTP(sfc_parent_function.SfcCommonTestCase):
t1 = threading.Thread(target=odl_utils.wait_for_classification_rules,
args=(self.ovs_logger, self.compute_nodes,
self.odl_ip, self.odl_port,
- self.client_instance.compute_host,
- [self.neutron_client_port],))
-
+ self.client_instance.hypervisor_hostname,
+ [self.neutron_port],))
try:
t1.start()
except Exception as e:
@@ -82,8 +81,8 @@ class SfcTwoChainsSSHandHTTP(sfc_parent_function.SfcCommonTestCase):
t2 = threading.Thread(target=odl_utils.wait_for_classification_rules,
args=(self.ovs_logger, self.compute_nodes,
self.odl_ip, self.odl_port,
- self.client_instance.compute_host,
- self.neutron_client_port,))
+ self.client_instance.hypervisor_hostname,
+ self.neutron_port,))
try:
t2.start()
except Exception as e:
diff --git a/sfc/unit_tests/unit/lib/test_cleanup.py b/sfc/unit_tests/unit/lib/test_cleanup.py
index b83e229f..8e68ce5b 100644
--- a/sfc/unit_tests/unit/lib/test_cleanup.py
+++ b/sfc/unit_tests/unit/lib/test_cleanup.py
@@ -14,8 +14,9 @@ import sfc.lib.cleanup as cleanup
from mock import patch
from mock import call
-from mock import Mock
from mock import DEFAULT
+from mock import Mock
+
__author__ = "Dimitrios Markou <mardim@intracom-telecom.com>"
@@ -262,37 +263,6 @@ class SfcCleanupTesting(unittest.TestCase):
mock_log.assert_has_calls(log_calls)
mock_del_vim.assert_has_calls(del_calls)
- @patch('sfc.lib.cleanup.logger.info')
- @patch('sfc.lib.cleanup.logger.error')
- def test_delete_openstack_objects_exception(self, mock_log_err,
- mock_log_info):
-
- """
- Check the proper functionality of the delete_openstack_objects
- function when exception occurs.
- """
-
- mock_creator_obj_one = Mock()
- mock_creator_obj_two = Mock()
- exception_one = Exception('First Boom!')
- exception_two = Exception('Second Boom!')
- attrs_list = [{'clean.side_effect': exception_one},
- {'clean.side_effect': exception_two}]
-
- mock_creator_obj_one.configure_mock(**attrs_list[0])
- mock_creator_obj_two.configure_mock(**attrs_list[1])
-
- mock_creator_objs_list = [mock_creator_obj_one, mock_creator_obj_two]
-
- log_calls = [call('Unexpected error cleaning - %s', exception_two),
- call('Unexpected error cleaning - %s', exception_one),
- call('Deleting the openstack objects...')]
-
- cleanup.delete_openstack_objects(mock_creator_objs_list)
-
- mock_log_err.assert_has_calls(log_calls[:2])
- mock_log_info.assert_has_calls(log_calls[2:])
-
@patch('sfc.lib.openstack_utils.OpenStackSFC', autospec=True)
def test_delete_untracked_security_groups(self,
mock_obj):
@@ -352,6 +322,119 @@ class SfcCleanupTesting(unittest.TestCase):
cleanup.cleanup_mano_objects('no-mano')
mock_cleanup_nsfc.assert_called_once()
+ @patch('sfc.lib.cleanup.connection')
+ @patch('sfc.lib.cleanup.logger.info')
+ def test_delete_openstack_objects(self, mock_log, mock_conn):
+ """
+ Checks the delete_chain method
+ """
+ testcase_config = Mock()
+ conn = Mock()
+ mock_creator_obj_one = Mock()
+ mock_creator_obj_one.name = 'subnet_name'
+ mock_creator_obj_two = Mock()
+ mock_creator_obj_two.name = 'creator_name'
+ mock_creator_objs_list = [mock_creator_obj_one, mock_creator_obj_two]
+
+ mock_conn.from_config.return_value = conn
+ testcase_config.subnet_name = mock_creator_obj_one.name
+ log_calls = [call('Deleting ' + mock_creator_obj_two.name),
+ call('Deleting ' + mock_creator_obj_one.name)]
+
+ cleanup.delete_openstack_objects(testcase_config,
+ mock_creator_objs_list)
+ mock_creator_obj_one.delete.\
+ assert_called_once_with(conn.session)
+ mock_creator_obj_two.delete.\
+ assert_called_once_with(conn.session)
+ mock_log.assert_has_calls(log_calls)
+
+ @patch('sfc.lib.cleanup.connection')
+ @patch('sfc.lib.cleanup.logger.info')
+ def test_delete_openstack_objects_router(self, mock_log, mock_conn):
+ """
+ Checks the delete_chain method
+ """
+ testcase_config = Mock()
+ conn = Mock()
+ mock_creator_obj = Mock()
+ mock_creator_obj.name = 'creator_name'
+ mock_creator_router = Mock()
+ mock_creator_router.name = 'router_name'
+ mock_creator_router.id = '1'
+ mock_creator_subnet = Mock()
+ mock_creator_subnet.name = 'subnet_name'
+ mock_creator_subnet.id = '2'
+ mock_creator_objs_list = [mock_creator_subnet,
+ mock_creator_router,
+ mock_creator_obj]
+
+ mock_conn.from_config.return_value = conn
+ testcase_config.router_name = mock_creator_router.name
+ testcase_config.subnet_name = mock_creator_subnet.name
+
+ conn.network.get_subnet.return_value = mock_creator_subnet
+ log_calls = [call('Deleting ' + mock_creator_obj.name),
+ call('Deleting ' + mock_creator_router.name),
+ call('Removing subnet from router'),
+ call('Deleting router'),
+ call('Deleting ' + mock_creator_subnet.name)]
+
+ cleanup.delete_openstack_objects(testcase_config,
+ mock_creator_objs_list)
+ conn.network.remove_interface_from_router.\
+ assert_called_once_with(mock_creator_router.id,
+ mock_creator_subnet.id)
+ conn.network.delete_router.\
+ assert_called_once_with(mock_creator_router)
+ mock_creator_obj.delete.\
+ assert_called_once_with(conn.session)
+ mock_creator_subnet.delete.\
+ assert_called_once_with(conn.session)
+ mock_log.assert_has_calls(log_calls)
+
+ @patch('sfc.lib.cleanup.connection')
+ @patch('sfc.lib.cleanup.logger.info')
+ @patch('sfc.lib.cleanup.logger.error')
+ def test_delete_openstack_objects_exception(self, mock_log_err,
+ mock_log_info, mock_conn):
+ """
+ Check the proper functionality of the delete_openstack_objects
+ function when exception occurs.
+ """
+ testcase_config = Mock()
+ conn = Mock()
+ mock_creator_obj_one = Mock()
+ mock_creator_obj_one.name = 'subnet_name'
+ mock_creator_obj_two = Mock()
+ mock_creator_obj_two.name = 'creator_name'
+ exception_one = Exception('First Boom!')
+ exception_two = Exception('Second Boom!')
+ attrs_list = [{'delete.side_effect': exception_one},
+ {'delete.side_effect': exception_two}]
+
+ mock_creator_obj_one.configure_mock(**attrs_list[0])
+ mock_creator_obj_two.configure_mock(**attrs_list[1])
+
+ mock_creator_objs_list = [mock_creator_obj_one, mock_creator_obj_two]
+ mock_conn.from_config.return_value = conn
+ testcase_config.subnet_name = mock_creator_obj_one.name
+
+ log_calls = [call('Deleting ' + mock_creator_obj_two.name),
+ call('Deleting ' + mock_creator_obj_one.name),
+ call('Unexpected error cleaning - %s', exception_two),
+ call('Unexpected error cleaning - %s', exception_one)]
+
+ cleanup.delete_openstack_objects(testcase_config,
+ mock_creator_objs_list)
+ mock_creator_obj_one.delete.\
+ assert_called_once_with(conn.session)
+ mock_creator_obj_two.delete.\
+ assert_called_once_with(conn.session)
+
+ mock_log_info.assert_has_calls(log_calls[:2])
+ mock_log_err.assert_has_calls(log_calls[2:])
+
@patch('sfc.lib.cleanup.delete_untracked_security_groups')
@patch('sfc.lib.cleanup.cleanup_mano_objects')
@patch('sfc.lib.cleanup.delete_openstack_objects')
@@ -362,14 +445,15 @@ class SfcCleanupTesting(unittest.TestCase):
mock_cleanup_mano,
mock_untr_sec_grps):
- cleanup.cleanup(['creator_one', 'creator_two'],
+ cleanup.cleanup('testcase_config', ['creator_one', 'creator_two'],
'mano',
self.odl_ip,
self.odl_port)
mock_cleanup_odl.assert_called_once_with(self.odl_ip,
self.odl_port)
- mock_del_os_obj.assert_called_once_with(['creator_one', 'creator_two'])
+ mock_del_os_obj.assert_called_once_with('testcase_config',
+ ['creator_one', 'creator_two'])
mock_cleanup_mano.assert_called_once_with('mano')
mock_untr_sec_grps.assert_called_once()
diff --git a/sfc/unit_tests/unit/lib/test_openstack_utils.py b/sfc/unit_tests/unit/lib/test_openstack_utils.py
index 3397c55a..78be88f0 100644
--- a/sfc/unit_tests/unit/lib/test_openstack_utils.py
+++ b/sfc/unit_tests/unit/lib/test_openstack_utils.py
@@ -14,7 +14,9 @@ import unittest
from mock import call
from mock import Mock
from mock import patch
+from mock import mock_open
from mock import DEFAULT
+from mock import MagicMock
import sfc.lib.openstack_utils as os_sfc_utils
from tackerclient.v1_0 import client as tacker_client
@@ -37,6 +39,10 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
'heat_client', autospec=True)
self.patcher6 = patch.object(os_sfc_utils.keystone_utils,
'keystone_client', autospec=True)
+ self.patcher7 = patch.object(os_sfc_utils.connection,
+ 'from_config', autospec=True,)
+ self.patcher8 = patch.object(os_sfc_utils.neutronclient,
+ 'Client', autospec=True,)
self.env_file = self.patcher1.start().return_value
self.os_creds = self.patcher2.start().return_value
@@ -44,6 +50,8 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
self.neutron = self.patcher4.start().return_value
self.heat = self.patcher5.start().return_value
self.keystone = self.patcher6.start().return_value
+ self.conn = self.patcher7.start().return_value
+ self.neutron_client = self.patcher8.start().return_value
self.os_sfc = os_sfc_utils.OpenStackSFC()
@@ -54,327 +62,498 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
self.patcher4.stop()
self.patcher5.stop()
self.patcher6.stop()
+ self.patcher7.stop()
+ self.patcher8.stop()
@patch('sfc.lib.openstack_utils.logger', autospec=True)
- @patch('sfc.lib.openstack_utils.ImageConfig', autospec=True)
- @patch('sfc.lib.openstack_utils.OpenStackImage', autospec=True)
+ def test_register_glance_image_already_exists(self,
+ mock_log):
+ """
+ Checks the proper functionality of register_glance_image
+ function when the image is local
+ """
+ image_obj = Mock()
+ image_obj.name = 'name'
+ log_calls = [call('Registering the image...'),
+ call('Image ' + image_obj.name + ' already exists.')]
+
+ self.conn.image.find_image.return_value = image_obj
+ result = self.os_sfc.register_glance_image('name',
+ 'url',
+ 'img_format',
+ 'public')
+
+ self.conn.image.find_image.assert_called_once_with(image_obj.name)
+
+ assert result is image_obj
+
+ mock_log.info.assert_has_calls(log_calls)
+
+ @patch('sfc.lib.openstack_utils.logger', autospec=True)
+ @patch("__builtin__.open", autospec=True)
def test_register_glance_image_is_local(self,
- mock_openstack_image,
- mock_image_config,
+ mock_open_fn,
mock_log):
"""
Checks the proper functionality of register_glance_image
function when the image is local
"""
+ log_calls = [call('Registering the image...'),
+ call('Image created')]
- mock_os_image_ins = mock_openstack_image.return_value
- mock_image_config_ins = mock_image_config.return_value
- log_calls = [call('Registering the image...')]
+ image_obj_None = None
+ image_obj_name = 'name'
+ image_obj = Mock()
+ mocked_file = mock_open(read_data='url').return_value
+ mock_open_fn.return_value = mocked_file
+ self.conn.image.find_image.return_value = image_obj_None
+ self.conn.image.upload_image.return_value = image_obj
result = self.os_sfc.register_glance_image('name',
'url',
'img_format',
'public')
- assert result is mock_os_image_ins
- mock_log.info.assert_has_calls(log_calls)
- mock_image_config.assert_called_once_with(name='name',
- img_format='img_format',
- image_file='url',
- public='public',
- image_user='admin')
+ assert result is image_obj
- mock_openstack_image.assert_called_with(self.os_creds,
- mock_image_config_ins)
- mock_os_image_ins.create.assert_called_once_with()
- self.assertEqual([mock_os_image_ins], self.os_sfc.creators)
+ self.conn.image.find_image.assert_called_once_with(image_obj_name)
+
+ self.conn.image.upload_image.\
+ assert_called_once_with(name='name',
+ disk_format='img_format',
+ data='url',
+ is_public='public',
+ container_format='bare')
+
+ self.assertEqual([image_obj], self.os_sfc.creators)
+ mock_log.info.assert_has_calls(log_calls)
@patch('sfc.lib.openstack_utils.logger', autospec=True)
- @patch('sfc.lib.openstack_utils.ImageConfig', autospec=True)
- @patch('sfc.lib.openstack_utils.OpenStackImage', autospec=True)
+ @patch('sfc.lib.openstack_utils.urllib2.urlopen', autospec=True)
def test_register_glance_image_is_not_local(self,
- mock_openstack_image,
- mock_image_config,
+ mock_urlopen,
mock_log):
"""
Checks the proper functionality of register_glance_image
function when the image is not local
"""
+ log_calls = [call('Registering the image...'),
+ call('Downloading image'),
+ call('Image created')]
+
+ image_obj_None = None
+ image_obj_name = 'name'
+ image_obj = Mock()
+ mock_file = Mock()
+ mock_file.read.side_effect = ['http://url']
+ mock_urlopen.return_value = mock_file
- mock_os_image_ins = mock_openstack_image.return_value
- mock_image_config_ins = mock_image_config.return_value
- log_calls = [call('Registering the image...')]
+ self.conn.image.find_image.return_value = image_obj_None
+ self.conn.image.upload_image.return_value = image_obj
result = self.os_sfc.register_glance_image('name',
'http://url',
'img_format',
'public')
- assert result is mock_os_image_ins
- mock_log.info.assert_has_calls(log_calls)
- mock_image_config.assert_called_with(name='name',
- img_format='img_format',
- url='http://url',
- public='public',
- image_user='admin')
- mock_openstack_image.assert_called_with(self.os_creds,
- mock_image_config_ins)
- mock_os_image_ins.create.assert_called_once_with()
- self.assertEqual([mock_os_image_ins], self.os_sfc.creators)
+ assert result is image_obj
+
+ self.conn.image.find_image.assert_called_once_with(image_obj_name)
+
+ self.conn.image.upload_image.\
+ assert_called_once_with(name='name',
+ disk_format='img_format',
+ data='http://url',
+ is_public='public',
+ container_format='bare')
+
+ self.assertEqual([image_obj], self.os_sfc.creators)
+ mock_log.info.assert_has_calls(log_calls)
@patch('sfc.lib.openstack_utils.logger', autospec=True)
- @patch('sfc.lib.openstack_utils.FlavorConfig', autospec=True)
- @patch('sfc.lib.openstack_utils.OpenStackFlavor', autospec=True)
def test_create_flavour(self,
- mock_openstack_flavor,
- mock_flavor_config,
mock_log):
"""
Checks the proper functionality of create_flavor
function
"""
- mock_openstack_flavor_ins = mock_openstack_flavor.return_value
- log_calls = [call('Creating the flavor...')]
+ mock_openstack_flavor_ins = self.conn.compute.\
+ create_flavor.return_value
+ log_calls = [call('Creating flavor...')]
result = self.os_sfc.create_flavor('name',
'ram',
'disk',
'vcpus')
- assert result is mock_openstack_flavor_ins.create.return_value
+ assert result is mock_openstack_flavor_ins
self.assertEqual([mock_openstack_flavor_ins],
self.os_sfc.creators)
- mock_flavor_config.assert_called_once_with(name='name',
- ram='ram',
- disk='disk',
- vcpus='vcpus')
+ self.conn.compute.create_flavor.\
+ assert_called_once_with(name='name',
+ ram='ram',
+ disk='disk',
+ vcpus='vcpus')
mock_log.info.assert_has_calls(log_calls)
@patch('sfc.lib.openstack_utils.logger', autospec=True)
@patch('sfc.lib.openstack_utils.env.get', autospec=True)
- @patch('sfc.lib.openstack_utils.RouterConfig', autospec=True)
- @patch('sfc.lib.openstack_utils.SubnetConfig', autospec=True)
- @patch('sfc.lib.openstack_utils.NetworkConfig', autospec=True)
- @patch('sfc.lib.openstack_utils.OpenStackRouter', autospec=True)
- @patch('sfc.lib.openstack_utils.OpenStackNetwork', autospec=True)
- def test_create_network_infrastructure(self,
- mock_openstack_network,
- mock_openstack_router,
- mock_network_config,
- mock_subnet_config,
- mock_router_config,
- mock_env_get,
- mock_log):
- """
- Checks the proper functionality of create_network_infrastructure
- function
- """
-
- net_ins = mock_openstack_network.return_value
- rou_ins = mock_openstack_router.return_value
- nec_ins = mock_network_config.return_value
- sub_ins = mock_subnet_config.return_value
- roc_ins = mock_router_config.return_value
- mock_env_get.return_value = 'nw_name'
- expected = (net_ins.create.return_value, rou_ins.create.return_value)
- log_calls = [call('Creating networks...'),
- call('Creating the router...')]
-
+ def test_create_network_infrastructure(self, mock_env_get, mock_log):
+ log_calls = [call('Creating Networks...'),
+ call('Creating Router...')]
+ network_obj = Mock()
+ network_obj.id = '1'
+ subnet_obj = Mock()
+ subnet_obj.id = '2'
+ ext_network_obj = Mock()
+ ext_network_obj.id = '3'
+ router_obj = Mock()
+ router_obj.id = '4'
+
+ self.conn.network.create_network.return_value = network_obj
+ self.conn.network.create_subnet.return_value = subnet_obj
+ self.conn.network.find_network.return_value = ext_network_obj
+ self.conn.network.create_router.return_value = router_obj
+ self.conn.network.get_router.return_value = router_obj
+ mock_env_get.return_value = 'ext_net_name'
+
+ expected = (network_obj, router_obj)
result = self.os_sfc.create_network_infrastructure('net_name',
'sn_name',
- 'subnet_cdir',
+ 'subnet_cidr',
'router_name')
+ self.conn.network.create_network.\
+ assert_called_once_with(name='net_name')
+ self.conn.network.create_subnet.\
+ assert_called_once_with(name='sn_name', cidr='subnet_cidr',
+ network_id=network_obj.id, ip_version='4')
+ self.conn.network.find_network.\
+ assert_called_once_with('ext_net_name')
+ self.conn.network.create_router.\
+ assert_called_once_with(name='router_name')
+ self.conn.network.add_interface_to_router.\
+ assert_called_once_with(router_obj.id, subnet_id=subnet_obj.id)
+ self.conn.network.update_router.\
+ assert_called_once_with(
+ router_obj.id,
+ external_gateway_info={'network_id': ext_network_obj.id})
+ self.conn.network.get_router.assert_called_once_with(router_obj.id)
+
self.assertEqual(expected, result)
- mock_subnet_config.assert_called_once_with(name='sn_name',
- cidr='subnet_cdir')
- mock_network_config.assert_called_once_with(name='net_name',
- subnet_settings=[sub_ins])
- mock_openstack_network.assert_called_once_with(self.os_creds,
- nec_ins)
- mock_env_get.assert_called_once_with('EXTERNAL_NETWORK')
- mock_router_config.assert_called_once_with(name='router_name',
- external_gateway='nw_name',
- internal_subnets=['sn_name']
- )
- mock_openstack_router.assert_called_once_with(self.os_creds, roc_ins)
- self.assertEqual([net_ins, rou_ins], self.os_sfc.creators)
+ self.assertEqual([network_obj, subnet_obj, router_obj],
+ self.os_sfc.creators)
mock_log.info.assert_has_calls(log_calls)
@patch('sfc.lib.openstack_utils.logger', autospec=True)
- @patch('sfc.lib.openstack_utils.Protocol', autospec=True)
- @patch('sfc.lib.openstack_utils.Direction', autospec=True)
- @patch('sfc.lib.openstack_utils.SecurityGroupConfig', autospec=True)
- @patch('sfc.lib.openstack_utils.OpenStackSecurityGroup', autospec=True)
- @patch('sfc.lib.openstack_utils.SecurityGroupRuleConfig', autospec=True)
def test_create_security_group(self,
- mock_security_group_rule_config,
- mock_openstack_security_group,
- mock_security_group_config,
- mock_direction,
- mock_protocol,
mock_log):
"""
Checks the proper functionality of create_security_group
function
"""
-
- oss_gro_ins = mock_openstack_security_group.return_value
- sec_gro_con_ins = mock_security_group_config.return_value
- mock_security_group_rule_config.side_effect = ['ping', 'ssh', 'http']
- ins = ['ping', 'ssh', 'http']
-
- sgrc_calls = [call(sec_grp_name='sec_grp_name',
- direction=mock_direction.ingress,
- protocol=mock_protocol.icmp),
- call(sec_grp_name='sec_grp_name',
- direction=mock_direction.ingress,
- protocol=mock_protocol.tcp,
- port_range_min=22,
- port_range_max=22),
- call(sec_grp_name='sec_grp_name',
- direction=mock_direction.ingress,
- protocol=mock_protocol.tcp,
- port_range_min=80,
- port_range_max=80)]
-
log_calls = [call('Creating the security groups...')]
+ sec_group_obj = Mock()
+ sec_group_obj.id = '1'
+
+ self.conn.network.create_security_group.return_value = sec_group_obj
result = self.os_sfc.create_security_group('sec_grp_name')
- assert result is oss_gro_ins.create.return_value
- self.assertEqual([oss_gro_ins], self.os_sfc.creators)
- mock_security_group_config.assert_called_once_with(name='sec_grp_name',
- rule_settings=ins)
- mock_security_group_rule_config.assert_has_calls(sgrc_calls)
- mock_openstack_security_group.assert_called_once_with(self.os_creds,
- sec_gro_con_ins)
- oss_gro_ins.create.assert_called_with()
+ assert result is sec_group_obj
+
+ self.conn.network.create_security_group.\
+ assert_called_once_with(name='sec_grp_name')
+
+ pc_calls = [call(security_group_id=sec_group_obj.id,
+ direction='ingress',
+ protocol='icmp'),
+ call(security_group_id=sec_group_obj.id,
+ direction='ingress',
+ protocol='tcp',
+ port_range_min=22,
+ port_range_max=22),
+ call(security_group_id=sec_group_obj.id,
+ direction='ingress',
+ protocol='tcp',
+ port_range_min=80,
+ port_range_max=80)]
+
+ self.conn.network.create_security_group_rule.\
+ assert_has_calls(pc_calls)
+
+ self.assertEqual([sec_group_obj], self.os_sfc.creators)
mock_log.info.assert_has_calls(log_calls)
@patch('sfc.lib.openstack_utils.logger', autospec=True)
- @patch('sfc.lib.openstack_utils.PortConfig', autospec=True)
- @patch('sfc.lib.openstack_utils.VmInstanceConfig', autospec=True)
- @patch('sfc.lib.openstack_utils.cr_inst.OpenStackVmInstance',
- autospec=True)
def test_create_instance_port_security_false(self,
- mock_os_vm_instance,
- mock_vm_instance_config,
- mock_port_config,
mock_log):
"""
Checks the proper functionality of create_instance
function
"""
- vm_con_ins = mock_vm_instance_config.return_value
- pc_inss = ['pc_config1', 'pc_config2']
- mock_port_config.side_effect = pc_inss
- os_vm_ins = mock_os_vm_instance.return_value
- os_vm_ins_cre = os_vm_ins.create.return_value
- expected = (os_vm_ins_cre, os_vm_ins)
+ keypair_obj = Mock()
+ keypair_obj.name = 'keypair_name'
+ flavor_obj = Mock()
+ flavor_obj.id = '1'
+ port_obj1 = Mock()
+ port_obj1.id = '2'
+ port_obj2 = Mock()
+ port_obj2.id = '3'
+ instance_obj = Mock()
+ instance_obj.name = 'instance_name'
secgrp = Mock()
secgrp.name = 'sec_grp'
- network = Mock()
- network.name = 'nw_name'
+ secgrp.id = '4'
img_cre = Mock()
- img_cre.image_settings = 'image_settings'
-
- log_calls = [call('Creating the instance vm_name...')]
- pc_calls = [call(name='port1',
- network_name='nw_name',
- port_security_enabled=False),
- call(name='port2',
- network_name='nw_name',
- port_security_enabled=False)]
+ img_cre.id = '5'
+ network = Mock()
+ network.id = '6'
+ ports = ['port1', 'port2']
+ port_security = False
+
+ log_calls = [call('Creating Key Pair vm_name...'),
+ call('Creating Port ' + str(ports) + '...'),
+ call('Creating the instance vm_name...'),
+ call('Waiting for instance_name to become Active'),
+ call('instance_name is active')]
+
+ self.conn.compute.create_keypair.return_value = keypair_obj
+ self.conn.compute.find_flavor.return_value = flavor_obj
+ self.conn.network.create_port.side_effect = [port_obj1, port_obj2]
+ self.conn.compute.create_server.return_value = instance_obj
+
+ port_obj_list = [port_obj1, port_obj2]
+
+ expected = (instance_obj, port_obj_list)
result = self.os_sfc.create_instance('vm_name',
'flavor_name',
img_cre,
network,
secgrp,
'av_zone',
- ['port1', 'port2'],
- port_security=False)
+ ports,
+ port_security=port_security)
self.assertEqual(expected, result)
- mock_vm_instance_config.assert_called_once_with(name='vm_name',
- flavor='flavor_name',
- port_settings=pc_inss,
- availability_zone='av'
- '_zone')
- mock_os_vm_instance.assert_called_once_with(self.os_creds,
- vm_con_ins,
- 'image_settings')
- self.assertEqual([os_vm_ins], self.os_sfc.creators)
+
+ pc_calls = [call(name=ports[0],
+ is_port_security_enabled=port_security,
+ network_id=network.id),
+ call(name=ports[1],
+ is_port_security_enabled=port_security,
+ network_id=network.id)]
+
+ self.conn.compute.create_keypair.\
+ assert_called_once_with(name='vm_name' + "_keypair")
+
+ self.conn.compute.find_flavor.assert_called_once_with('flavor_name')
+
+ self.conn.network.create_port.\
+ assert_has_calls(pc_calls)
+
+ self.conn.compute.create_server.\
+ assert_called_once_with(name='vm_name',
+ image_id=img_cre.id,
+ flavor_id=flavor_obj.id,
+ networks=[{"port": port_obj1.id},
+ {"port": port_obj2.id}],
+ key_name=keypair_obj.name,
+ availability_zone='av_zone')
+
+ self.conn.compute.wait_for_server.\
+ assert_called_once_with(instance_obj)
+
+ self.assertEqual([keypair_obj, port_obj1, port_obj2, instance_obj],
+ self.os_sfc.creators)
mock_log.info.assert_has_calls(log_calls)
- mock_port_config.assert_has_calls(pc_calls)
@patch('sfc.lib.openstack_utils.logger', autospec=True)
- @patch('sfc.lib.openstack_utils.PortConfig', autospec=True)
- @patch('sfc.lib.openstack_utils.VmInstanceConfig', autospec=True)
- @patch('sfc.lib.openstack_utils.cr_inst.OpenStackVmInstance',
- autospec=True)
def test_create_instance(self,
- mock_os_vm_instance,
- mock_vm_instance_config,
- mock_port_config,
mock_log):
"""
Checks the proper functionality of create_instance
function
"""
- vm_con_ins = mock_vm_instance_config.return_value
- pc_inss = ['pc_config1', 'pc_config2']
- mock_port_config.side_effect = pc_inss
- os_vm_ins = mock_os_vm_instance.return_value
- os_vm_ins_cre = os_vm_ins.create.return_value
- expected = (os_vm_ins_cre, os_vm_ins)
+ keypair_obj = Mock()
+ keypair_obj.name = 'keypair_name'
+ flavor_obj = Mock()
+ flavor_obj.id = '1'
+ port_obj = Mock()
+ port_obj.id = '2'
+ instance_obj = Mock()
+ instance_obj.name = 'instance_name'
secgrp = Mock()
secgrp.name = 'sec_grp'
+ secgrp.id = '4'
+ img_cre = Mock()
+ img_cre.id = '5'
network = Mock()
- network.name = 'nw_name'
+ network.id = '6'
+ ports = ['port1']
+ port_obj_list = [port_obj]
+ port_security = True
+
+ log_calls = [call('Creating Key Pair vm_name...'),
+ call('Creating Port ' + str(ports) + '...'),
+ call('Creating the instance vm_name...'),
+ call('Waiting for instance_name to become Active'),
+ call('instance_name is active')]
+
+ self.conn.compute.create_keypair.return_value = keypair_obj
+ self.conn.compute.find_flavor.return_value = flavor_obj
+ self.conn.network.create_port.return_value = port_obj
+ self.conn.compute.create_server.return_value = instance_obj
+ # self.conn.compute.wait_for_server.return_value = wait_ins_obj
+
+ expected = (instance_obj, port_obj_list)
+ result = self.os_sfc.create_instance('vm_name',
+ 'flavor_name',
+ img_cre,
+ network,
+ secgrp,
+ 'av_zone',
+ ports,
+ port_security=port_security)
+ self.assertEqual(expected, result)
+
+ pc_calls = [call(name=ports[0],
+ is_port_security_enabled=port_security,
+ network_id=network.id,
+ security_group_ids=[secgrp.id])]
+
+ self.conn.compute.create_keypair.\
+ assert_called_once_with(name='vm_name' + "_keypair")
+
+ self.conn.compute.find_flavor.assert_called_once_with('flavor_name')
+
+ self.conn.network.create_port.\
+ assert_has_calls(pc_calls)
+
+ self.conn.compute.create_server.\
+ assert_called_once_with(name='vm_name',
+ image_id=img_cre.id,
+ flavor_id=flavor_obj.id,
+ networks=[{"port": port_obj.id}],
+ key_name=keypair_obj.name,
+ availability_zone='av_zone')
+
+ self.conn.compute.wait_for_server.\
+ assert_called_once_with(instance_obj)
+
+ self.assertEqual([keypair_obj, port_obj, instance_obj],
+ self.os_sfc.creators)
+ mock_log.info.assert_has_calls(log_calls)
+
+ @patch('sfc.lib.openstack_utils.logger', autospec=True)
+ def test_create_instance_port_security_false_one_port(self,
+ mock_log):
+ """
+ Checks the proper functionality of create_instance
+ function
+ """
+
+ keypair_obj = Mock()
+ keypair_obj.name = 'keypair_name'
+ flavor_obj = Mock()
+ flavor_obj.id = '1'
+ port_obj = Mock()
+ port_obj.id = '2'
+ instance_obj = Mock()
+ instance_obj.name = 'instance_name'
+ secgrp = Mock()
+ secgrp.name = 'sec_grp'
+ secgrp.id = '4'
img_cre = Mock()
- img_cre.image_settings = 'image_settings'
-
- log_calls = [call('Creating the instance vm_name...')]
- pc_calls = [call(name='port1',
- network_name='nw_name',
- port_security_enabled=True),
- call(name='port2',
- network_name='nw_name',
- port_security_enabled=True)]
+ img_cre.id = '5'
+ network = Mock()
+ network.id = '6'
+ ports = ['port1']
+ port_obj_list = [port_obj]
+ port_security = False
+
+ log_calls = [call('Creating Key Pair vm_name...'),
+ call('Creating Port ' + str(ports) + '...'),
+ call('Creating the instance vm_name...'),
+ call('Waiting for instance_name to become Active'),
+ call('instance_name is active')]
+
+ self.conn.compute.create_keypair.return_value = keypair_obj
+ self.conn.compute.find_flavor.return_value = flavor_obj
+ self.conn.network.create_port.return_value = port_obj
+ self.conn.compute.create_server.return_value = instance_obj
+
+ expected = (instance_obj, port_obj_list)
result = self.os_sfc.create_instance('vm_name',
'flavor_name',
img_cre,
network,
secgrp,
'av_zone',
- ['port1', 'port2'])
+ ports,
+ port_security=port_security)
self.assertEqual(expected, result)
- mock_vm_instance_config.assert_called_once_with(name='vm_name',
- flavor='flavor_name',
- security_group_names=''
- 'sec_grp',
- port_settings=pc_inss,
- availability_zone='av'
- '_zone')
- mock_os_vm_instance.assert_called_once_with(self.os_creds,
- vm_con_ins,
- 'image_settings')
- self.assertEqual([os_vm_ins], self.os_sfc.creators)
+
+ pc_calls = [call(name=ports[0],
+ is_port_security_enabled=port_security,
+ network_id=network.id)]
+
+ self.conn.compute.create_keypair.\
+ assert_called_once_with(name='vm_name' + "_keypair")
+
+ self.conn.compute.find_flavor.assert_called_once_with('flavor_name')
+
+ self.conn.network.create_port.\
+ assert_has_calls(pc_calls)
+
+ self.conn.compute.create_server.\
+ assert_called_once_with(name='vm_name',
+ image_id=img_cre.id,
+ flavor_id=flavor_obj.id,
+ networks=[{"port": port_obj.id}],
+ key_name=keypair_obj.name,
+ availability_zone='av_zone')
+
+ self.conn.compute.wait_for_server.\
+ assert_called_once_with(instance_obj)
+
+ self.assertEqual([keypair_obj, port_obj, instance_obj],
+ self.os_sfc.creators)
mock_log.info.assert_has_calls(log_calls)
- mock_port_config.assert_has_calls(pc_calls)
- @patch('sfc.lib.openstack_utils.nova_utils.get_hypervisor_hosts',
- autospec=True)
- def test_get_av_zones(self, mock_get_hypervisor_hosts):
+ @patch.object(os_sfc_utils.OpenStackSFC, 'get_hypervisor_hosts')
+ def test_get_av_zones(self, mock_hosts):
"""
Checks the proper functionality of get_av_zone
function
"""
-
- mock_get_hypervisor_hosts.return_value = ['host1', 'host2']
+ mock_hosts.return_value = ['host1', 'host2']
result = self.os_sfc.get_av_zones()
- mock_get_hypervisor_hosts.assert_called_once_with(self.nova)
+ mock_hosts.assert_called_once()
self.assertEqual(['nova::host1', 'nova::host2'], result)
+ def test_get_hypervisor_hosts(self):
+ """
+ Checks the proper functionality of get_av_zone
+ function
+ """
+ from openstack.compute.v2 import hypervisor
+
+ hypervisor1 = Mock()
+ hypervisor1.state = 'up'
+ hypervisor1.name = 'compute00'
+ hypervisor2 = Mock()
+ hypervisor2.state = 'up'
+ hypervisor2.name = 'compute01'
+ nodes = [hypervisor1.name, hypervisor2.name]
+ hypervisors_list = MagicMock()
+ mock_obj = patch.object(hypervisor, 'Hypervisor')
+ mock_obj.side_effect = [hypervisor1, hypervisor2]
+ self.conn.compute.hypervisors.return_value = hypervisors_list
+ hypervisors_list.__iter__.return_value = [hypervisor1, hypervisor2]
+
+ result = self.os_sfc.get_hypervisor_hosts()
+ self.conn.compute.hypervisors.assert_called_once()
+ self.assertEqual(nodes, result)
+
@patch('sfc.lib.openstack_utils.OpenStackSFC.get_vm_compute',
autospec=True, return_value='mock_client')
def test_compute_client(self, mock_get_vm_compute):
@@ -427,40 +606,93 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
result = self.os_sfc.get_vm_compute('dev_vm')
self.assertEqual('mock_host', result)
+ @patch('sfc.lib.openstack_utils.logger', autospec=True)
@patch('sfc.lib.openstack_utils.cr_inst.OpenStackVmInstance',
autospec=True)
- @patch('sfc.lib.openstack_utils.FloatingIpConfig', autospec=True)
- def test_assign_floating_ip(self,
- mock_floating_ip_config,
- mock_os_vm):
+ def test_get_instance_port_raised_exceptioin(self,
+ mock_os_vm,
+ mock_log):
"""
- Checks the proper functionality of assigning_floating_ip
- function
+ Checks the proper functionality of get_client_port
+ function when no port is returned
"""
- mock_router = Mock()
+ mock_os_vm_ins = mock_os_vm.return_value
mock_vm = Mock()
- mock_ip = Mock()
- mock_ip.ip = 'mock_ip'
- mock_router.name = 'mock_router_name'
mock_vm.name = 'mock_vm_name'
- port_1 = port_2 = Mock()
- port_1.name = 'port_1'
- mock_vm.ports = [port_1, port_2]
+ mock_os_vm_ins.get_port_by_name.return_value = None
+ ErrorMSG = 'Client VM does not have the desired port'
+ log_calls = [call("The VM mock_vm_name does not have any port"
+ " with name mock_vm_name-port")]
+
+ with self.assertRaises(Exception) as cm:
+ self.os_sfc.get_instance_port(mock_vm, mock_os_vm_ins)
+
+ self.assertEqual(cm.exception.message, ErrorMSG)
+ mock_log.error.assert_has_calls(log_calls)
+
+ @patch('sfc.lib.openstack_utils.logger', autospec=True)
+ @patch('sfc.lib.openstack_utils.cr_inst.OpenStackVmInstance',
+ autospec=True)
+ def test_get_instance_port(self,
+ mock_os_vm,
+ mock_log):
+ """
+ Checks the proper functionality of get_client_port
+ function when no port is returned
+ """
- flo_ip_ins = mock_floating_ip_config.return_value
mock_os_vm_ins = mock_os_vm.return_value
- mock_os_vm_ins.add_floating_ip.return_value = mock_ip
- result = self.os_sfc.assign_floating_ip(mock_router,
- mock_vm,
- mock_os_vm_ins)
- self.assertEqual('mock_ip', result)
- mock_floating_ip_config.assert_called_once_with(name='mock_vm_name'
- '-float',
- port_name='port_1',
- router_name='mock_'
- 'router_name')
- mock_os_vm_ins.add_floating_ip.assert_called_once_with(flo_ip_ins)
+ mock_vm = Mock()
+ mock_vm.name = 'mock_vm_name'
+ mock_os_vm_ins.get_port_by_name.return_value = 'mock_port'
+ result = self.os_sfc.get_instance_port(mock_vm, mock_os_vm_ins)
+ self.assertEqual('mock_port', result)
+
+ @patch('sfc.lib.openstack_utils.logger', autospec=True)
+ @patch('sfc.lib.openstack_utils.env.get', autospec=True)
+ def test_assign_floating_ip(self,
+ mock_env_get,
+ mock_log):
+ """
+ Checks the proper functionality of assigning_floating_ip
+ function
+ """
+ ext_network_obj = Mock()
+ ext_network_obj.id = '1'
+ fip_obj = Mock()
+ fip_obj.floating_ip_address = 'floating_ip_address'
+ port_obj = Mock()
+ port_obj.id = '2'
+ instance_obj = Mock()
+ instance_obj.id = '3'
+
+ log_calls = [call(' Creating floating ips '),
+ call(' FLoating IP address '
+ + fip_obj.floating_ip_address
+ + ' created'),
+ call(' Adding Floating IPs to instances ')]
+
+ mock_env_get.return_value = 'ext_net_name'
+ self.conn.network.find_network.return_value = ext_network_obj
+ self.conn.network.create_ip.return_value = fip_obj
+ self.conn.netwotk.get_port.return_value = port_obj
+ self.conn.compute.get_server.return_value = instance_obj
+
+ result = self.os_sfc.assign_floating_ip(instance_obj, port_obj)
+ assert result is fip_obj.floating_ip_address
+
+ self.conn.network.find_network.assert_called_once_with('ext_net_name')
+ self.conn.network.create_ip.\
+ assert_called_once_with(floating_network_id=ext_network_obj.id,
+ port_id=port_obj.id)
+ self.conn.compute.add_floating_ip_to_server.\
+ assert_called_once_with(instance_obj.id,
+ fip_obj.floating_ip_address)
+
+ self.assertEqual([fip_obj],
+ self.os_sfc.creators)
+ mock_log.info.assert_has_calls(log_calls)
@patch('sfc.lib.openstack_utils.logger', autospec=True)
@patch('sfc.lib.openstack_utils.heat_utils.get_stack_servers',
@@ -494,7 +726,7 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
self.assertEqual(cm.exception.message, ErrorMSG)
mock_get_stack_servers.assert_called_once_with(self.heat,
self.nova,
- self.neutron,
+ self.neutron_client,
self.keystone,
'stack_obj',
'admin')
@@ -533,7 +765,7 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
mock_get_stack_servers.assert_called_once_with(self.heat,
self.nova,
- self.neutron,
+ self.neutron_client,
self.keystone,
'stack_obj',
'admin')
@@ -582,7 +814,7 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
self.assertEqual([mock_os_vm_ins], self.os_sfc.creators)
mock_get_stack_servers.assert_called_once_with(self.heat,
self.nova,
- self.neutron,
+ self.neutron_client,
self.keystone,
'stack_obj', 'admin')
mock_generate_creator.assert_called_once_with(self.os_creds,
@@ -595,119 +827,73 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
mock_os_vm_ins.add_floating_ip.assert_called_once_with(float_ip_ins)
@patch('sfc.lib.openstack_utils.logger', autospec=True)
- @patch('sfc.lib.openstack_utils.cr_inst.OpenStackVmInstance',
- autospec=True)
- def test_get_instance_port_raised_exceptioin(self,
- mock_os_vm,
- mock_log):
- """
- Checks the proper functionality of get_client_port
- function when no port is returned
- """
-
- mock_os_vm_ins = mock_os_vm.return_value
- mock_vm = Mock()
- mock_vm.name = 'mock_vm_name'
- mock_os_vm_ins.get_port_by_name.return_value = None
- ErrorMSG = 'Client VM does not have the desired port'
- log_calls = [call("The VM mock_vm_name does not have any port"
- " with name mock_vm_name-port")]
-
- with self.assertRaises(Exception) as cm:
- self.os_sfc.get_instance_port(mock_vm, mock_os_vm_ins)
-
- self.assertEqual(cm.exception.message, ErrorMSG)
- mock_log.error.assert_has_calls(log_calls)
-
- @patch('sfc.lib.openstack_utils.logger', autospec=True)
- @patch('sfc.lib.openstack_utils.cr_inst.OpenStackVmInstance',
- autospec=True)
- def test_get_instance_port(self,
- mock_os_vm,
- mock_log):
- """
- Checks the proper functionality of get_client_port
- function when no port is returned
- """
-
- mock_os_vm_ins = mock_os_vm.return_value
- mock_vm = Mock()
- mock_vm.name = 'mock_vm_name'
- mock_os_vm_ins.get_port_by_name.return_value = 'mock_port'
- result = self.os_sfc.get_instance_port(mock_vm, mock_os_vm_ins)
- self.assertEqual('mock_port', result)
-
- @patch('sfc.lib.openstack_utils.neutron_utils.list_security_groups',
- autospec=True)
- @patch('sfc.lib.openstack_utils.neutron_utils.delete_security_group',
- autospec=True)
- def test_delete_all_security_groups(self,
- mock_delete_security_group,
- mock_list_security_groups):
+ def test_delete_all_security_groups(self, mock_log):
"""
Checks the proper functionality of delete_all_security_groups
function
"""
- del_calls = [call(self.neutron, 'sec_group_1'),
- call(self.neutron, 'sec_group_2')]
- mock_list_security_groups.return_value = ['sec_group_1', 'sec_group_2']
- self.os_sfc.delete_all_security_groups()
- mock_list_security_groups.assert_called_once_with(self.neutron)
- mock_delete_security_group.assert_has_calls(del_calls)
+ log_calls_info = [call('Deleting remaining security groups...')]
+ secgrp1_obj = Mock()
+ secgrp2_obj = Mock()
+ secgrp_list = MagicMock()
- @patch('snaps.openstack.create_instance.OpenStackVmInstance',
- autospec=True)
- def test_wait_for_vnf(self, mock_osvminstance):
- """
- Checks the method wait_for_vnf
- """
+ self.conn.network.create_security_groups.side_effect = [secgrp1_obj,
+ secgrp2_obj]
+ self.conn.network.security_groups.return_value = secgrp_list
- mock_osvm_ins = mock_osvminstance.return_value
- mock_osvm_ins.vm_active.return_value = True
- result = self.os_sfc.wait_for_vnf(mock_osvm_ins)
- self.assertTrue(result)
+ secgrp_list.__iter__.return_value = [secgrp1_obj, secgrp2_obj]
+ del_calls = [call(secgrp1_obj),
+ call(secgrp2_obj)]
+
+ self.os_sfc.delete_all_security_groups()
+ self.conn.network.security_groups.assert_called_once()
+ self.conn.network.delete_security_group.assert_has_calls(del_calls)
+ mock_log.info.assert_has_calls(log_calls_info)
- @patch('snaps.domain.vm_inst.VmInst', autospec=True)
@patch('sfc.lib.openstack_utils.logger', autospec=True)
- def test_create_port_groups_raises_exception(self, mock_log, mock_vm):
+ def test_create_port_groups_raises_exception(self, mock_log):
"""
Checks the create_port_groups when length of ports is greater than 2
"""
- mock_vm_ins = mock_vm.return_value
- mock_vm_ins.name = 'vm'
- log_calls_info = [call('Creating the port pairs for vm')]
+ instance_obj = Mock()
+ instance_obj.name = 'name'
+ self.conn.compute.get_server.return_value = instance_obj
+
+ log_calls_info = [call('Creating the port pairs...')]
log_calls_err = [call('Only SFs with one or two ports are supported')]
exception_message = "Failed to create port pairs"
vnf_ports = ['p1', 'p2', 'p3']
with self.assertRaises(Exception) as cm:
- self.os_sfc.create_port_groups(vnf_ports, mock_vm_ins)
+ self.os_sfc.create_port_groups(vnf_ports, instance_obj)
self.assertEqual(exception_message, cm.exception.message)
mock_log.info.assert_has_calls(log_calls_info)
mock_log.error.assert_has_calls(log_calls_err)
- @patch('snaps.domain.network.Port', autospec=True)
- @patch('snaps.domain.vm_inst.VmInst', autospec=True)
@patch('sfc.lib.openstack_utils.logger', autospec=True)
- def test_create_port_groups_returns_none_from_pp(self, mock_log,
- mock_vm,
- mock_port):
+ def test_create_port_groups_returns_none_from_pp(self, mock_log):
"""
Checks the create_port_groups when something goes wrong in port pair
creation
"""
- mock_vm_ins = mock_vm.return_value
- mock_vm_ins.name = 'vm'
- log_calls_info = [call('Creating the port pairs for vm')]
+ instance_obj = Mock()
+ instance_obj.name = 'name'
+ port_obj1 = Mock()
+ port_obj2 = Mock()
+ port_obj1.id = '123abc'
+ port_obj2.id = '456def'
+
+ self.conn.compute.get_server.return_value = instance_obj
+ self.conn.network.get_port.return_value = port_obj1
+ self.conn.network.get_port.return_value = port_obj2
+
+ log_calls_info = [call('Creating the port pairs...')]
log_calls_warn = [call('Chain creation failed due to port pair '
- 'creation failed for vnf %(vnf)s', {'vnf': 'vm'})]
- mock_port_ins = mock_port.return_value
- mock_port_ins2 = mock_port.return_value
- mock_port_ins.id = '123abc'
- mock_port_ins2.id = '456def'
- self.neutron.create_sfc_port_pair.return_value = None
+ 'creation failed for vnf %(vnf)s',
+ {'vnf': instance_obj.name})]
+ self.neutron_client.create_sfc_port_pair.return_value = None
result = self.os_sfc.create_port_groups(
- [mock_port_ins, mock_port_ins2], mock_vm_ins)
+ [port_obj1, port_obj2], instance_obj)
self.assertIsNone(result)
mock_log.info.assert_has_calls(log_calls_info)
mock_log.warning.assert_has_calls(log_calls_warn)
@@ -721,7 +907,7 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
Checks the create_port_groups when openstack does not commit the pp
"""
- log_calls_info = [call('Creating the port pairs for vm')]
+ log_calls_info = [call('Creating the port pairs...')]
mock_port_ins = mock_port.return_value
mock_port_ins.id = '123abc'
mock_vm_ins = mock_osvm.return_value
@@ -731,14 +917,14 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
'description': 'port pair for vm',
'ingress': '123abc',
'egress': '123abc'}
- self.neutron.create_sfc_port_pair.return_value = \
+ self.neutron_client.create_sfc_port_pair.return_value = \
{'port_pair': {'id': 'pp_id'}}
- self.neutron.list_sfc_port_pairs.return_value = \
+ self.neutron_client.list_sfc_port_pairs.return_value = \
{'port_pairs': [{'id': 'xxxx'}]}
with self.assertRaises(Exception) as cm:
self.os_sfc.create_port_groups([mock_port_ins], mock_vm_ins)
self.assertEqual(exception_message, cm.exception.message)
- self.neutron.create_sfc_port_pair.assert_has_calls(
+ self.neutron_client.create_sfc_port_pair.assert_has_calls(
[call({'port_pair': expected_port_pair})])
mock_log.info.assert_has_calls(log_calls_info)
@@ -752,57 +938,63 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
Checks the create_port_groups when something goes wrong in port pair
group creation
"""
- mock_vm_ins = mock_vm.return_value
- mock_vm_ins.name = 'vm'
- log_calls_info = [call('Creating the port pairs for vm'),
- call('Creating the port pair groups for vm')]
+
+ instance_obj = Mock()
+ instance_obj.name = 'name'
+ port_obj = Mock()
+ port_obj.id = '123abc'
+
+ self.conn.compute.get_server.return_value = instance_obj
+ self.conn.network.get_port.return_value = port_obj
+
+ log_calls_info = [call('Creating the port pairs...'),
+ call('Creating the port pair groups for name')]
log_calls_warn = [call('Chain creation failed due to port pair group '
- 'creation failed for vnf %(vnf)', 'vm')]
- mock_port_ins = mock_port.return_value
- mock_port_ins.id = '123abc'
- self.neutron.create_sfc_port_pair.return_value = \
+ 'creation failed for vnf %(vnf)',
+ instance_obj.name)]
+ self.neutron_client.create_sfc_port_pair.return_value = \
{'port_pair': {'id': 'pp_id'}}
- self.neutron.list_sfc_port_pairs.return_value = \
+ self.neutron_client.list_sfc_port_pairs.return_value = \
{'port_pairs': [{'id': 'pp_id'}]}
- self.neutron.create_sfc_port_pair_group.return_value = None
- result = self.os_sfc.create_port_groups([mock_port_ins], mock_vm_ins)
+ self.neutron_client.create_sfc_port_pair_group.return_value = None
+ result = self.os_sfc.create_port_groups([port_obj], instance_obj)
self.assertIsNone(result)
mock_log.info.assert_has_calls(log_calls_info)
mock_log.warning.assert_has_calls(log_calls_warn)
- @patch('snaps.domain.network.Port', autospec=True)
- @patch('snaps.domain.vm_inst.VmInst', autospec=True)
@patch('sfc.lib.openstack_utils.logger', autospec=True)
- def test_create_port_groups_returns_id(self, mock_log, mock_osvm,
- mock_port):
+ def test_create_port_groups_returns_id(self, mock_log):
"""
Checks the create_port_groups when everything goes as expected
"""
- log_calls_info = [call('Creating the port pairs for vm'),
- call('Creating the port pair groups for vm')]
- mock_port_ins = mock_port.return_value
- mock_port_ins.id = '123abc'
- mock_osvm_ins = mock_osvm.return_value
- mock_osvm_ins.name = 'vm'
+ log_calls_info = [call('Creating the port pairs...')]
+
+ instance_obj = Mock()
+ instance_obj.name = 'vm'
+ port_obj = Mock()
+ port_obj.id = '123abc'
+ self.conn.compute.get_server.return_value = instance_obj
+ self.conn.network.get_port.return_value = port_obj
+
expected_port_pair = {'name': 'vm-connection-points',
'description': 'port pair for vm',
'ingress': '123abc',
'egress': '123abc'}
- self.neutron.create_sfc_port_pair.return_value = \
+ self.neutron_client.create_sfc_port_pair.return_value = \
{'port_pair': {'id': 'pp_id'}}
- self.neutron.list_sfc_port_pairs.return_value = \
+ self.neutron_client.list_sfc_port_pairs.return_value = \
{'port_pairs': [{'id': 'pp_id'}]}
- self.neutron.create_sfc_port_pair_group.return_value = \
+ self.neutron_client.create_sfc_port_pair_group.return_value = \
{'port_pair_group': {'id': 'pp_id'}}
expected_port_pair_gr = {'name': 'vm-port-pair-group',
'description': 'port pair group for vm',
'port_pairs': ['pp_id']}
- self.os_sfc.create_port_groups([mock_port_ins], mock_osvm_ins)
- self.neutron.create_sfc_port_pair.assert_has_calls(
+ self.os_sfc.create_port_groups([port_obj], instance_obj)
+ self.neutron_client.create_sfc_port_pair.assert_has_calls(
[call({'port_pair': expected_port_pair})])
- self.neutron.create_sfc_port_pair_group.assert_has_calls(
+ self.neutron_client.create_sfc_port_pair_group.assert_has_calls(
[call({'port_pair_group': expected_port_pair_gr})])
mock_log.info.assert_has_calls(log_calls_info)
@@ -820,9 +1012,9 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
protocol = 'tcp'
vnffg_name = 'red_http'
symmetrical = False
- self.neutron.create_sfc_flow_classifier.return_value = \
+ self.neutron_client.create_sfc_flow_classifier.return_value = \
{'flow_classifier': {'id': 'fc_id'}}
- self.neutron.create_sfc_port_chain.return_value = \
+ self.neutron_client.create_sfc_port_chain.return_value = \
{'port_chain': {'id': 'pc_id'}}
expected_sfc_classifier_params = {'name': vnffg_name + '-classifier',
@@ -838,9 +1030,9 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
self.os_sfc.create_chain(port_groups, neutron_port, port,
protocol, vnffg_name, symmetrical)
- self.neutron.create_sfc_flow_classifier.assert_has_calls(
+ self.neutron_client.create_sfc_flow_classifier.assert_has_calls(
[call({'flow_classifier': expected_sfc_classifier_params})])
- self.neutron.create_sfc_port_chain.assert_has_calls(
+ self.neutron_client.create_sfc_port_chain.assert_has_calls(
[call({'port_chain': expected_chain_config})])
mock_log.info.assert_has_calls(log_calls)
@@ -860,9 +1052,9 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
symmetrical = True
serv_p = '123abc'
server_ip = '1.1.1.2'
- self.neutron.create_sfc_flow_classifier.return_value = \
+ self.neutron_client.create_sfc_flow_classifier.return_value = \
{'flow_classifier': {'id': 'fc_id'}}
- self.neutron.create_sfc_port_chain.return_value = \
+ self.neutron_client.create_sfc_port_chain.return_value = \
{'port_chain': {'id': 'pc_id'}}
expected_sfc_classifier_params = {'name': vnffg_name + '-classifier',
@@ -882,9 +1074,9 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
protocol, vnffg_name, symmetrical,
server_port=serv_p, server_ip=server_ip)
- self.neutron.create_sfc_flow_classifier.assert_has_calls(
+ self.neutron_client.create_sfc_flow_classifier.assert_has_calls(
[call({'flow_classifier': expected_sfc_classifier_params})])
- self.neutron.create_sfc_port_chain.assert_has_calls(
+ self.neutron_client.create_sfc_port_chain.assert_has_calls(
[call({'port_chain': expected_chain_config})])
mock_log.info.assert_has_calls(log_calls)
@@ -895,15 +1087,15 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
"""
log_calls = [call('Deleting the port groups...'),
call('Deleting the port pairs...')]
- self.neutron.list_sfc_port_pair_groups.return_value = \
+ self.neutron_client.list_sfc_port_pair_groups.return_value = \
{'port_pair_groups': [{'id': 'id_ppg1'}, {'id': 'id_ppg2'}]}
- self.neutron.list_sfc_port_pairs.return_value = \
+ self.neutron_client.list_sfc_port_pairs.return_value = \
{'port_pairs': [{'id': 'id_pp1'}, {'id': 'id_pp2'}]}
self.os_sfc.delete_port_groups()
- self.neutron.delete_sfc_port_pair_group.assert_has_calls(
+ self.neutron_client.delete_sfc_port_pair_group.assert_has_calls(
[call('id_ppg1'), call('id_ppg2')])
- self.neutron.delete_sfc_port_pair.assert_has_calls(
+ self.neutron_client.delete_sfc_port_pair.assert_has_calls(
[call('id_pp1'), call('id_pp2')])
mock_log.info.assert_has_calls(log_calls)
@@ -914,14 +1106,15 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
"""
log_calls = [call('Deleting the chain...'),
call('Deleting the classifiers...')]
- self.neutron.list_sfc_port_chains.return_value = \
+ self.neutron_client.list_sfc_port_chains.return_value = \
{'port_chains': [{'id': 'id_pc1'}]}
- self.neutron.list_sfc_flow_classifiers.return_value = \
+ self.neutron_client.list_sfc_flow_classifiers.return_value = \
{'flow_classifiers': [{'id': 'id_fc1'}]}
self.os_sfc.delete_chain()
- self.neutron.delete_sfc_port_chain.assert_has_calls([call('id_pc1')])
- self.neutron.delete_sfc_flow_classifier.assert_has_calls(
+ self.neutron_client.delete_sfc_port_chain.\
+ assert_has_calls([call('id_pc1')])
+ self.neutron_client.delete_sfc_flow_classifier.assert_has_calls(
[call('id_fc1')])
mock_log.info.assert_has_calls(log_calls)