Function create_security_group_full now uses shade client.
Remove openstack util's private 'get_security_groups',
'get_security_group_id' & 'create_security_group' functions as
'create_security_group_full' now uses 'get_security_group' &
'create_security_group' function defined in shade client library.
JIRA: YARDSTICK-890
Change-Id: Ibd5b9ca9b7ce99b3e40c5dcc545f2af672ce2203
Signed-off-by: Shobhi Jain <shobhi.jain@intel.com>
# 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__)
def __init__(self, scenario_cfg, context_cfg):
self.scenario_cfg = scenario_cfg
self.context_cfg = context_cfg
- self.options = self.scenario_cfg['options']
+ self.options = self.scenario_cfg["options"]
- self.sg_name = self.options.get("sg_name", "yardstick_sec_group")
- self.description = self.options.get("description", None)
- self.neutron_client = op_utils.get_neutron_client()
+ self.sg_name = self.options["sg_name"]
+ self.description = self.options.get("description", "")
+ self.project_id = self.options.get("project_id")
+ self.shade_client = openstack_utils.get_shade_client()
self.setup_done = False
if not self.setup_done:
self.setup()
- sg_id = op_utils.create_security_group_full(self.neutron_client,
- sg_name=self.sg_name,
- sg_description=self.description)
-
- if sg_id:
- result.update({"sg_create": 1})
- LOG.info("Create security group successful!")
- else:
+ sg_id = openstack_utils.create_security_group_full(
+ self.shade_client, self.sg_name, sg_description=self.description,
+ project_id=self.project_id)
+ if not sg_id:
result.update({"sg_create": 0})
LOG.error("Create security group failed!")
+ raise exceptions.ScenarioCreateSecurityGroupError
- try:
- keys = self.scenario_cfg.get('output', '').split()
- except KeyError:
- pass
- else:
- values = [sg_id]
- return self._push_to_outputs(keys, values)
+ result.update({"sg_create": 1})
+ LOG.info("Create security group successful!")
+ keys = self.scenario_cfg.get("output", '').split()
+ values = [sg_id]
+ return self._push_to_outputs(keys, values)
class ScenarioDeleteFloatingIPError(YardstickException):
message = 'Delete Neutron Floating IP Scenario failed'
+
+
+class ScenarioCreateSecurityGroupError(YardstickException):
+ message = 'Create Neutron Security Group Scenario failed'
return False
-def get_security_groups(neutron_client): # pragma: no cover
- try:
- security_groups = neutron_client.list_security_groups()[
- 'security_groups']
- return security_groups
- except Exception: # pylint: disable=broad-except
- log.error("Error [get_security_groups(neutron_client)]")
- return None
-
-
-def get_security_group_id(neutron_client, sg_name): # pragma: no cover
- security_groups = get_security_groups(neutron_client)
- id = ''
- for sg in security_groups:
- if sg['name'] == sg_name:
- id = sg['id']
- break
- return id
-
-
-def create_security_group(neutron_client, sg_name,
- sg_description): # pragma: no cover
- json_body = {'security_group': {'name': sg_name,
- 'description': sg_description}}
- try:
- secgroup = neutron_client.create_security_group(json_body)
- return secgroup['security_group']
- except Exception: # pylint: disable=broad-except
- log.error("Error [create_security_group(neutron_client, '%s', "
- "'%s')]", sg_name, sg_description)
- return None
-
-
def create_security_group_rule(shade_client, secgroup_name_or_id,
port_range_min=None, port_range_max=None,
protocol=None, remote_ip_prefix=None,
return False
-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)
- else:
- log.info("Creating security group '%s'...", sg_name)
- SECGROUP = create_security_group(neutron_client,
- sg_name,
- sg_description)
- if not SECGROUP:
- log.error("Failed to create the security group...")
- return None
-
- sg_id = SECGROUP['id']
-
- log.debug("Security group '%s' with ID=%s created successfully.",
- SECGROUP['name'], sg_id)
+def create_security_group_full(shade_client, sg_name,
+ sg_description, project_id=None):
+ security_group = shade_client.get_security_group(sg_name)
- log.debug("Adding ICMP rules in security group '%s'...", sg_name)
- if not create_security_group_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)
- if not create_security_group_rule(
- neutron_client, sg_id, 'ingress', 'tcp', '22', '22'):
- log.error("Failed to create the security group rule...")
- return None
-
- if not create_security_group_rule(
- neutron_client, sg_id, 'egress', 'tcp', '22', '22'):
- log.error("Failed to create the security group rule...")
- return None
- return sg_id
+ if security_group:
+ log.info("Using existing security group '%s'...", sg_name)
+ return security_group['id']
+
+ log.info("Creating security group '%s'...", sg_name)
+ try:
+ security_group = shade_client.create_security_group(
+ sg_name, sg_description, project_id=project_id)
+ except (exc.OpenStackCloudException,
+ exc.OpenStackCloudUnavailableFeature) as op_exc:
+ log.error("Error [create_security_group(shade_client, %s, %s)]. "
+ "Exception message: %s", sg_name, sg_description,
+ op_exc.orig_message)
+ return
+
+ log.debug("Security group '%s' with ID=%s created successfully.",
+ security_group['name'], security_group['id'])
+
+ log.debug("Adding ICMP rules in security group '%s'...", sg_name)
+ if not create_security_group_rule(shade_client, security_group['id'],
+ direction='ingress', protocol='icmp'):
+ log.error("Failed to create the security group rule...")
+ shade_client.delete_security_group(sg_name)
+ return
+
+ log.debug("Adding SSH rules in security group '%s'...", sg_name)
+ if not create_security_group_rule(shade_client, security_group['id'],
+ direction='ingress', protocol='tcp',
+ port_range_min='22',
+ port_range_max='22'):
+ log.error("Failed to create the security group rule...")
+ shade_client.delete_security_group(sg_name)
+ return
+
+ if not create_security_group_rule(shade_client, security_group['id'],
+ direction='egress', protocol='tcp',
+ port_range_min='22',
+ port_range_max='22'):
+ log.error("Failed to create the security group rule...")
+ shade_client.delete_security_group(sg_name)
+ return
+ return security_group['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_sec_group import CreateSecgroup
-
-
-class CreateSecGroupTestCase(unittest.TestCase):
-
- @mock.patch('yardstick.common.openstack_utils.get_neutron_client')
- @mock.patch('yardstick.common.openstack_utils.create_security_group_full')
- def test_create_sec_group(self, mock_get_neutron_client, mock_create_security_group_full):
- options = {
- 'openstack_paras': {
- 'sg_name': 'yardstick_sec_group',
- 'description': 'security group for yardstick manual VM'
- }
- }
- args = {"options": options}
- obj = CreateSecgroup(args, {})
- obj.run({})
- mock_get_neutron_client.assert_called_once()
- mock_create_security_group_full.assert_called_once()
+from yardstick.common import openstack_utils
+from yardstick.common import exceptions
+from yardstick.benchmark.scenarios.lib import create_sec_group
+
+
+class CreateSecurityGroupTestCase(unittest.TestCase):
+
+ def setUp(self):
+
+ self._mock_create_security_group_full = mock.patch.object(
+ openstack_utils, 'create_security_group_full')
+ self.mock_create_security_group_full = (
+ self._mock_create_security_group_full.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_sec_group, 'LOG')
+ self.mock_log = self._mock_log.start()
+ self.args = {'options': {'sg_name': 'yardstick_sg'}}
+ self.result = {}
+
+ self.csecgp_obj = create_sec_group.CreateSecgroup(self.args, mock.ANY)
+ self.addCleanup(self._stop_mock)
+
+ def _stop_mock(self):
+ self._mock_create_security_group_full.stop()
+ self._mock_get_shade_client.stop()
+ self._mock_log.stop()
+
+ def test_run(self):
+ _uuid = uuidutils.generate_uuid()
+ self.csecgp_obj.scenario_cfg = {'output': 'id'}
+ self.mock_create_security_group_full.return_value = _uuid
+ output = self.csecgp_obj.run(self.result)
+ self.assertEqual({'sg_create': 1}, self.result)
+ self.assertEqual({'id': _uuid}, output)
+ self.mock_log.info.asset_called_once_with(
+ 'Create security group successful!')
+
+ def test_run_fail(self):
+ self.mock_create_security_group_full.return_value = None
+ with self.assertRaises(exceptions.ScenarioCreateSecurityGroupError):
+ self.csecgp_obj.run(self.result)
+ self.assertEqual({'sg_create': 0}, self.result)
+ self.mock_log.error.assert_called_once_with(
+ 'Create security group failed!')
images = openstack_utils.list_images(mock_shade_client)
mock_logger.error.assert_called_once()
self.assertFalse(images)
+
+
+class SecurityGroupTestCase(unittest.TestCase):
+
+ def setUp(self):
+ self.mock_shade_client = mock.Mock()
+ self.sg_name = 'sg_name'
+ self.sg_description = 'sg_description'
+ self._uuid = uuidutils.generate_uuid()
+
+ def test_create_security_group_full_existing_security_group(self):
+ self.mock_shade_client.get_security_group.return_value = (
+ {'name': 'name', 'id': self._uuid})
+ output = openstack_utils.create_security_group_full(
+ self.mock_shade_client, self.sg_name, self.sg_description)
+ self.mock_shade_client.get_security_group.assert_called_once()
+ self.assertEqual(self._uuid, output)
+
+ @mock.patch.object(openstack_utils, 'log')
+ def test_create_security_group_full_non_existing_security_group(
+ self, mock_logger):
+ self.mock_shade_client.get_security_group.return_value = None
+ self.mock_shade_client.create_security_group.side_effect = (
+ exc.OpenStackCloudException('error message'))
+ output = openstack_utils.create_security_group_full(
+ self.mock_shade_client, self.sg_name, self.sg_description)
+ mock_logger.error.assert_called_once()
+ self.assertIsNone(output)
+
+ @mock.patch.object(openstack_utils, 'create_security_group_rule')
+ @mock.patch.object(openstack_utils, 'log')
+ def test_create_security_group_full_create_rule_fail(
+ self, mock_logger, mock_create_security_group_rule):
+ self.mock_shade_client.get_security_group.return_value = None
+ self.mock_shade_client.create_security_group.return_value = (
+ {'name': 'name', 'id': self._uuid})
+ mock_create_security_group_rule.return_value = False
+ output = openstack_utils.create_security_group_full(
+ self.mock_shade_client, self.sg_name, self.sg_description)
+ mock_create_security_group_rule.assert_called()
+ self.mock_shade_client.delete_security_group(self.sg_name)
+ mock_logger.error.assert_called_once()
+ self.assertIsNone(output)
+
+ @mock.patch.object(openstack_utils, 'create_security_group_rule')
+ def test_create_security_group_full(
+ self, mock_create_security_group_rule):
+ self.mock_shade_client.get_security_group.return_value = None
+ self.mock_shade_client.create_security_group.return_value = (
+ {'name': 'name', 'id': self._uuid})
+ mock_create_security_group_rule.return_value = True
+ output = openstack_utils.create_security_group_full(
+ self.mock_shade_client, self.sg_name, self.sg_description)
+ mock_create_security_group_rule.assert_called()
+ self.mock_shade_client.delete_security_group(self.sg_name)
+ self.assertEqual(self._uuid, output)
Code Review / yardstick.git / commitdiff