aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--tests/unit/benchmark/scenarios/lib/test_create_floating_ip.py56
-rw-r--r--yardstick/benchmark/scenarios/lib/create_floating_ip.py26
-rw-r--r--yardstick/common/openstack_utils.py147
-rw-r--r--yardstick/tests/unit/common/test_openstack_utils.py28
4 files changed, 142 insertions, 115 deletions
diff --git a/tests/unit/benchmark/scenarios/lib/test_create_floating_ip.py b/tests/unit/benchmark/scenarios/lib/test_create_floating_ip.py
index 72dbcd7cd..a7286f5da 100644
--- a/tests/unit/benchmark/scenarios/lib/test_create_floating_ip.py
+++ b/tests/unit/benchmark/scenarios/lib/test_create_floating_ip.py
@@ -6,29 +6,53 @@
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
+
import unittest
import mock
-from yardstick.benchmark.scenarios.lib.create_floating_ip import CreateFloatingIp
+from yardstick.benchmark.scenarios.lib import create_floating_ip
+import yardstick.common.openstack_utils as op_utils
class CreateFloatingIpTestCase(unittest.TestCase):
- @mock.patch('yardstick.common.openstack_utils.create_floating_ip')
- @mock.patch('yardstick.common.openstack_utils.get_network_id')
- @mock.patch('yardstick.common.openstack_utils.get_neutron_client')
- def test_create_floating_ip(self, mock_create_floating_ip, mock_get_network_id, mock_get_neutron_client):
- options = {}
- args = {"options": options}
- obj = CreateFloatingIp(args, {})
- obj.run({})
- self.assertTrue(mock_create_floating_ip.called)
- self.assertTrue(mock_get_network_id.called)
- self.assertTrue(mock_get_neutron_client.called)
+ def setUp(self):
+ self._mock_get_network_id = mock.patch.object(
+ op_utils, 'get_network_id')
+ self.mock_get_network_id = self._mock_get_network_id.start()
+ self._mock_create_floating_ip = mock.patch.object(
+ op_utils, 'create_floating_ip')
+ self.mock_create_floating_ip = self._mock_create_floating_ip.start()
+ self._mock_get_neutron_client = mock.patch.object(
+ op_utils, 'get_neutron_client')
+ self.mock_get_neutron_client = self._mock_get_neutron_client.start()
+ self._mock_get_shade_client = mock.patch.object(
+ op_utils, 'get_shade_client')
+ self.mock_get_shade_client = self._mock_get_shade_client.start()
+ self._mock_log = mock.patch.object(create_floating_ip, 'LOG')
+ self.mock_log = self._mock_log.start()
+
+ self._fip_obj = create_floating_ip.CreateFloatingIp(mock.ANY, mock.ANY)
+ self._fip_obj.scenario_cfg = {'output': 'key1\nkey2'}
+
+ self.addCleanup(self._stop_mock)
-def main():
- unittest.main()
+ def _stop_mock(self):
+ self._mock_get_network_id.stop()
+ self._mock_create_floating_ip.stop()
+ self._mock_get_neutron_client.stop()
+ self._mock_get_shade_client.stop()
+ self._mock_log.stop()
+ def test_run(self):
+ self.mock_create_floating_ip.return_value = {'fip_id': 'value1',
+ 'fip_addr': 'value2'}
+ output = self._fip_obj.run(mock.ANY)
+ self.assertDictEqual({'key1': 'value1', 'key2': 'value2'}, output)
-if __name__ == '__main__':
- main()
+ def test_run_no_fip(self):
+ self.mock_create_floating_ip.return_value = None
+ output = self._fip_obj.run(mock.ANY)
+ self.assertIsNone(output)
+ self.mock_log.error.assert_called_once_with(
+ 'Creating floating ip failed!')
diff --git a/yardstick/benchmark/scenarios/lib/create_floating_ip.py b/yardstick/benchmark/scenarios/lib/create_floating_ip.py
index 328566d48..7108722af 100644
--- a/yardstick/benchmark/scenarios/lib/create_floating_ip.py
+++ b/yardstick/benchmark/scenarios/lib/create_floating_ip.py
@@ -7,15 +7,13 @@
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-from __future__ import print_function
-from __future__ import absolute_import
-
import logging
import os
from yardstick.benchmark.scenarios import base
import yardstick.common.openstack_utils as op_utils
+
LOG = logging.getLogger(__name__)
@@ -30,6 +28,7 @@ class CreateFloatingIp(base.Scenario):
self.ext_net_id = os.getenv("EXTERNAL_NETWORK", "external")
self.neutron_client = op_utils.get_neutron_client()
+ self.shade_client = op_utils.get_shade_client()
self.setup_done = False
def setup(self):
@@ -37,24 +36,21 @@ class CreateFloatingIp(base.Scenario):
self.setup_done = True
- def run(self, result):
+ def run(self, *args):
"""execute the test"""
if not self.setup_done:
self.setup()
- net_id = op_utils.get_network_id(self.neutron_client, self.ext_net_id)
+ net_id = op_utils.get_network_id(self.shade_client, self.ext_net_id)
floating_info = op_utils.create_floating_ip(self.neutron_client,
extnet_id=net_id)
- if floating_info:
- LOG.info("Creating floating ip successful!")
- else:
+
+ if not floating_info:
LOG.error("Creating floating ip failed!")
+ return
- try:
- keys = self.scenario_cfg.get('output', '').split()
- except KeyError:
- pass
- else:
- values = [floating_info["fip_id"], floating_info["fip_addr"]]
- return self._push_to_outputs(keys, values)
+ LOG.info("Creating floating ip successful!")
+ keys = self.scenario_cfg.get('output', '').split()
+ values = [floating_info["fip_id"], floating_info["fip_addr"]]
+ return self._push_to_outputs(keys, values)
diff --git a/yardstick/common/openstack_utils.py b/yardstick/common/openstack_utils.py
index d1223edd2..c5b17c270 100644
--- a/yardstick/common/openstack_utils.py
+++ b/yardstick/common/openstack_utils.py
@@ -7,8 +7,6 @@
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-from __future__ import absolute_import
-
import os
import time
import sys
@@ -16,11 +14,14 @@ import logging
from keystoneauth1 import loading
from keystoneauth1 import session
+import shade
+
from cinderclient import client as cinderclient
from novaclient import client as novaclient
from glanceclient import client as glanceclient
from neutronclient.neutron import client as neutronclient
+
log = logging.getLogger(__name__)
DEFAULT_HEAT_API_VERSION = '1'
@@ -170,27 +171,30 @@ def get_glance_client(): # pragma: no cover
return glanceclient.Client(get_glance_client_version(), session=sess)
+def get_shade_client():
+ return shade.openstack_cloud()
+
# *********************************************
# NOVA
# *********************************************
-def get_instances(nova_client): # pragma: no cover
+def get_instances(nova_client):
try:
return nova_client.servers.list(search_opts={'all_tenants': 1})
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.exception("Error [get_instances(nova_client)]")
def get_instance_status(nova_client, instance): # pragma: no cover
try:
return nova_client.servers.get(instance.id).status
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.exception("Error [get_instance_status(nova_client)]")
def get_instance_by_name(nova_client, instance_name): # pragma: no cover
try:
return nova_client.servers.find(name=instance_name)
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.exception("Error [get_instance_by_name(nova_client, '%s')]",
instance_name)
@@ -198,28 +202,28 @@ def get_instance_by_name(nova_client, instance_name): # pragma: no cover
def get_aggregates(nova_client): # pragma: no cover
try:
return nova_client.aggregates.list()
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.exception("Error [get_aggregates(nova_client)]")
def get_availability_zones(nova_client): # pragma: no cover
try:
return nova_client.availability_zones.list()
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.exception("Error [get_availability_zones(nova_client)]")
def get_availability_zone_names(nova_client): # pragma: no cover
try:
return [az.zoneName for az in get_availability_zones(nova_client)]
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.exception("Error [get_availability_zone_names(nova_client)]")
def create_aggregate(nova_client, aggregate_name, av_zone): # pragma: no cover
try:
nova_client.aggregates.create(aggregate_name, av_zone)
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.exception("Error [create_aggregate(nova_client, %s, %s)]",
aggregate_name, av_zone)
return False
@@ -231,7 +235,7 @@ def get_aggregate_id(nova_client, aggregate_name): # pragma: no cover
try:
aggregates = get_aggregates(nova_client)
_id = next((ag.id for ag in aggregates if ag.name == aggregate_name))
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.exception("Error [get_aggregate_id(nova_client, %s)]",
aggregate_name)
else:
@@ -243,7 +247,7 @@ def add_host_to_aggregate(nova_client, aggregate_name,
try:
aggregate_id = get_aggregate_id(nova_client, aggregate_name)
nova_client.aggregates.add_host(aggregate_id, compute_host)
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.exception("Error [add_host_to_aggregate(nova_client, %s, %s)]",
aggregate_name, compute_host)
return False
@@ -256,7 +260,7 @@ def create_aggregate_with_host(nova_client, aggregate_name, av_zone,
try:
create_aggregate(nova_client, aggregate_name, av_zone)
add_host_to_aggregate(nova_client, aggregate_name, compute_host)
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.exception("Error [create_aggregate_with_host("
"nova_client, %s, %s, %s)]",
aggregate_name, av_zone, compute_host)
@@ -265,19 +269,19 @@ def create_aggregate_with_host(nova_client, aggregate_name, av_zone,
return True
-def create_keypair(nova_client, name, key_path=None): # pragma: no cover
+def create_keypair(name, key_path=None): # pragma: no cover
try:
with open(key_path) as fpubkey:
keypair = get_nova_client().keypairs.create(name=name, public_key=fpubkey.read())
return keypair
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.exception("Error [create_keypair(nova_client)]")
def create_instance(json_body): # pragma: no cover
try:
return get_nova_client().servers.create(**json_body)
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.exception("Error create instance failed")
return None
@@ -288,7 +292,7 @@ def create_instance_and_wait_for_active(json_body): # pragma: no cover
nova_client = get_nova_client()
instance = create_instance(json_body)
count = VM_BOOT_TIMEOUT / SLEEP
- for n in range(count, -1, -1):
+ for _ in range(count, -1, -1):
status = get_instance_status(nova_client, instance)
if status.lower() == "active":
return instance
@@ -303,7 +307,7 @@ def create_instance_and_wait_for_active(json_body): # pragma: no cover
def attach_server_volume(server_id, volume_id, device=None): # pragma: no cover
try:
get_nova_client().volumes.create_server_volume(server_id, volume_id, device)
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.exception("Error [attach_server_volume(nova_client, '%s', '%s')]",
server_id, volume_id)
return False
@@ -314,7 +318,7 @@ def attach_server_volume(server_id, volume_id, device=None): # pragma: no cov
def delete_instance(nova_client, instance_id): # pragma: no cover
try:
nova_client.servers.force_delete(instance_id)
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.exception("Error [delete_instance(nova_client, '%s')]",
instance_id)
return False
@@ -327,7 +331,7 @@ def remove_host_from_aggregate(nova_client, aggregate_name,
try:
aggregate_id = get_aggregate_id(nova_client, aggregate_name)
nova_client.aggregates.remove_host(aggregate_id, compute_host)
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.exception("Error remove_host_from_aggregate(nova_client, %s, %s)",
aggregate_name, compute_host)
return False
@@ -348,7 +352,7 @@ def delete_aggregate(nova_client, aggregate_name): # pragma: no cover
try:
remove_hosts_from_aggregate(nova_client, aggregate_name)
nova_client.aggregates.delete(aggregate_name)
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.exception("Error [delete_aggregate(nova_client, %s)]",
aggregate_name)
return False
@@ -367,7 +371,7 @@ def get_server_by_name(name): # pragma: no cover
def create_flavor(name, ram, vcpus, disk, **kwargs): # pragma: no cover
try:
return get_nova_client().flavors.create(name, ram, vcpus, disk, **kwargs)
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.exception("Error [create_flavor(nova_client, %s, %s, %s, %s, %s)]",
name, ram, disk, vcpus, kwargs['is_public'])
return None
@@ -400,7 +404,7 @@ def get_flavor_by_name(name): # pragma: no cover
def check_status(status, name, iterations, interval): # pragma: no cover
- for i in range(iterations):
+ for _ in range(iterations):
try:
server = get_server_by_name(name)
except IndexError:
@@ -417,7 +421,7 @@ def check_status(status, name, iterations, interval): # pragma: no cover
def delete_flavor(flavor_id): # pragma: no cover
try:
get_nova_client().flavors.delete(flavor_id)
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.exception("Error [delete_flavor(nova_client, %s)]", flavor_id)
return False
else:
@@ -428,7 +432,7 @@ def delete_keypair(nova_client, key): # pragma: no cover
try:
nova_client.keypairs.delete(key=key)
return True
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.exception("Error [delete_keypair(nova_client)]")
return False
@@ -436,33 +440,28 @@ def delete_keypair(nova_client, key): # pragma: no cover
# *********************************************
# NEUTRON
# *********************************************
-def get_network_id(neutron_client, network_name): # pragma: no cover
- networks = neutron_client.list_networks()['networks']
- return next((n['id'] for n in networks if n['name'] == network_name), None)
-
-
-def get_port_id_by_ip(neutron_client, ip_address): # pragma: no cover
- ports = neutron_client.list_ports()['ports']
- return next((i['id'] for i in ports for j in i.get(
- 'fixed_ips') if j['ip_address'] == ip_address), None)
+def get_network_id(shade_client, network_name):
+ networks = shade_client.list_networks({'name': network_name})
+ if networks:
+ return networks[0]['id']
def create_neutron_net(neutron_client, json_body): # pragma: no cover
try:
network = neutron_client.create_network(body=json_body)
return network['network']['id']
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.error("Error [create_neutron_net(neutron_client)]")
raise Exception("operation error")
- return None
def delete_neutron_net(neutron_client, network_id): # pragma: no cover
try:
neutron_client.delete_network(network_id)
return True
- except Exception:
- log.error("Error [delete_neutron_net(neutron_client, '%s')]" % network_id)
+ except Exception: # pylint: disable=broad-except
+ log.error("Error [delete_neutron_net(neutron_client, '%s')]",
+ network_id)
return False
@@ -470,28 +469,27 @@ def create_neutron_subnet(neutron_client, json_body): # pragma: no cover
try:
subnet = neutron_client.create_subnet(body=json_body)
return subnet['subnets'][0]['id']
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.error("Error [create_neutron_subnet")
raise Exception("operation error")
- return None
def create_neutron_router(neutron_client, json_body): # pragma: no cover
try:
router = neutron_client.create_router(json_body)
return router['router']['id']
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.error("Error [create_neutron_router(neutron_client)]")
raise Exception("operation error")
- return None
def delete_neutron_router(neutron_client, router_id): # pragma: no cover
try:
neutron_client.delete_router(router=router_id)
return True
- except Exception:
- log.error("Error [delete_neutron_router(neutron_client, '%s')]" % router_id)
+ except Exception: # pylint: disable=broad-except
+ log.error("Error [delete_neutron_router(neutron_client, '%s')]",
+ router_id)
return False
@@ -499,8 +497,9 @@ def remove_gateway_router(neutron_client, router_id): # pragma: no cover
try:
neutron_client.remove_gateway_router(router_id)
return True
- except Exception:
- log.error("Error [remove_gateway_router(neutron_client, '%s')]" % router_id)
+ except Exception: # pylint: disable=broad-except
+ log.error("Error [remove_gateway_router(neutron_client, '%s')]",
+ router_id)
return False
@@ -511,9 +510,9 @@ def remove_interface_router(neutron_client, router_id, subnet_id,
neutron_client.remove_interface_router(router=router_id,
body=json_body)
return True
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.error("Error [remove_interface_router(neutron_client, '%s', "
- "'%s')]" % (router_id, subnet_id))
+ "'%s')]", router_id, subnet_id)
return False
@@ -523,7 +522,7 @@ def create_floating_ip(neutron_client, extnet_id): # pragma: no cover
ip_json = neutron_client.create_floatingip({'floatingip': props})
fip_addr = ip_json['floatingip']['floating_ip_address']
fip_id = ip_json['floatingip']['id']
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.error("Error [create_floating_ip(neutron_client)]")
return None
return {'fip_addr': fip_addr, 'fip_id': fip_id}
@@ -533,8 +532,9 @@ def delete_floating_ip(nova_client, floatingip_id): # pragma: no cover
try:
nova_client.floating_ips.delete(floatingip_id)
return True
- except Exception:
- log.error("Error [delete_floating_ip(nova_client, '%s')]" % floatingip_id)
+ except Exception: # pylint: disable=broad-except
+ log.error("Error [delete_floating_ip(nova_client, '%s')]",
+ floatingip_id)
return False
@@ -543,7 +543,7 @@ def get_security_groups(neutron_client): # pragma: no cover
security_groups = neutron_client.list_security_groups()[
'security_groups']
return security_groups
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.error("Error [get_security_groups(neutron_client)]")
return None
@@ -564,9 +564,9 @@ def create_security_group(neutron_client, sg_name, sg_description): # pragm
try:
secgroup = neutron_client.create_security_group(json_body)
return secgroup['security_group']
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.error("Error [create_security_group(neutron_client, '%s', "
- "'%s')]" % (sg_name, sg_description))
+ "'%s')]", sg_name, sg_description)
return None
@@ -597,16 +597,15 @@ def create_secgroup_rule(neutron_client, sg_id, direction, protocol,
else:
log.error("Bad security group format."
"One of the port range is not properly set:"
- "range min: {},"
- "range max: {}".format(port_range_min,
- port_range_max))
+ "range min: %s, range max: %s", port_range_min,
+ port_range_max)
return False
# Create security group using neutron client
try:
neutron_client.create_security_group_rule(json_body)
return True
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.exception("Impossible to create_security_group_rule,"
"security group rule probably already exists")
return False
@@ -616,9 +615,9 @@ def create_security_group_full(neutron_client,
sg_name, sg_description): # pragma: no cover
sg_id = get_security_group_id(neutron_client, sg_name)
if sg_id != '':
- log.info("Using existing security group '%s'..." % sg_name)
+ log.info("Using existing security group '%s'...", sg_name)
else:
- log.info("Creating security group '%s'..." % sg_name)
+ log.info("Creating security group '%s'...", sg_name)
SECGROUP = create_security_group(neutron_client,
sg_name,
sg_description)
@@ -628,18 +627,16 @@ def create_security_group_full(neutron_client,
sg_id = SECGROUP['id']
- log.debug("Security group '%s' with ID=%s created successfully."
- % (SECGROUP['name'], sg_id))
+ log.debug("Security group '%s' with ID=%s created successfully.",
+ SECGROUP['name'], sg_id)
- log.debug("Adding ICMP rules in security group '%s'..."
- % sg_name)
+ log.debug("Adding ICMP rules in security group '%s'...", sg_name)
if not create_secgroup_rule(neutron_client, sg_id,
'ingress', 'icmp'):
log.error("Failed to create the security group rule...")
return None
- log.debug("Adding SSH rules in security group '%s'..."
- % sg_name)
+ log.debug("Adding SSH rules in security group '%s'...", sg_name)
if not create_secgroup_rule(
neutron_client, sg_id, 'ingress', 'tcp', '22', '22'):
log.error("Failed to create the security group rule...")
@@ -664,12 +661,12 @@ def create_image(glance_client, image_name, file_path, disk_format,
container_format, min_disk, min_ram, protected, tag,
public, **kwargs): # pragma: no cover
if not os.path.isfile(file_path):
- log.error("Error: file %s does not exist." % file_path)
+ log.error("Error: file %s does not exist.", file_path)
return None
try:
image_id = get_image_id(glance_client, image_name)
if image_id is not None:
- log.info("Image %s already exists." % image_name)
+ log.info("Image %s already exists.", image_name)
else:
log.info("Creating image '%s' from '%s'...", image_name, file_path)
@@ -686,7 +683,7 @@ def create_image(glance_client, image_name, file_path, disk_format,
with open(file_path) as image_data:
glance_client.images.upload(image_id, image_data)
return image_id
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.error("Error [create_glance_image(glance_client, '%s', '%s', '%s')]",
image_name, file_path, public)
return None
@@ -696,7 +693,7 @@ def delete_image(glance_client, image_id): # pragma: no cover
try:
glance_client.images.delete(image_id)
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.exception("Error [delete_flavor(glance_client, %s)]", image_id)
return False
else:
@@ -722,7 +719,7 @@ def create_volume(cinder_client, volume_name, volume_size,
volume = cinder_client.volumes.create(name=volume_name,
size=volume_size)
return volume
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.exception("Error [create_volume(cinder_client, %s)]",
(volume_name, volume_size))
return None
@@ -733,7 +730,7 @@ def delete_volume(cinder_client, volume_id, forced=False): # pragma: no cov
if forced:
try:
cinder_client.volumes.detach(volume_id)
- except:
+ except Exception: # pylint: disable=broad-except
log.error(sys.exc_info()[0])
cinder_client.volumes.force_delete(volume_id)
else:
@@ -743,8 +740,8 @@ def delete_volume(cinder_client, volume_id, forced=False): # pragma: no cov
break
cinder_client.volumes.delete(volume_id)
return True
- except Exception:
- log.exception("Error [delete_volume(cinder_client, '%s')]" % volume_id)
+ except Exception: # pylint: disable=broad-except
+ log.exception("Error [delete_volume(cinder_client, '%s')]", volume_id)
return False
@@ -752,7 +749,7 @@ def detach_volume(server_id, volume_id): # pragma: no cover
try:
get_nova_client().volumes.delete_server_volume(server_id, volume_id)
return True
- except Exception:
+ except Exception: # pylint: disable=broad-except
log.exception("Error [detach_server_volume(nova_client, '%s', '%s')]",
server_id, volume_id)
return False
diff --git a/yardstick/tests/unit/common/test_openstack_utils.py b/yardstick/tests/unit/common/test_openstack_utils.py
index bf468489e..b685e63be 100644
--- a/yardstick/tests/unit/common/test_openstack_utils.py
+++ b/yardstick/tests/unit/common/test_openstack_utils.py
@@ -1,5 +1,3 @@
-#!/usr/bin/env python
-
##############################################################################
# Copyright (c) 2016 Huawei Technologies Co.,Ltd and others.
#
@@ -9,9 +7,7 @@
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-# Unittest for yardstick.common.openstack_utils
-
-from __future__ import absolute_import
+from oslo_utils import uuidutils
import unittest
import mock
@@ -38,9 +34,23 @@ class GetHeatApiVersionTestCase(unittest.TestCase):
self.assertEqual(api_version, expected_result)
-def main():
- unittest.main()
+class GetNetworkIdTestCase(unittest.TestCase):
+
+ def test_get_network_id(self):
+ _uuid = uuidutils.generate_uuid()
+ mock_shade_client = mock.Mock()
+ mock_shade_client.list_networks = mock.Mock()
+ mock_shade_client.list_networks.return_value = [{'id': _uuid}]
+
+ output = openstack_utils.get_network_id(mock_shade_client,
+ 'network_name')
+ self.assertEqual(_uuid, output)
+ def test_get_network_id_no_network(self):
+ mock_shade_client = mock.Mock()
+ mock_shade_client.list_networks = mock.Mock()
+ mock_shade_client.list_networks.return_value = None
-if __name__ == '__main__':
- main()
+ output = openstack_utils.get_network_id(mock_shade_client,
+ 'network_name')
+ self.assertEqual(None, output)