Added OS APIs to associate/disassociate security groups to/from instances.

I will add views to return list of security groups associated with the servers later after this branch is merged into trunk. The reason I will do this later is because my previous merge proposal (https://code.launchpad.net/~tpatil/nova/add-options-network-create-os-apis/+merge/68292) is dependent on this work. In this merge proposal I have added a new extension which still uses default OS v1.1 controllers and views, but I am going to override views in this extension to send extra information like security groups.

1 files changed, 271 insertions, 0 deletions

diff --git a/nova/tests/api/openstack/contrib/test_security_groups.py b/nova/tests/api/openstack/contrib/test_security_groups.py
index 4317880ca..b44ebc9fb 100644
--- a/nova/tests/api/openstack/contrib/test_security_groups.py
+++ b/nova/tests/api/openstack/contrib/test_security_groups.py
@@ -15,10 +15,13 @@
 #    under the License.
 
 import json
+import mox
+import nova
 import unittest
 import webob
 from xml.dom import minidom
 
+from nova import exception
 from nova import test
 from nova.api.openstack.contrib import security_groups
 from nova.tests.api.openstack import fakes
@@ -51,6 +54,28 @@ def _create_security_group_request_dict(security_group):
     return {'security_group': sg}
 
 
+def return_server(context, server_id):
+    return {'id': server_id, 'state': 0x01, 'host': "localhost"}
+
+
+def return_non_running_server(context, server_id):
+    return {'id': server_id, 'state': 0x02,
+                        'host': "localhost"}
+
+
+def return_security_group(context, project_id, group_name):
+    return {'id': 1, 'name': group_name, "instances": [
+              {'id': 1}]}
+
+
+def return_security_group_without_instances(context, project_id, group_name):
+    return {'id': 1, 'name': group_name}
+
+
+def return_server_nonexistant(context, server_id):
+    raise exception.InstanceNotFound(instance_id=server_id)
+
+
 class TestSecurityGroups(test.TestCase):
     def setUp(self):
         super(TestSecurityGroups, self).setUp()
@@ -325,6 +350,252 @@ class TestSecurityGroups(test.TestCase):
         response = self._delete_security_group(11111111)
         self.assertEquals(response.status_int, 404)
 
+    def test_associate_by_non_existing_security_group_name(self):
+        body = dict(addSecurityGroup=dict(name='non-existing'))
+        req = webob.Request.blank('/v1.1/servers/1/action')
+        req.headers['Content-Type'] = 'application/json'
+        req.method = 'POST'
+        req.body = json.dumps(body)
+        response = req.get_response(fakes.wsgi_app())
+        self.assertEquals(response.status_int, 404)
+
+    def test_associate_by_invalid_server_id(self):
+        body = dict(addSecurityGroup=dict(name='test'))
+        self.stubs.Set(nova.db, 'security_group_get_by_name',
+                       return_security_group)
+        req = webob.Request.blank('/v1.1/servers/invalid/action')
+        req.headers['Content-Type'] = 'application/json'
+        req.method = 'POST'
+        req.body = json.dumps(body)
+        response = req.get_response(fakes.wsgi_app())
+        self.assertEquals(response.status_int, 400)
+
+    def test_associate_without_body(self):
+        req = webob.Request.blank('/v1.1/servers/1/action')
+        body = dict(addSecurityGroup=None)
+        self.stubs.Set(nova.db, 'instance_get', return_server)
+        req.headers['Content-Type'] = 'application/json'
+        req.method = 'POST'
+        req.body = json.dumps(body)
+        response = req.get_response(fakes.wsgi_app())
+        self.assertEquals(response.status_int, 400)
+
+    def test_associate_no_security_group_name(self):
+        req = webob.Request.blank('/v1.1/servers/1/action')
+        body = dict(addSecurityGroup=dict())
+        self.stubs.Set(nova.db, 'instance_get', return_server)
+        req.headers['Content-Type'] = 'application/json'
+        req.method = 'POST'
+        req.body = json.dumps(body)
+        response = req.get_response(fakes.wsgi_app())
+        self.assertEquals(response.status_int, 400)
+
+    def test_associate_security_group_name_with_whitespaces(self):
+        req = webob.Request.blank('/v1.1/servers/1/action')
+        body = dict(addSecurityGroup=dict(name="   "))
+        self.stubs.Set(nova.db, 'instance_get', return_server)
+        req.headers['Content-Type'] = 'application/json'
+        req.method = 'POST'
+        req.body = json.dumps(body)
+        response = req.get_response(fakes.wsgi_app())
+        self.assertEquals(response.status_int, 400)
+
+    def test_associate_non_existing_instance(self):
+        self.stubs.Set(nova.db, 'instance_get', return_server_nonexistant)
+        body = dict(addSecurityGroup=dict(name="test"))
+        self.stubs.Set(nova.db, 'security_group_get_by_name',
+                       return_security_group)
+        req = webob.Request.blank('/v1.1/servers/10000/action')
+        req.headers['Content-Type'] = 'application/json'
+        req.method = 'POST'
+        req.body = json.dumps(body)
+        response = req.get_response(fakes.wsgi_app())
+        self.assertEquals(response.status_int, 404)
+
+    def test_associate_non_running_instance(self):
+        self.stubs.Set(nova.db, 'instance_get', return_non_running_server)
+        self.stubs.Set(nova.db, 'security_group_get_by_name',
+                       return_security_group_without_instances)
+        body = dict(addSecurityGroup=dict(name="test"))
+        req = webob.Request.blank('/v1.1/servers/1/action')
+        req.headers['Content-Type'] = 'application/json'
+        req.method = 'POST'
+        req.body = json.dumps(body)
+        response = req.get_response(fakes.wsgi_app())
+        self.assertEquals(response.status_int, 400)
+
+    def test_associate_already_associated_security_group_to_instance(self):
+        self.stubs.Set(nova.db, 'instance_get', return_server)
+        self.stubs.Set(nova.db, 'security_group_get_by_name',
+                       return_security_group)
+        body = dict(addSecurityGroup=dict(name="test"))
+        req = webob.Request.blank('/v1.1/servers/1/action')
+        req.headers['Content-Type'] = 'application/json'
+        req.method = 'POST'
+        req.body = json.dumps(body)
+        response = req.get_response(fakes.wsgi_app())
+        self.assertEquals(response.status_int, 400)
+
+    def test_associate(self):
+        self.stubs.Set(nova.db, 'instance_get', return_server)
+        self.mox.StubOutWithMock(nova.db, 'instance_add_security_group')
+        nova.db.instance_add_security_group(mox.IgnoreArg(),
+                                    mox.IgnoreArg(),
+                                    mox.IgnoreArg())
+        self.stubs.Set(nova.db, 'security_group_get_by_name',
+                       return_security_group_without_instances)
+        self.mox.ReplayAll()
+
+        body = dict(addSecurityGroup=dict(name="test"))
+        req = webob.Request.blank('/v1.1/servers/1/action')
+        req.headers['Content-Type'] = 'application/json'
+        req.method = 'POST'
+        req.body = json.dumps(body)
+        response = req.get_response(fakes.wsgi_app())
+        self.assertEquals(response.status_int, 202)
+
+    def test_associate_xml(self):
+        self.stubs.Set(nova.db, 'instance_get', return_server)
+        self.mox.StubOutWithMock(nova.db, 'instance_add_security_group')
+        nova.db.instance_add_security_group(mox.IgnoreArg(),
+                                    mox.IgnoreArg(),
+                                    mox.IgnoreArg())
+        self.stubs.Set(nova.db, 'security_group_get_by_name',
+                       return_security_group_without_instances)
+        self.mox.ReplayAll()
+
+        req = webob.Request.blank('/v1.1/servers/1/action')
+        req.headers['Content-Type'] = 'application/xml'
+        req.method = 'POST'
+        req.body = """<addSecurityGroup>
+                           <name>test</name>
+                    </addSecurityGroup>"""
+        response = req.get_response(fakes.wsgi_app())
+        self.assertEquals(response.status_int, 202)
+
+    def test_disassociate_by_non_existing_security_group_name(self):
+        body = dict(removeSecurityGroup=dict(name='non-existing'))
+        req = webob.Request.blank('/v1.1/servers/1/action')
+        req.headers['Content-Type'] = 'application/json'
+        req.method = 'POST'
+        req.body = json.dumps(body)
+        response = req.get_response(fakes.wsgi_app())
+        self.assertEquals(response.status_int, 404)
+
+    def test_disassociate_by_invalid_server_id(self):
+        body = dict(removeSecurityGroup=dict(name='test'))
+        self.stubs.Set(nova.db, 'security_group_get_by_name',
+                       return_security_group)
+        req = webob.Request.blank('/v1.1/servers/invalid/action')
+        req.headers['Content-Type'] = 'application/json'
+        req.method = 'POST'
+        req.body = json.dumps(body)
+        response = req.get_response(fakes.wsgi_app())
+        self.assertEquals(response.status_int, 400)
+
+    def test_disassociate_without_body(self):
+        req = webob.Request.blank('/v1.1/servers/1/action')
+        body = dict(removeSecurityGroup=None)
+        self.stubs.Set(nova.db, 'instance_get', return_server)
+        req.headers['Content-Type'] = 'application/json'
+        req.method = 'POST'
+        req.body = json.dumps(body)
+        response = req.get_response(fakes.wsgi_app())
+        self.assertEquals(response.status_int, 400)
+
+    def test_disassociate_no_security_group_name(self):
+        req = webob.Request.blank('/v1.1/servers/1/action')
+        body = dict(removeSecurityGroup=dict())
+        self.stubs.Set(nova.db, 'instance_get', return_server)
+        req.headers['Content-Type'] = 'application/json'
+        req.method = 'POST'
+        req.body = json.dumps(body)
+        response = req.get_response(fakes.wsgi_app())
+        self.assertEquals(response.status_int, 400)
+
+    def test_disassociate_security_group_name_with_whitespaces(self):
+        req = webob.Request.blank('/v1.1/servers/1/action')
+        body = dict(removeSecurityGroup=dict(name="   "))
+        self.stubs.Set(nova.db, 'instance_get', return_server)
+        req.headers['Content-Type'] = 'application/json'
+        req.method = 'POST'
+        req.body = json.dumps(body)
+        response = req.get_response(fakes.wsgi_app())
+        self.assertEquals(response.status_int, 400)
+
+    def test_disassociate_non_existing_instance(self):
+        self.stubs.Set(nova.db, 'instance_get', return_server_nonexistant)
+        body = dict(removeSecurityGroup=dict(name="test"))
+        self.stubs.Set(nova.db, 'security_group_get_by_name',
+                       return_security_group)
+        req = webob.Request.blank('/v1.1/servers/10000/action')
+        req.headers['Content-Type'] = 'application/json'
+        req.method = 'POST'
+        req.body = json.dumps(body)
+        response = req.get_response(fakes.wsgi_app())
+        self.assertEquals(response.status_int, 404)
+
+    def test_disassociate_non_running_instance(self):
+        self.stubs.Set(nova.db, 'instance_get', return_non_running_server)
+        self.stubs.Set(nova.db, 'security_group_get_by_name',
+                       return_security_group)
+        body = dict(removeSecurityGroup=dict(name="test"))
+        req = webob.Request.blank('/v1.1/servers/1/action')
+        req.headers['Content-Type'] = 'application/json'
+        req.method = 'POST'
+        req.body = json.dumps(body)
+        response = req.get_response(fakes.wsgi_app())
+        self.assertEquals(response.status_int, 400)
+
+    def test_disassociate_already_associated_security_group_to_instance(self):
+        self.stubs.Set(nova.db, 'instance_get', return_server)
+        self.stubs.Set(nova.db, 'security_group_get_by_name',
+                       return_security_group_without_instances)
+        body = dict(removeSecurityGroup=dict(name="test"))
+        req = webob.Request.blank('/v1.1/servers/1/action')
+        req.headers['Content-Type'] = 'application/json'
+        req.method = 'POST'
+        req.body = json.dumps(body)
+        response = req.get_response(fakes.wsgi_app())
+        self.assertEquals(response.status_int, 400)
+
+    def test_disassociate(self):
+        self.stubs.Set(nova.db, 'instance_get', return_server)
+        self.mox.StubOutWithMock(nova.db, 'instance_remove_security_group')
+        nova.db.instance_remove_security_group(mox.IgnoreArg(),
+                                    mox.IgnoreArg(),
+                                    mox.IgnoreArg())
+        self.stubs.Set(nova.db, 'security_group_get_by_name',
+                       return_security_group)
+        self.mox.ReplayAll()
+
+        body = dict(removeSecurityGroup=dict(name="test"))
+        req = webob.Request.blank('/v1.1/servers/1/action')
+        req.headers['Content-Type'] = 'application/json'
+        req.method = 'POST'
+        req.body = json.dumps(body)
+        response = req.get_response(fakes.wsgi_app())
+        self.assertEquals(response.status_int, 202)
+
+    def test_disassociate_xml(self):
+        self.stubs.Set(nova.db, 'instance_get', return_server)
+        self.mox.StubOutWithMock(nova.db, 'instance_remove_security_group')
+        nova.db.instance_remove_security_group(mox.IgnoreArg(),
+                                    mox.IgnoreArg(),
+                                    mox.IgnoreArg())
+        self.stubs.Set(nova.db, 'security_group_get_by_name',
+                       return_security_group)
+        self.mox.ReplayAll()
+
+        req = webob.Request.blank('/v1.1/servers/1/action')
+        req.headers['Content-Type'] = 'application/xml'
+        req.method = 'POST'
+        req.body = """<removeSecurityGroup>
+                           <name>test</name>
+                    </removeSecurityGroup>"""
+        response = req.get_response(fakes.wsgi_app())
+        self.assertEquals(response.status_int, 202)
+
 
 class TestSecurityGroupRules(test.TestCase):
     def setUp(self):