remote_group_id will also remaining membership update management
"""
+ @abc.abstractmethod
def prepare_port_filter(self, port):
"""Prepare filters for the port.
This method should be called before the port is created.
"""
- raise NotImplementedError()
def apply_port_filter(self, port):
"""Apply port filter.
"""
raise NotImplementedError()
+ @abc.abstractmethod
def update_port_filter(self, port):
"""Refresh security group rules from data store
the security group the port is a member of or if the
group gains or looses a rule.
"""
- raise NotImplementedError()
def remove_port_filter(self, port):
"""Stop filtering port."""
group='SECURITYGROUP')
+class FakeFirewallDriver(firewall_base.FirewallDriver):
+ """Fake FirewallDriver
+
+ FirewallDriver is base class for other types of drivers. To be able to
+ use it in tests, it's needed to overwrite all abstract methods.
+ """
+ def prepare_port_filter(self, port):
+ raise NotImplementedError()
+
+ def update_port_filter(self, port):
+ raise NotImplementedError()
+
+
class SecurityGroupRpcTestPlugin(test_sg.SecurityGroupTestPlugin,
sg_db_rpc.SecurityGroupServerRpcMixin):
def __init__(self):
mock.patch('neutron.agent.linux.iptables_manager').start()
self.default_firewall = self.agent.firewall
self.firewall = mock.Mock()
- firewall_object = firewall_base.FirewallDriver()
+ firewall_object = FakeFirewallDriver()
self.firewall.defer_apply.side_effect = firewall_object.defer_apply
self.agent.firewall = self.firewall
self.fake_device = {'device': 'fake_device',