class SecurityGroupPortBinding(model_base.BASEV2):
"""Represents binding between quantum ports and security profiles"""
- port_id = sa.Column(sa.String(36), sa.ForeignKey("ports.id"),
+ port_id = sa.Column(sa.String(36),
+ sa.ForeignKey("ports.id",
+ ondelete='CASCADE'),
primary_key=True)
security_group_id = sa.Column(sa.String(36),
sa.ForeignKey("securitygroups.id"),
rule.get('external_id')):
raise ext_sg.SecurityGroupNotProxyMode()
- protocol = rule.get('protocol')
- ethertype = rule.get('ethertype')
-
# Check that port_range's are valid
if (rule['port_range_min'] is None and
rule['port_range_max'] is None):
import mock
import webob.exc
-from quantum.api.extensions import PluginAwareExtensionManager
-from quantum.api.v2 import attributes
-from quantum.api.v2.router import APIRouter
-from quantum.common import config
from quantum.common.test_lib import test_config
from quantum import context
-from quantum.db import api as db
from quantum.db import db_base_plugin_v2
from quantum.db import securitygroups_db
from quantum.extensions import securitygroup as ext_sg
-from quantum.manager import QuantumManager
from quantum.openstack.common import cfg
from quantum.tests.unit import test_db_plugin
-from quantum.tests.unit import test_extensions
-from quantum.wsgi import JSONDeserializer
DB_PLUGIN_KLASS = ('quantum.tests.unit.test_extension_security_group.'
'SecurityGroupTestPlugin')
self._extend_port_dict_security_group(context, port)
return port
- def delete_port(self, context, id):
- session = context.session
- with session.begin(subtransactions=True):
- self._delete_port_security_group_bindings(context, id)
- super(SecurityGroupTestPlugin, self).delete_port(context, id)
-
def create_network(self, context, network):
tenant_id = self._get_tenant_id_for_create(context, network['network'])
self._ensure_default_security_group(context, tenant_id)