# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-from __future__ import print_function
-from __future__ import absolute_import
-
import logging
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.context_cfg = context_cfg
self.options = self.scenario_cfg['options']
- self.openstack = self.options.get("openstack_paras", None)
+ self.network_name = self.options["network_name"]
+ self.shared = self.options.get("shared", False)
+ self.admin_state_up = self.options.get("admin_state_up", True)
+ self.external = self.options.get("external", False)
+ self.provider = self.options.get("provider")
+ self.project_id = self.options.get("project_id")
- self.neutron_client = op_utils.get_neutron_client()
+ self.shade_client = openstack_utils.get_shade_client()
self.setup_done = False
if not self.setup_done:
self.setup()
- openstack_paras = {'network': self.openstack}
- network_id = op_utils.create_neutron_net(self.neutron_client,
- openstack_paras)
- if network_id:
- result.update({"network_create": 1})
- LOG.info("Create network successful!")
- else:
+ network_id = openstack_utils.create_neutron_net(
+ self.shade_client, self.network_name, shared=self.shared,
+ admin_state_up=self.admin_state_up, external=self.external,
+ provider=self.provider, project_id=self.project_id)
+ if not network_id:
result.update({"network_create": 0})
LOG.error("Create network failed!")
+ raise exceptions.ScenarioCreateNetworkError
- try:
- keys = self.scenario_cfg.get('output', '').split()
- except KeyError:
- pass
- else:
- values = [network_id]
- return self._push_to_outputs(keys, values)
+ result.update({"network_create": 1})
+ LOG.info("Create network successful!")
+ keys = self.scenario_cfg.get('output', '').split()
+ values = [network_id]
+ return self._push_to_outputs(keys, values)
class TaskRenderError(YardstickException):
message = 'Failed to render template:\n%(input_task)s'
+
+
+class ScenarioCreateNetworkError(YardstickException):
+ message = 'Create Neutron Network Scenario failed'
return networks[0]['id']
-def create_neutron_net(neutron_client, json_body): # pragma: no cover
+def create_neutron_net(shade_client, network_name, shared=False,
+ admin_state_up=True, external=False, provider=None,
+ project_id=None):
+ """Create a neutron network.
+
+ :param network_name:(string) name of the network being created.
+ :param shared:(bool) whether the network is shared.
+ :param admin_state_up:(bool) set the network administrative state.
+ :param external:(bool) whether this network is externally accessible.
+ :param provider:(dict) a dict of network provider options.
+ :param project_id:(string) specify the project ID this network
+ will be created on (admin-only).
+ :returns:(string) the network id.
+ """
try:
- network = neutron_client.create_network(body=json_body)
- return network['network']['id']
- except Exception: # pylint: disable=broad-except
- log.error("Error [create_neutron_net(neutron_client)]")
- raise Exception("operation error")
+ networks = shade_client.create_network(
+ name=network_name, shared=shared, admin_state_up=admin_state_up,
+ external=external, provider=provider, project_id=project_id)
+ return networks['id']
+ except exc.OpenStackCloudException as o_exc:
+ log.error("Error [create_neutron_net(shade_client)]."
+ "Exception message, '%s'", o_exc.orig_message)
+ return None
def delete_neutron_net(shade_client, network_id):
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
+
+from oslo_utils import uuidutils
import unittest
import mock
-from yardstick.benchmark.scenarios.lib.create_network import CreateNetwork
+from yardstick.common import openstack_utils
+from yardstick.common import exceptions
+from yardstick.benchmark.scenarios.lib import create_network
class CreateNetworkTestCase(unittest.TestCase):
- @mock.patch('yardstick.common.openstack_utils.get_neutron_client')
- @mock.patch('yardstick.common.openstack_utils.create_neutron_net')
- def test_create_network(self, mock_get_neutron_client, mock_create_neutron_net):
- options = {
- 'openstack_paras': {
- 'name': 'yardstick_net',
- 'admin_state_up': 'True'
- }
- }
- args = {"options": options}
- obj = CreateNetwork(args, {})
- obj.run({})
- mock_get_neutron_client.assert_called_once()
- mock_create_neutron_net.assert_called_once()
+ def setUp(self):
+
+ self._mock_create_neutron_net = mock.patch.object(
+ openstack_utils, 'create_neutron_net')
+ self.mock_create_neutron_net = self._mock_create_neutron_net.start()
+ self._mock_get_shade_client = mock.patch.object(
+ openstack_utils, 'get_shade_client')
+ self.mock_get_shade_client = self._mock_get_shade_client.start()
+ self._mock_log = mock.patch.object(create_network, 'LOG')
+ self.mock_log = self._mock_log.start()
+ self.args = {'options': {'network_name': 'yardstick_net'}}
+ self.result = {}
+
+ self._cnet_obj = create_network.CreateNetwork(self.args, mock.ANY)
+ self.addCleanup(self._stop_mock)
+
+ def _stop_mock(self):
+ self._mock_create_neutron_net.stop()
+ self._mock_get_shade_client.stop()
+ self._mock_log.stop()
+
+ def test_run(self):
+ _uuid = uuidutils.generate_uuid()
+ self._cnet_obj.scenario_cfg = {'output': 'id'}
+ self.mock_create_neutron_net.return_value = _uuid
+ output = self._cnet_obj.run(self.result)
+ self.assertEqual({"network_create": 1}, self.result)
+ self.assertEqual({'id': _uuid}, output)
+ self.mock_log.info.asset_called_once_with('Create network successful!')
+
+ def test_run_fail_exception(self):
+ self.mock_create_neutron_net.return_value = None
+ with self.assertRaises(exceptions.ScenarioCreateNetworkError):
+ self._cnet_obj.run(self.result)
+ self.assertEqual({"network_create": 0}, self.result)
+ self.mock_log.error.assert_called_once_with(
+ 'Create network failed!')
'network_id')
self.assertFalse(output)
mock_logger.error.assert_called_once()
+
+
+class CreateNeutronNetTestCase(unittest.TestCase):
+
+ def setUp(self):
+ self.mock_shade_client = mock.Mock()
+ self.network_name = 'name'
+ self.mock_shade_client.create_network = mock.Mock()
+
+ def test_create_neutron_net(self):
+ _uuid = uuidutils.generate_uuid()
+ self.mock_shade_client.create_network.return_value = {'id': _uuid}
+ output = openstack_utils.create_neutron_net(self.mock_shade_client,
+ self.network_name)
+ self.assertEqual(_uuid, output)
+
+ @mock.patch.object(openstack_utils, 'log')
+ def test_create_neutron_net_exception(self, mock_logger):
+ self.mock_shade_client.create_network.side_effect = (
+ exc.OpenStackCloudException('error message'))
+
+ output = openstack_utils.create_neutron_net(self.mock_shade_client,
+ self.network_name)
+ mock_logger.error.assert_called_once()
+ self.assertIsNone(output)
Code Review / yardstick.git / commitdiff