x/vmware-nsx
Code Issues Proposed changes

Merge "NSX-V3| Do not allow adding QoS to router ports"

Browse Source
This commit is contained in:
Jenkins 2017-01-21 19:14:34 +00:00 committed by Gerrit Code Review
commit 5ad22c9ec7
2 changed files with 53 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
vmware_nsx/plugins/nsx_v3/plugin.py
View File

@ -721,7 +721,7 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
def _assert_on_external_net_with_qos(self, net_data):
# Prevent creating/update external network with QoS policy
if validators.is_attr_set(net_data.get(qos_consts.QOS_POLICY_ID)):
err_msg = _("Cannot configure QOS on networks")
err_msg = _("Cannot configure QOS on external networks")
raise n_exc.InvalidInput(error_message=err_msg)
def get_subnets(self, context, filters=None, fields=None, sorts=None,
@ -1675,6 +1675,16 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
LOG.warning(err_msg)
raise n_exc.InvalidInput(error_message=err_msg)
def _assert_on_router_port_with_qos(self, port_data, device_owner):
# Prevent creating/update port with QoS policy
# on router-interface ports.
if (device_owner == l3_db.DEVICE_OWNER_ROUTER_INTF and
validators.is_attr_set(port_data.get(qos_consts.QOS_POLICY_ID))):
err_msg = _("Unable to update/create a router port with a QoS "
"policy")
LOG.warning(err_msg)
raise n_exc.InvalidInput(error_message=err_msg)
def _filter_ipv4_dhcp_fixed_ips(self, context, fixed_ips):
ips = []
for fixed_ip in fixed_ips:
@ -1923,6 +1933,9 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
self._assert_on_external_net_with_compute(port_data)
self._assert_on_external_net_port_with_qos(port_data)
self._assert_on_router_port_with_qos(
port_data, port_data.get('device_owner'))
neutron_db = super(NsxV3Plugin, self).create_port(context, port)
self._extension_manager.process_create_port(
context, port_data, neutron_db)
@ -2282,6 +2295,12 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
self._assert_on_external_net_with_compute(port['port'])
self._assert_on_external_net_port_with_qos(port['port'])
device_owner = (port['port']['device_owner']
if 'device_owner' in port['port']
else original_port.get('device_owner'))
self._assert_on_router_port_with_qos(
port['port'], device_owner)
old_mac_learning_state = original_port.get(mac_ext.MAC_LEARNING)
updated_port = super(NsxV3Plugin, self).update_port(context,
id, port)

34
vmware_nsx/tests/unit/nsx_v3/test_plugin.py
View File

@ -351,10 +351,42 @@ class TestPortsV2(test_plugin.TestPortsV2, NsxV3PluginTestCaseMixin,
data = {'port': {'network_id': network['network']['id'],
'tenant_id': self._tenant_id,
'qos_policy_id': policy_id}}
# Cannot add qos policy to a port on ext network
# Cannot add qos policy to a router port
self.assertRaises(n_exc.InvalidInput,
self.plugin.create_port, self.ctx, data)
def test_create_router_port_with_qos_fail(self):
with self.network() as network:
with self.subnet(network=network, cidr='10.0.0.0/24'):
policy_id = uuidutils.generate_uuid()
data = {'port': {'network_id': network['network']['id'],
'tenant_id': self._tenant_id,
'device_owner': 'network:router_interface',
'qos_policy_id': policy_id}}
# Cannot add qos policy to a router interface port
self.assertRaises(n_exc.InvalidInput,
self.plugin.create_port, self.ctx, data)
def test_update_router_port_with_qos_fail(self):
with self.network() as network:
with self.subnet(network=network, cidr='10.0.0.0/24'):
policy_id = uuidutils.generate_uuid()
data = {'port': {'network_id': network['network']['id'],
'tenant_id': self._tenant_id,
'name': 'qos_port',
'admin_state_up': True,
'fixed_ips': [],
'mac_address': '00:00:00:00:00:01',
'device_id': 'dummy',
'device_owner': ''}}
port = self.plugin.create_port(self.ctx, data)
policy_id = uuidutils.generate_uuid()
data['port'] = {'qos_policy_id': policy_id,
'device_owner': 'network:router_interface'}
# Cannot add qos policy to a router interface port
self.assertRaises(n_exc.InvalidInput,
self.plugin.update_port, self.ctx, port['id'], data)
def test_create_port_with_qos_on_net(self):
with self.network() as network:
policy_id = uuidutils.generate_uuid()