import os
from yardstick.benchmark.scenarios import base
-import yardstick.common.openstack_utils as op_utils
+from yardstick.common import openstack_utils
+from yardstick.common import exceptions
LOG = logging.getLogger(__name__)
self.scenario_cfg = scenario_cfg
self.context_cfg = context_cfg
self.ext_net_id = os.getenv("EXTERNAL_NETWORK", "external")
+ self.options = self.scenario_cfg["options"]
+
+ self.network_name_or_id = self.options.get("network_name_or_id", self.ext_net_id)
+ self.server = self.options.get("server")
+ self.fixed_address = self.options.get("fixed_address")
+ self.nat_destination = self.options.get("nat_destination")
+ self.port = self.options.get("port")
+ self.wait = self.options.get("wait", False)
+ self.timeout = self.options.get("timeout", 60)
+
+ self.shade_client = openstack_utils.get_shade_client()
- self.neutron_client = op_utils.get_neutron_client()
- self.shade_client = op_utils.get_shade_client()
self.setup_done = False
def setup(self):
self.setup_done = True
- def run(self, *args):
+ def run(self, result):
"""execute the test"""
if not self.setup_done:
self.setup()
- 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)
+ floating_info = openstack_utils.create_floating_ip(
+ self.shade_client, network_name_or_id=self.network_name_or_id,
+ server=self.server, fixed_address=self.fixed_address,
+ nat_destination=self.nat_destination, port=self.port,
+ wait=self.wait, timeout=self.timeout)
if not floating_info:
+ result.update({"floating_ip_create": 0})
LOG.error("Creating floating ip failed!")
- return
+ raise exceptions.ScenarioCreateFloatingIPError
+ result.update({"floating_ip_create": 1})
LOG.info("Creating floating ip successful!")
- keys = self.scenario_cfg.get('output', '').split()
+ keys = self.scenario_cfg.get("output", '').split()
values = [floating_info["fip_id"], floating_info["fip_addr"]]
return self._push_to_outputs(keys, values)
class ScenarioRemoveRouterIntError(YardstickException):
message = 'Remove Neutron Router Interface Scenario failed'
+
+
+class ScenarioCreateFloatingIPError(YardstickException):
+ message = 'Create Neutron Floating IP Scenario failed'
# *********************************************
# NEUTRON
# *********************************************
-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(shade_client, network_name, shared=False,
admin_state_up=True, external=False, provider=None,
project_id=None):
return False
-def create_floating_ip(neutron_client, extnet_id): # pragma: no cover
- props = {'floating_network_id': extnet_id}
+def create_floating_ip(shade_client, network_name_or_id=None, server=None,
+ fixed_address=None, nat_destination=None,
+ port=None, wait=False, timeout=60):
+ """Allocate a new floating IP from a network or a pool.
+
+ :param network_name_or_id: Name or ID of the network
+ that the floating IP should come from.
+ :param server: Server dict for the server to create
+ the IP for and to which it should be attached.
+ :param fixed_address: Fixed IP to attach the floating ip to.
+ :param nat_destination: Name or ID of the network
+ that the fixed IP to attach the floating
+ IP to should be on.
+ :param port: The port ID that the floating IP should be
+ attached to. Specifying a port conflicts with specifying a
+ server,fixed_address or nat_destination.
+ :param wait: Whether to wait for the IP to be active.Only applies
+ if a server is provided.
+ :param timeout: How long to wait for the IP to be active.Only
+ applies if a server is provided.
+
+ :returns:Floating IP id and address
+ """
try:
- ip_json = neutron_client.create_floatingip({'floatingip': props})
- fip_addr = ip_json['floatingip']['floating_ip_address']
- fip_id = ip_json['floatingip']['id']
- 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}
+ fip = shade_client.create_floating_ip(
+ network=network_name_or_id, server=server,
+ fixed_address=fixed_address, nat_destination=nat_destination,
+ port=port, wait=wait, timeout=timeout)
+ return {'fip_addr': fip['floating_ip_address'], 'fip_id': fip['id']}
+ except exc.OpenStackCloudException as o_exc:
+ log.error("Error [create_floating_ip(shade_client)]. "
+ "Exception message: %s", o_exc.orig_message)
def delete_floating_ip(nova_client, floatingip_id): # pragma: no cover
import mock
from yardstick.benchmark.scenarios.lib import create_floating_ip
-import yardstick.common.openstack_utils as op_utils
+from yardstick.common import openstack_utils
+from yardstick.common import exceptions
class CreateFloatingIpTestCase(unittest.TestCase):
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')
+ openstack_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')
+ openstack_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.args = {'options': {'network_name_or_id': 'yardstick_net'}}
+ self.result = {}
- self._fip_obj = create_floating_ip.CreateFloatingIp(mock.ANY, mock.ANY)
- self._fip_obj.scenario_cfg = {'output': 'key1\nkey2'}
+ self.fip_obj = create_floating_ip.CreateFloatingIp(self.args, mock.ANY)
+ self.fip_obj.scenario_cfg = {'output': 'key1\nkey2'}
self.addCleanup(self._stop_mock)
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)
+ output = self.fip_obj.run(self.result)
+ self.assertEqual({'floating_ip_create': 1}, self.result)
+ self.assertEqual({'key1': 'value1', 'key2': 'value2'}, output)
+ self.mock_log.info.asset_called_once_with(
+ 'Creating floating ip successful!')
def test_run_no_fip(self):
self.mock_create_floating_ip.return_value = None
- output = self._fip_obj.run(mock.ANY)
- self.assertIsNone(output)
+ with self.assertRaises(exceptions.ScenarioCreateFloatingIPError):
+ self.fip_obj.run(self.result)
+ self.assertEqual({'floating_ip_create': 0}, self.result)
self.mock_log.error.assert_called_once_with(
'Creating floating ip failed!')
self.assertEqual(api_version, expected_result)
-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
-
- output = openstack_utils.get_network_id(mock_shade_client,
- 'network_name')
- self.assertIsNone(output)
-
-
class DeleteNeutronNetTestCase(unittest.TestCase):
def setUp(self):
self.mock_shade_client, self.router)
mock_logger.error.assert_called_once()
self.assertFalse(output)
+
+
+class CreateFloatingIpTestCase(unittest.TestCase):
+
+ def setUp(self):
+ self.mock_shade_client = mock.Mock()
+ self.network_name_or_id = 'name'
+ self.mock_shade_client.create_floating_ip = mock.Mock()
+
+ def test_create_floating_ip(self):
+ self.mock_shade_client.create_floating_ip.return_value = \
+ {'floating_ip_address': 'value1', 'id': 'value2'}
+ output = openstack_utils.create_floating_ip(self.mock_shade_client,
+ self.network_name_or_id)
+ self.assertEqual({'fip_addr': 'value1', 'fip_id': 'value2'}, output)
+
+ @mock.patch.object(openstack_utils, 'log')
+ def test_create_floating_ip_exception(self, mock_logger):
+ self.mock_shade_client.create_floating_ip.side_effect = (
+ exc.OpenStackCloudException('error message'))
+ output = openstack_utils.create_floating_ip(
+ self.mock_shade_client, self.network_name_or_id)
+ mock_logger.error.assert_called_once()
+ self.assertIsNone(output)
Code Review / yardstick.git / commitdiff