openstack/shade
Code Issues Proposed changes

Add delete method for security groups

Browse Source 
Adds the API method to delete a security group, using nova or neutron,
and consolidates existing and new security group tests into a new file.

Change-Id: Ib85b3ec0de8b2b42f691c3e2127adb16e8b14c99
This commit is contained in:
David Shrewsbury
2015-06-04 15:11:20 -04:00
parent 18cf7c2ccb
commit 50975f52ed
5 changed files with 193 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
shade/__init__.py
View File

@@ -2327,6 +2327,54 @@ class OpenStackCloud(object):
"failed to delete port '{port}': {msg}".format(
port=name_or_id, msg=str(e)))
def delete_security_group(self, name_or_id):
"""Delete a security group
:param string name_or_id: The name or unique ID of the security group.
:raises: OpenStackCloudException on operation error.
:raises: OpenStackCloudUnavailableFeature if security groups are
not supported on this cloud.
"""
secgroup = self.get_security_group(name_or_id)
if secgroup is None:
self.log.debug('security group %s was not found' % name_or_id)
return
if self.secgroup_source == 'neutron':
try:
self.manager.submitTask(
_tasks.NeutronSecurityGroupDelete(
security_group=secgroup['id']
)
)
except Exception as e:
self.log.debug(
"neutron failed to delete security group '{group}'".format(
group=name_or_id), exc_info=True)
raise OpenStackCloudException(
"failed to delete security group '{group}': {msg}".format(
group=name_or_id, msg=str(e)))
elif self.secgroup_source == 'nova':
try:
self.manager.submitTask(
_tasks.NovaSecurityGroupDelete(group=secgroup['id'])
)
except Exception as e:
self.log.debug(
"nova failed to delete security group '{group}'".format(
group=name_or_id), exc_info=True)
raise OpenStackCloudException(
"failed to delete security group '{group}': {msg}".format(
group=name_or_id, msg=str(e)))
# Security groups not supported
else:
raise OpenStackCloudUnavailableFeature(
"Unavailable feature: security groups"
)
class OperatorCloud(OpenStackCloud):
"""Represent a privileged/operator connection to an OpenStack Cloud.

10
shade/_tasks.py
View File

@@ -197,11 +197,21 @@ class NeutronSecurityGroupList(task_manager.Task):
return client.neutron_client.list_security_groups()
class NeutronSecurityGroupDelete(task_manager.Task):
def main(self, client):
return client.neutron_client.delete_security_group(**self.args)
class NovaSecurityGroupList(task_manager.Task):
def main(self, client):
return client.nova_client.security_groups.list()
class NovaSecurityGroupDelete(task_manager.Task):
def main(self, client):
return client.nova_client.security_groups.delete(**self.args)
# TODO: Do this with neutron instead of nova if possible
class FloatingIPList(task_manager.Task):
def main(self, client):

8
shade/tests/fakes.py
View File

@@ -89,3 +89,11 @@ class FakeMachinePort(object):
self.id = id
self.address = address
self.node_id = node_id
class FakeSecgroup(object):
def __init__(self, id, name, description='', rules=None):
self.id = id
self.name = name
self.description = description
self.rules = rules

127
shade/tests/unit/test_security_groups.py Normal file
View File

@@ -0,0 +1,127 @@
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import mock
import shade
from shade import meta
from shade.tests.unit import base
from shade.tests import fakes
neutron_grp_obj = fakes.FakeSecgroup(
id='1',
name='neutron-sec-group',
description='Test Neutron security group',
rules=[
dict(id='1', port_range_min=80, port_range_max=81,
protocol='tcp', remote_ip_prefix='0.0.0.0/0')
]
)
nova_grp_obj = fakes.FakeSecgroup(
id='2',
name='nova-sec-group',
description='Test Nova security group #1',
rules=[
dict(id='2', from_port=8000, to_port=8001, ip_protocol='tcp',
ip_range=dict(cidr='0.0.0.0/0'), parent_group_id=None)
]
)
# Neutron returns dicts instead of objects, so the dict versions should
# be used as expected return values from neutron API methods.
neutron_grp_dict = meta.obj_to_dict(neutron_grp_obj)
nova_grp_dict = meta.obj_to_dict(nova_grp_obj)
class TestSecurityGroups(base.TestCase):
def setUp(self):
super(TestSecurityGroups, self).setUp()
self.cloud = shade.openstack_cloud()
@mock.patch.object(shade.OpenStackCloud, 'neutron_client')
@mock.patch.object(shade.OpenStackCloud, 'nova_client')
def test_list_security_groups_neutron(self, mock_nova, mock_neutron):
self.cloud.secgroup_source = 'neutron'
self.cloud.list_security_groups()
self.assertTrue(mock_neutron.list_security_groups.called)
self.assertFalse(mock_nova.security_groups.list.called)
@mock.patch.object(shade.OpenStackCloud, 'neutron_client')
@mock.patch.object(shade.OpenStackCloud, 'nova_client')
def test_list_security_groups_nova(self, mock_nova, mock_neutron):
self.cloud.secgroup_source = 'nova'
self.cloud.list_security_groups()
self.assertFalse(mock_neutron.list_security_groups.called)
self.assertTrue(mock_nova.security_groups.list.called)
@mock.patch.object(shade.OpenStackCloud, 'neutron_client')
@mock.patch.object(shade.OpenStackCloud, 'nova_client')
def test_list_security_groups_none(self, mock_nova, mock_neutron):
self.cloud.secgroup_source = None
self.assertRaises(shade.OpenStackCloudUnavailableFeature,
self.cloud.list_security_groups)
self.assertFalse(mock_neutron.list_security_groups.called)
self.assertFalse(mock_nova.security_groups.list.called)
@mock.patch.object(shade.OpenStackCloud, 'neutron_client')
def test_delete_security_group_neutron(self, mock_neutron):
self.cloud.secgroup_source = 'neutron'
neutron_return = dict(security_groups=[neutron_grp_dict])
mock_neutron.list_security_groups.return_value = neutron_return
self.cloud.delete_security_group('1')
mock_neutron.delete_security_group.assert_called_once_with(
security_group='1'
)
@mock.patch.object(shade.OpenStackCloud, 'nova_client')
def test_delete_security_group_nova(self, mock_nova):
self.cloud.secgroup_source = 'nova'
nova_return = [nova_grp_obj]
mock_nova.security_groups.list.return_value = nova_return
self.cloud.delete_security_group('2')
mock_nova.security_groups.delete.assert_called_once_with(
group='2'
)
@mock.patch.object(shade.OpenStackCloud, 'neutron_client')
def test_delete_security_group_neutron_not_found(self, mock_neutron):
self.cloud.secgroup_source = 'neutron'
neutron_return = dict(security_groups=[neutron_grp_dict])
mock_neutron.list_security_groups.return_value = neutron_return
self.cloud.delete_security_group('doesNotExist')
self.assertFalse(mock_neutron.delete_security_group.called)
@mock.patch.object(shade.OpenStackCloud, 'nova_client')
def test_delete_security_group_nova_not_found(self, mock_nova):
self.cloud.secgroup_source = 'nova'
nova_return = [nova_grp_obj]
mock_nova.security_groups.list.return_value = nova_return
self.cloud.delete_security_group('doesNotExist')
self.assertFalse(mock_nova.security_groups.delete.called)
@mock.patch.object(shade.OpenStackCloud, 'neutron_client')
@mock.patch.object(shade.OpenStackCloud, 'nova_client')
def test_delete_security_group_none(self, mock_nova, mock_neutron):
self.cloud.secgroup_source = None
self.assertRaises(shade.OpenStackCloudUnavailableFeature,
self.cloud.delete_security_group,
'doesNotExist')
self.assertFalse(mock_neutron.delete_security_group.called)
self.assertFalse(mock_nova.security_groups.delete.called)

25
shade/tests/unit/test_shade.py
View File

@@ -325,31 +325,6 @@ class TestShade(base.TestCase):
flavor2 = self.cloud.get_flavor(1)
self.assertEquals(vanilla, flavor2)
@mock.patch.object(shade.OpenStackCloud, 'neutron_client')
@mock.patch.object(shade.OpenStackCloud, 'nova_client')
def test_list_security_groups_neutron(self, mock_nova, mock_neutron):
self.cloud.secgroup_source = 'neutron'
self.cloud.list_security_groups()
self.assertTrue(mock_neutron.list_security_groups.called)
self.assertFalse(mock_nova.security_groups.list.called)
@mock.patch.object(shade.OpenStackCloud, 'neutron_client')
@mock.patch.object(shade.OpenStackCloud, 'nova_client')
def test_list_security_groups_nova(self, mock_nova, mock_neutron):
self.cloud.secgroup_source = 'nova'
self.cloud.list_security_groups()
self.assertFalse(mock_neutron.list_security_groups.called)
self.assertTrue(mock_nova.security_groups.list.called)
@mock.patch.object(shade.OpenStackCloud, 'neutron_client')
@mock.patch.object(shade.OpenStackCloud, 'nova_client')
def test_list_security_groups_none(self, mock_nova, mock_neutron):
self.cloud.secgroup_source = None
self.assertRaises(shade.OpenStackCloudUnavailableFeature,
self.cloud.list_security_groups)
self.assertFalse(mock_neutron.list_security_groups.called)
self.assertFalse(mock_nova.security_groups.list.called)
class TestShadeOperator(base.TestCase):