--- /dev/null
+# 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.
+#
+
+"""nsx_sec_group_mapping
+
+Revision ID: 1b2580001654
+Revises: abc88c33f74f
+Create Date: 2013-12-27 13:02:42.894648
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '1b2580001654'
+down_revision = 'abc88c33f74f'
+
+# Change to ['*'] if this migration applies to all plugins
+
+migration_for_plugins = [
+ 'neutron.plugins.nicira.NeutronPlugin.NvpPluginV2',
+ 'neutron.plugins.nicira.NeutronServicePlugin.NvpAdvancedPlugin',
+ 'neutron.plugins.vmware.plugin.NsxPlugin',
+ 'neutron.plugins.vmware.plugin.NsxServicePlugin'
+]
+
+from alembic import op
+import sqlalchemy as sa
+
+from neutron.db import migration
+
+
+def upgrade(active_plugins=None, options=None):
+ if not migration.should_run(active_plugins, migration_for_plugins):
+ return
+ # Create table for security group mappings
+ op.create_table(
+ 'neutron_nsx_security_group_mappings',
+ sa.Column('neutron_id', sa.String(length=36), nullable=False),
+ sa.Column('nsx_id', sa.String(length=36), nullable=False),
+ sa.ForeignKeyConstraint(['neutron_id'], ['securitygroups.id'],
+ ondelete='CASCADE'),
+ sa.PrimaryKeyConstraint('neutron_id', 'nsx_id'))
+ # Execute statement to add a record in security group mappings for
+ # each record in securitygroups
+ op.execute("INSERT INTO neutron_nsx_security_group_mappings SELECT id,id "
+ "from securitygroups")
+
+
+def downgrade(active_plugins=None, options=None):
+ if not migration.should_run(active_plugins, migration_for_plugins):
+ return
+ op.drop_table('neutron_nsx_security_group_mappings')
from neutron.plugins.nicira.common import config # noqa
from neutron.plugins.nicira.common import exceptions as nvp_exc
from neutron.plugins.nicira.common import nsx_utils
-from neutron.plugins.nicira.common import securitygroups as nvp_sec
+from neutron.plugins.nicira.common import securitygroups as sg_utils
from neutron.plugins.nicira.common import sync
from neutron.plugins.nicira.dbexts import db as nsx_db
from neutron.plugins.nicira.dbexts import distributedrouter as dist_rtr
l3_gwmode_db.L3_NAT_db_mixin,
mac_db.MacLearningDbMixin,
networkgw_db.NetworkGatewayMixin,
- nvp_sec.NVPSecurityGroups,
portbindings_db.PortBindingMixin,
portsecurity_db.PortSecurityDbMixin,
qos_db.NVPQoSDbMixin,
LOG.exception(err_desc)
raise nvp_exc.NvpPluginException(err_msg=err_desc)
- def _nvp_create_port_helper(self, cluster, ls_uuid, port_data,
+ def _nvp_create_port_helper(self, session, ls_uuid, port_data,
do_port_security=True):
- return switchlib.create_lport(cluster, ls_uuid, port_data['tenant_id'],
- port_data['id'], port_data['name'],
+ # Convert Neutron security groups identifiers into NSX security
+ # profiles identifiers
+ nsx_sec_profile_ids = [
+ nsx_utils.get_nsx_security_group_id(
+ session, self.cluster, neutron_sg_id) for
+ neutron_sg_id in (port_data[ext_sg.SECURITYGROUPS] or [])]
+ return switchlib.create_lport(self.cluster,
+ ls_uuid,
+ port_data['tenant_id'],
+ port_data['id'],
+ port_data['name'],
port_data['device_id'],
port_data['admin_state_up'],
port_data['mac_address'],
port_data['fixed_ips'],
port_data[psec.PORTSECURITY],
- port_data[ext_sg.SECURITYGROUPS],
+ nsx_sec_profile_ids,
port_data.get(qos.QUEUE),
port_data.get(mac_ext.MAC_LEARNING),
port_data.get(addr_pair.ADDRESS_PAIRS))
try:
selected_lswitch = self._nvp_find_lswitch_for_port(context,
port_data)
- lport = self._nvp_create_port_helper(self.cluster,
+ lport = self._nvp_create_port_helper(context.session,
selected_lswitch['uuid'],
port_data,
True)
context, port_data)
# Do not apply port security here!
ls_port = self._nvp_create_port_helper(
- self.cluster, selected_lswitch['uuid'],
+ context.session, selected_lswitch['uuid'],
port_data, False)
# Assuming subnet being attached is on first fixed ip
# element in port data
selected_lswitch = self._nvp_find_lswitch_for_port(
context, port_data)
lport = self._nvp_create_port_helper(
- self.cluster,
+ context.session,
selected_lswitch['uuid'],
port_data,
True)
tenant_id = self._get_tenant_id_for_create(context, s)
if not default_sg:
self._ensure_default_security_group(context, tenant_id)
-
- nsx_secgroup = secgrouplib.create_security_profile(self.cluster,
- tenant_id, s)
- security_group['security_group']['id'] = nsx_secgroup['uuid']
- return super(NvpPluginV2, self).create_security_group(
- context, security_group, default_sg)
+ # NOTE(salv-orlando): Pre-generating Neutron ID for security group.
+ neutron_id = str(uuid.uuid4())
+ nvp_secgroup = secgrouplib.create_security_profile(
+ self.cluster, neutron_id, tenant_id, s)
+ with context.session.begin(subtransactions=True):
+ s['id'] = neutron_id
+ sec_group = super(NvpPluginV2, self).create_security_group(
+ context, security_group, default_sg)
+ context.session.flush()
+ # Add mapping between neutron and nsx identifiers
+ nsx_db.add_neutron_nsx_security_group_mapping(
+ context.session, neutron_id, nvp_secgroup['uuid'])
+ return sec_group
def delete_security_group(self, context, security_group_id):
"""Delete a security group.
if super(NvpPluginV2, self)._get_port_security_group_bindings(
context, filters):
raise ext_sg.SecurityGroupInUse(id=security_group['id'])
+ nsx_sec_profile_id = nsx_utils.get_nsx_security_group_id(
+ context.session, self.cluster, security_group_id)
+
try:
secgrouplib.delete_security_profile(
- self.cluster, security_group['id'])
+ self.cluster, nsx_sec_profile_id)
except q_exc.NotFound:
- LOG.info(_("Security group: %s was already deleted "
- "from backend"), security_group_id)
+ # The security profile was not found on the backend
+ # do not fail in this case.
+ LOG.warning(_("The NSX security profile %(sec_profile_id)s, "
+ "associated with the Neutron security group "
+ "%(sec_group_id)s was not found on the backend"),
+ {'sec_profile_id': nsx_sec_profile_id,
+ 'sec_group_id': security_group_id})
+ except api_exc.NsxApiException:
+ # Raise and fail the operation, as there is a problem which
+ # prevented the sec group from being removed from the backend
+ LOG.exception(_("An exception occurred while removing the "
+ "NSX security profile %(sec_profile_id)s, "
+ "associated with Netron security group "
+ "%(sec_group_id)s"),
+ {'sec_profile_id': nsx_sec_profile_id,
+ 'sec_group_id': security_group_id})
+ raise nvp_exc.NvpPluginException(
+ _("Unable to remove security group %s from backend"),
+ security_group['id'])
return super(NvpPluginV2, self).delete_security_group(
context, security_group_id)
self._ensure_default_security_group(context, tenant_id)
security_group_id = self._validate_security_group_rules(
context, security_group_rule)
-
# Check to make sure security group exists
security_group = super(NvpPluginV2, self).get_security_group(
context, security_group_id)
# Check for duplicate rules
self._check_for_duplicate_rules(context, s)
# gather all the existing security group rules since we need all
- # of them to PUT to NVP.
- combined_rules = self._merge_security_group_rules_with_current(
- context, s, security_group['id'])
+ # of them to PUT to NSX.
+ existing_rules = self.get_security_group_rules(
+ context, {'security_group_id': [security_group['id']]})
+ combined_rules = sg_utils.merge_security_group_rules_with_current(
+ context.session, self.cluster, s, existing_rules)
+ nsx_sec_profile_id = nsx_utils.get_nsx_security_group_id(
+ context.session, self.cluster, security_group_id)
secgrouplib.update_security_group_rules(self.cluster,
- security_group['id'],
+ nsx_sec_profile_id,
combined_rules)
return super(
NvpPluginV2, self).create_security_group_rule_bulk_native(
raise ext_sg.SecurityGroupRuleNotFound(id=sgrid)
sgid = security_group_rule['security_group_id']
- current_rules = self._get_security_group_rules_nvp_format(
- context, sgid, True)
-
- self._remove_security_group_with_id_and_id_field(
- current_rules, sgrid)
+ current_rules = self.get_security_group_rules(
+ context, {'security_group_id': [sgid]})
+ current_rules_nsx = sg_utils.get_security_group_rules_nsx_format(
+ context.session, self.cluster, current_rules, True)
+
+ sg_utils.remove_security_group_with_id_and_id_field(
+ current_rules_nsx, sgrid)
+ nsx_sec_profile_id = nsx_utils.get_nsx_security_group_id(
+ context.session, self.cluster, sgid)
secgrouplib.update_security_group_rules(
- self.cluster, sgid, current_rules)
+ self.cluster, nsx_sec_profile_id, current_rules_nsx)
return super(NvpPluginV2, self).delete_security_group_rule(context,
sgrid)
from neutron.plugins.nicira.dbexts import db as nsx_db
from neutron.plugins.nicira import nsx_cluster
from neutron.plugins.nicira.nsxlib import router as routerlib
+from neutron.plugins.nicira.nsxlib import secgroup as secgrouplib
from neutron.plugins.nicira.nsxlib import switch as switchlib
-
LOG = log.getLogger(__name__)
return nsx_switch_id, nsx_port_id
-def create_nsx_cluster(cluster_opts, concurrent_connections, gen_timeout):
- cluster = nsx_cluster.NSXCluster(**cluster_opts)
-
- def _ctrl_split(x, y):
- return (x, int(y), True)
+def get_nsx_security_group_id(session, cluster, neutron_id):
+ """Return the NSX sec profile uuid for a given neutron sec group.
- api_providers = [_ctrl_split(*ctrl.split(':'))
- for ctrl in cluster.nsx_controllers]
- cluster.api_client = client.NsxApiClient(
- api_providers, cluster.nsx_user, cluster.nsx_password,
- concurrent_connections=concurrent_connections,
- gen_timeout=gen_timeout,
- request_timeout=cluster.req_timeout,
- http_timeout=cluster.http_timeout,
- retries=cluster.retries,
- redirects=cluster.redirects)
- return cluster
+ First, look up the Neutron database. If not found, execute
+ a query on NSX platform as the mapping might be missing.
+ NOTE: Security groups are called 'security profiles' on the NSX backend.
+ """
+ nsx_id = nsx_db.get_nsx_security_group_id(session, neutron_id)
+ if not nsx_id:
+ # Find security profile on backend.
+ # This is a rather expensive query, but it won't be executed
+ # more than once for each security group in Neutron's lifetime
+ nsx_sec_profiles = secgrouplib.query_security_profiles(
+ cluster, '*',
+ filters={'tag': neutron_id,
+ 'tag_scope': 'q_sec_group_id'})
+ # Only one result expected
+ # NOTE(salv-orlando): Not handling the case where more than one
+ # security profile is found with the same neutron port tag
+ if not nsx_sec_profiles:
+ LOG.warn(_("Unable to find NSX security profile for Neutron "
+ "security group %s"), neutron_id)
+ return
+ elif len(nsx_sec_profiles) > 1:
+ LOG.warn(_("Multiple NSX security profiles found for Neutron "
+ "security group %s"), neutron_id)
+ nsx_sec_profile = nsx_sec_profiles[0]
+ nsx_id = nsx_sec_profile['uuid']
+ with session.begin(subtransactions=True):
+ # Create DB mapping
+ nsx_db.add_neutron_nsx_security_group_mapping(
+ session, neutron_id, nsx_id)
+ return nsx_id
def get_nsx_router_id(session, cluster, neutron_router_id):
neutron_router_id,
nsx_router_id)
return nsx_router_id
+
+
+def create_nsx_cluster(cluster_opts, concurrent_connections, gen_timeout):
+ cluster = nsx_cluster.NSXCluster(**cluster_opts)
+
+ def _ctrl_split(x, y):
+ return (x, int(y), True)
+
+ api_providers = [_ctrl_split(*ctrl.split(':'))
+ for ctrl in cluster.nsx_controllers]
+ cluster.api_client = client.NsxApiClient(
+ api_providers, cluster.nsx_user, cluster.nsx_password,
+ request_timeout=cluster.req_timeout,
+ http_timeout=cluster.http_timeout,
+ retries=cluster.retries,
+ redirects=cluster.redirects,
+ concurrent_connections=concurrent_connections,
+ gen_timeout=gen_timeout)
+ return cluster
#
# @author: Aaron Rosen, Nicira Networks, Inc.
-from neutron.extensions import securitygroup as ext_sg
+from neutron.openstack.common import log
+from neutron.plugins.nicira.common import nsx_utils
+LOG = log.getLogger(__name__)
# Protocol number look up for supported protocols
protocol_num_look_up = {'tcp': 6, 'icmp': 1, 'udp': 17}
-class NVPSecurityGroups(object):
+def _convert_to_nsx_rule(session, cluster, rule, with_id=False):
+ """Converts a Neutron security group rule to the NSX format.
- def _convert_to_nvp_rule(self, rule, with_id=False):
- """Converts Neutron API security group rule to NVP API."""
- nvp_rule = {}
- params = ['remote_ip_prefix', 'protocol',
- 'remote_group_id', 'port_range_min',
- 'port_range_max', 'ethertype']
+ This routine also replaces Neutron IDs with NSX UUIDs.
+ """
+ nsx_rule = {}
+ params = ['remote_ip_prefix', 'protocol',
+ 'remote_group_id', 'port_range_min',
+ 'port_range_max', 'ethertype']
+ if with_id:
+ params.append('id')
+
+ for param in params:
+ value = rule.get(param)
+ if param not in rule:
+ nsx_rule[param] = value
+ elif not value:
+ pass
+ elif param == 'remote_ip_prefix':
+ nsx_rule['ip_prefix'] = rule['remote_ip_prefix']
+ elif param == 'remote_group_id':
+ nsx_rule['profile_uuid'] = nsx_utils.get_nsx_security_group_id(
+ session, cluster, rule['remote_group_id'])
+
+ elif param == 'protocol':
+ try:
+ nsx_rule['protocol'] = int(rule['protocol'])
+ except (ValueError, TypeError):
+ nsx_rule['protocol'] = (
+ protocol_num_look_up[rule['protocol']])
+ else:
+ nsx_rule[param] = value
+ return nsx_rule
+
+
+def _convert_to_nsx_rules(session, cluster, rules, with_id=False):
+ """Converts a list of Neutron security group rules to the NSX format."""
+ nsx_rules = {'logical_port_ingress_rules': [],
+ 'logical_port_egress_rules': []}
+ for direction in ['logical_port_ingress_rules',
+ 'logical_port_egress_rules']:
+ for rule in rules[direction]:
+ nsx_rules[direction].append(
+ _convert_to_nsx_rule(session, cluster, rule, with_id))
+ return nsx_rules
+
+
+def get_security_group_rules_nsx_format(session, cluster,
+ security_group_rules, with_id=False):
+ """Convert neutron security group rules into NSX format.
+
+ This routine splits Neutron security group rules into two lists, one
+ for ingress rules and the other for egress rules.
+ """
+
+ def fields(rule):
+ _fields = ['remote_ip_prefix', 'remote_group_id', 'protocol',
+ 'port_range_min', 'port_range_max', 'protocol', 'ethertype']
if with_id:
- params.append('id')
-
- for param in params:
- value = rule.get(param)
- if param not in rule:
- nvp_rule[param] = value
- elif not value:
- pass
- elif param == 'remote_ip_prefix':
- nvp_rule['ip_prefix'] = rule['remote_ip_prefix']
- elif param == 'remote_group_id':
- nvp_rule['profile_uuid'] = rule['remote_group_id']
- elif param == 'protocol':
- try:
- nvp_rule['protocol'] = int(rule['protocol'])
- except (ValueError, TypeError):
- nvp_rule['protocol'] = (
- protocol_num_look_up[rule['protocol']])
+ _fields.append('id')
+ return dict((k, v) for k, v in rule.iteritems() if k in _fields)
+
+ ingress_rules = []
+ egress_rules = []
+ for rule in security_group_rules:
+ if rule.get('souce_group_id'):
+ rule['remote_group_id'] = nsx_utils.get_nsx_security_group_id(
+ session, cluster, rule['remote_group_id'])
+
+ if rule['direction'] == 'ingress':
+ ingress_rules.append(fields(rule))
+ elif rule['direction'] == 'egress':
+ egress_rules.append(fields(rule))
+ rules = {'logical_port_ingress_rules': egress_rules,
+ 'logical_port_egress_rules': ingress_rules}
+ return _convert_to_nsx_rules(session, cluster, rules, with_id)
+
+
+def merge_security_group_rules_with_current(session, cluster,
+ new_rules, current_rules):
+ merged_rules = get_security_group_rules_nsx_format(
+ session, cluster, current_rules)
+ for new_rule in new_rules:
+ rule = new_rule['security_group_rule']
+ if rule['direction'] == 'ingress':
+ merged_rules['logical_port_egress_rules'].append(
+ _convert_to_nsx_rule(session, cluster, rule))
+ elif rule['direction'] == 'egress':
+ merged_rules['logical_port_ingress_rules'].append(
+ _convert_to_nsx_rule(session, cluster, rule))
+ return merged_rules
+
+
+def remove_security_group_with_id_and_id_field(rules, rule_id):
+ """Remove rule by rule_id.
+
+ This function receives all of the current rule associated with a
+ security group and then removes the rule that matches the rule_id. In
+ addition it removes the id field in the dict with each rule since that
+ should not be passed to nvp.
+ """
+ for rule_direction in rules.values():
+ item_to_remove = None
+ for port_rule in rule_direction:
+ if port_rule['id'] == rule_id:
+ item_to_remove = port_rule
else:
- nvp_rule[param] = value
- return nvp_rule
-
- def _convert_to_nvp_rules(self, rules, with_id=False):
- """Converts a list of Neutron API security group rules to NVP API."""
- nvp_rules = {'logical_port_ingress_rules': [],
- 'logical_port_egress_rules': []}
- for direction in ['logical_port_ingress_rules',
- 'logical_port_egress_rules']:
- for rule in rules[direction]:
- nvp_rules[direction].append(
- self._convert_to_nvp_rule(rule, with_id))
- return nvp_rules
-
- def _get_security_group_rules_nvp_format(self, context, security_group_id,
- with_id=False):
- """Query neutron db for security group rules."""
- fields = ['remote_ip_prefix', 'remote_group_id', 'protocol',
- 'port_range_min', 'port_range_max', 'protocol', 'ethertype']
- if with_id:
- fields.append('id')
-
- filters = {'security_group_id': [security_group_id],
- 'direction': ['ingress']}
- ingress_rules = self.get_security_group_rules(context, filters, fields)
- filters = {'security_group_id': [security_group_id],
- 'direction': ['egress']}
- egress_rules = self.get_security_group_rules(context, filters, fields)
- rules = {'logical_port_ingress_rules': egress_rules,
- 'logical_port_egress_rules': ingress_rules}
- return self._convert_to_nvp_rules(rules, with_id)
-
- def _get_profile_uuid(self, context, remote_group_id):
- """Return profile id from novas group id."""
- security_group = self.get_security_group(context, remote_group_id)
- if not security_group:
- raise ext_sg.SecurityGroupNotFound(id=remote_group_id)
- return security_group['id']
-
- def _merge_security_group_rules_with_current(self, context, new_rules,
- security_group_id):
- merged_rules = self._get_security_group_rules_nvp_format(
- context, security_group_id)
- for new_rule in new_rules:
- rule = new_rule['security_group_rule']
- rule['security_group_id'] = security_group_id
- if rule.get('souce_group_id'):
- rule['remote_group_id'] = self._get_profile_uuid(
- context, rule['remote_group_id'])
- if rule['direction'] == 'ingress':
- merged_rules['logical_port_egress_rules'].append(
- self._convert_to_nvp_rule(rule))
- elif rule['direction'] == 'egress':
- merged_rules['logical_port_ingress_rules'].append(
- self._convert_to_nvp_rule(rule))
- return merged_rules
-
- def _remove_security_group_with_id_and_id_field(self, rules, rule_id):
- """Remove rule by rule_id.
-
- This function receives all of the current rule associated with a
- security group and then removes the rule that matches the rule_id. In
- addition it removes the id field in the dict with each rule since that
- should not be passed to nvp.
- """
- for rule_direction in rules.values():
- item_to_remove = None
- for port_rule in rule_direction:
- if port_rule['id'] == rule_id:
- item_to_remove = port_rule
- else:
- # remove key from dictionary for NVP
- del port_rule['id']
- if item_to_remove:
- rule_direction.remove(item_to_remove)
+ # remove key from dictionary for NVP
+ del port_rule['id']
+ if item_to_remove:
+ rule_direction.remove(item_to_remove)
return mapping
+def add_neutron_nsx_security_group_mapping(session, neutron_id, nsx_id):
+ """Map a Neutron security group to a NSX security profile.
+
+ :param session: a valid database session object
+ :param neutron_id: a neutron security group identifier
+ :param nsx_id: a nsx security profile identifier
+ """
+ with session.begin(subtransactions=True):
+ mapping = models.NeutronNsxSecurityGroupMapping(
+ neutron_id=neutron_id, nsx_id=nsx_id)
+ session.add(mapping)
+ return mapping
+
+
def get_nsx_switch_ids(session, neutron_id):
# This function returns a list of NSX switch identifiers because of
# the possibility of chained logical switches
"stored in Neutron DB"), neutron_id)
+def get_nsx_security_group_id(session, neutron_id):
+ """Return the id of a security group in the NSX backend.
+
+ Note: security groups are called 'security profiles' in NSX
+ """
+ try:
+ mapping = (session.query(models.NeutronNsxSecurityGroupMapping).
+ filter_by(neutron_id=neutron_id).
+ one())
+ return mapping['nsx_id']
+ except exc.NoResultFound:
+ LOG.debug(_("NSX identifiers for neutron security group %s not yet "
+ "stored in Neutron DB"), neutron_id)
+ return None
+
+
def _delete_by_neutron_id(session, model, neutron_id):
return session.query(model).filter_by(neutron_id=neutron_id).delete()
nsx_id = Column(String(36), primary_key=True)
+class NeutronNsxSecurityGroupMapping(model_base.BASEV2):
+ """Backend mappings for Neutron Security Group identifiers.
+
+ This class maps a neutron security group identifier to the corresponding
+ NSX security profile identifier.
+ """
+
+ __tablename__ = 'neutron_nsx_security_group_mappings'
+ neutron_id = Column(String(36),
+ ForeignKey('securitygroups.id', ondelete="CASCADE"),
+ primary_key=True)
+ nsx_id = Column(String(36), primary_key=True)
+
+
class NeutronNsxPortMapping(model_base.BASEV2):
"""Represents the mapping between neutron and nvp port uuids."""
from neutron.common import exceptions
from neutron.openstack.common import log
from neutron.plugins.nicira.common import utils
+from neutron.plugins.nicira.nvplib import _build_uri_path
from neutron.plugins.nicira.nvplib import do_request
from neutron.plugins.nicira.nvplib import format_exception
+from neutron.plugins.nicira.nvplib import get_all_query_pages
HTTP_GET = "GET"
HTTP_POST = "POST"
HTTP_DELETE = "DELETE"
HTTP_PUT = "PUT"
+SECPROF_RESOURCE = "security-profile"
+
LOG = log.getLogger(__name__)
return json.dumps(kwargs, ensure_ascii=False)
-def create_security_profile(cluster, tenant_id, security_profile):
+def query_security_profiles(cluster, fields=None, filters=None):
+ return get_all_query_pages(
+ _build_uri_path(SECPROF_RESOURCE,
+ fields=fields,
+ filters=filters),
+ cluster)
+
+
+def create_security_profile(cluster, tenant_id, neutron_id, security_profile):
+ """Create a security profile on the NSX backend.
+
+ :param cluster: a NSX cluster object reference
+ :param tenant_id: identifier of the Neutron tenant
+ :param neutron_id: neutron security group identifier
+ :param security_profile: dictionary with data for
+ configuring the NSX security profile.
+ """
path = "/ws.v1/security-profile"
# Allow all dhcp responses and all ingress traffic
hidden_rules = {'logical_port_egress_rules':
[{'ethertype': 'IPv4'},
{'ethertype': 'IPv6'}]}
display_name = utils.check_and_truncate(security_profile.get('name'))
+ # NOTE(salv-orlando): neutron-id tags are prepended with 'q' for
+ # historical reasons
body = mk_body(
- tags=utils.get_tags(os_tid=tenant_id), display_name=display_name,
+ tags=utils.get_tags(os_tid=tenant_id, q_sec_group_id=neutron_id),
+ display_name=display_name,
logical_port_ingress_rules=(
hidden_rules['logical_port_ingress_rules']),
logical_port_egress_rules=hidden_rules['logical_port_egress_rules']
from neutron.common import exceptions
from neutron.plugins.nicira.nsxlib import secgroup as secgrouplib
from neutron.plugins.nicira import nvplib as nsx_utils
+from neutron.tests.unit import test_api_v2
from neutron.tests.unit.vmware.nsxlib import base
+_uuid = test_api_v2._uuid
+
class SecurityProfileTestCase(base.NsxlibTestCase):
def test_create_and_get_security_profile(self):
sec_prof = secgrouplib.create_security_profile(
- self.fake_cluster, 'pippo', {'name': 'test'})
+ self.fake_cluster, _uuid(), 'pippo', {'name': 'test'})
sec_prof_res = secgrouplib.do_request(
secgrouplib.HTTP_GET,
nsx_utils._build_uri_path('security-profile',
def test_create_and_get_default_security_profile(self):
sec_prof = secgrouplib.create_security_profile(
- self.fake_cluster, 'pippo', {'name': 'default'})
+ self.fake_cluster, _uuid(), 'pippo', {'name': 'default'})
sec_prof_res = nsx_utils.do_request(
secgrouplib.HTTP_GET,
nsx_utils._build_uri_path('security-profile',
def test_update_security_profile_rules(self):
sec_prof = secgrouplib.create_security_profile(
- self.fake_cluster, 'pippo', {'name': 'test'})
+ self.fake_cluster, _uuid(), 'pippo', {'name': 'test'})
ingress_rule = {'ethertype': 'IPv4'}
egress_rule = {'ethertype': 'IPv4', 'profile_uuid': 'xyz'}
new_rules = {'logical_port_egress_rules': [egress_rule],
def test_update_security_profile_rules_noingress(self):
sec_prof = secgrouplib.create_security_profile(
- self.fake_cluster, 'pippo', {'name': 'test'})
+ self.fake_cluster, _uuid(), 'pippo', {'name': 'test'})
hidden_ingress_rule = {'ethertype': 'IPv4',
'ip_prefix': '127.0.0.1/32'}
egress_rule = {'ethertype': 'IPv4', 'profile_uuid': 'xyz'}
def test_delete_security_profile(self):
sec_prof = secgrouplib.create_security_profile(
- self.fake_cluster, 'pippo', {'name': 'test'})
+ self.fake_cluster, _uuid(), 'pippo', {'name': 'test'})
secgrouplib.delete_security_profile(
self.fake_cluster, sec_prof['uuid'])
self.assertRaises(exceptions.NotFound,
par_id, child_res, res_id, 'doh'))
self.assertEqual(expected, result)
+ def _mock_sec_group_mapping_db_calls(self, ret_value):
+ mock.patch(nsx_method('get_nsx_security_group_id',
+ module_name='dbexts.db'),
+ return_value=ret_value).start()
+ mock.patch(nsx_method('add_neutron_nsx_security_group_mapping',
+ module_name='dbexts.db')).start()
+ self.addCleanup(mock.patch.stopall)
+
+ def _verify_get_nsx_sec_profile_id(self, exp_sec_prof_uuid):
+ # The nvplib and db calls are mocked, therefore the cluster
+ # and the neutron_id parameters can be set to None
+ sec_prof_uuid = nsx_utils.get_nsx_security_group_id(
+ db_api.get_session(), None, None)
+ self.assertEqual(exp_sec_prof_uuid, sec_prof_uuid)
+
+ def test_get_nsx_sec_profile_id_from_db_mappings(self):
+ # This test is representative of the 'standard' case in which the
+ # security group mapping was stored in the neutron db
+ exp_sec_prof_uuid = uuidutils.generate_uuid()
+ self._mock_sec_group_mapping_db_calls(exp_sec_prof_uuid)
+ self._verify_get_nsx_sec_profile_id(exp_sec_prof_uuid)
+
+ def test_get_nsx_sec_profile_id_no_db_mapping(self):
+ # This test is representative of the case where db mappings where not
+ # found for a given security profile identifier
+ exp_sec_prof_uuid = uuidutils.generate_uuid()
+ self._mock_sec_group_mapping_db_calls(None)
+ with mock.patch(nsx_method('query_security_profiles',
+ module_name='nsxlib.secgroup'),
+ return_value=[{'uuid': exp_sec_prof_uuid}]):
+ self._verify_get_nsx_sec_profile_id(exp_sec_prof_uuid)
+
+ def test_get_nsx_sec_profile_id_no_mapping_returns_None(self):
+ # This test verifies that the function returns None if the mapping
+ # are not found both in the db and in the backend
+ self._mock_sec_group_mapping_db_calls(None)
+ with mock.patch(nsx_method('query_security_profiles',
+ module_name='nsxlib.secgroup'),
+ return_value=[]):
+ self._verify_get_nsx_sec_profile_id(None)
+
class ClusterManagementTestCase(nsx_base.NsxlibTestCase):
Code Review / openstack-build / neutron-build.git / commitdiff