Browse Source

Merge "Improve "OVSFirewallDriver.process_trusted_ports"" into stable/stein

tags/14.1.0
Zuul Gerrit Code Review 3 weeks ago
parent
commit
42ce4d05ee
2 changed files with 25 additions and 9 deletions
  1. +13
    -3
      neutron/agent/linux/openvswitch_firewall/firewall.py
  2. +12
    -6
      neutron/tests/unit/agent/linux/openvswitch_firewall/test_firewall.py

+ 13
- 3
neutron/agent/linux/openvswitch_firewall/firewall.py View File

@@ -512,6 +512,9 @@ class OVSFirewallDriver(firewall.FirewallDriver):
raise exceptions.OVSFWPortNotFound(port_id=port_id)
return ovs_port

def get_ovs_ports(self, port_ids):
return self.int_br.br.get_vifs_by_ids(port_ids)

def _get_port_vlan_tag(self, port_name):
return get_tag_from_other_config(self.int_br.br, port_name)

@@ -677,8 +680,10 @@ class OVSFirewallDriver(firewall.FirewallDriver):

def process_trusted_ports(self, port_ids):
"""Pass packets from these ports directly to ingress pipeline."""
ovs_ports = self.get_ovs_ports(port_ids)
for port_id in port_ids:
self._initialize_egress_no_port_security(port_id)
self._initialize_egress_no_port_security(port_id,
ovs_ports=ovs_ports)
# yield to let other greenthreads proceed
eventlet.sleep(0)

@@ -758,9 +763,14 @@ class OVSFirewallDriver(firewall.FirewallDriver):
ovs_consts.ACCEPTED_EGRESS_TRAFFIC_NORMAL_TABLE)
)

def _initialize_egress_no_port_security(self, port_id):
def _initialize_egress_no_port_security(self, port_id, ovs_ports=None):
try:
ovs_port = self.get_ovs_port(port_id)
if ovs_ports is not None:
ovs_port = ovs_ports.get(port_id)
if not ovs_port:
raise exceptions.OVSFWPortNotFound(port_id=port_id)
else:
ovs_port = self.get_ovs_port(port_id)
vlan_tag = self._get_port_vlan_tag(ovs_port.port_name)
except exceptions.OVSFWTagNotFound:
# It's a patch port, don't set anything


+ 12
- 6
neutron/tests/unit/agent/linux/openvswitch_firewall/test_firewall.py View File

@@ -744,15 +744,21 @@ class TestOVSFirewallDriver(base.BaseTestCase):
self.firewall._remove_egress_no_port_security('foo')

def test_process_trusted_ports_caches_port_id(self):
self.firewall.process_trusted_ports(['port_id'])
self.assertIn('port_id', self.firewall.sg_port_map.unfiltered)
vif_port = ovs_lib.VifPort('name', 1, 'id', 'mac', mock.ANY)
with mock.patch.object(self.firewall.int_br.br, 'get_vifs_by_ids',
return_value={'port_id': vif_port}):
self.firewall.process_trusted_ports(['port_id'])
self.assertEqual(1, len(self.firewall.sg_port_map.unfiltered))
self.assertEqual(vif_port.ofport,
self.firewall.sg_port_map.unfiltered['port_id'])

def test_process_trusted_ports_port_not_found(self):
"""Check that exception is not propagated outside."""
self.mock_bridge.br.get_vif_port_by_id.return_value = None
self.firewall.process_trusted_ports(['port_id'])
# Processing should have failed so port is not cached
self.assertNotIn('port_id', self.firewall.sg_port_map.unfiltered)
with mock.patch.object(self.firewall.int_br.br, 'get_vifs_by_ids',
return_value={}):
self.firewall.process_trusted_ports(['port_id'])
# Processing should have failed so port is not cached
self.assertEqual(0, len(self.firewall.sg_port_map.unfiltered))

def test_remove_trusted_ports_clears_cached_port_id(self):
self.firewall.sg_port_map.unfiltered['port_id'] = 1


Loading…
Cancel
Save