--- /dev/null
+[fwaas]
+#driver = neutron.services.firewall.drivers.linux.iptables_fwaas.IptablesFwaasDriver
+#enabled = True
from neutron.openstack.common.rpc import proxy
from neutron.openstack.common import service
from neutron import service as neutron_service
-
+from neutron.services.firewall.agents.l3reference import firewall_l3_agent
LOG = logging.getLogger(__name__)
NS_PREFIX = 'qrouter-'
self._snat_action = None
-class L3NATAgent(manager.Manager):
+class L3NATAgent(firewall_l3_agent.FWaaSL3AgentRpcCallback, manager.Manager):
"""Manager for L3NatAgent
API version history:
self.rpc_loop = loopingcall.FixedIntervalLoopingCall(
self._rpc_loop)
self.rpc_loop.start(interval=RPC_LOOP_INTERVAL)
- super(L3NATAgent, self).__init__(host=self.conf.host)
+ super(L3NATAgent, self).__init__(conf=self.conf)
def _destroy_router_namespaces(self, only_router_id=None):
"""Destroy router namespaces on the host to eliminate all stale
for c, r in self.metadata_nat_rules():
ri.iptables_manager.ipv4['nat'].add_rule(c, r)
ri.iptables_manager.apply()
+ super(L3NATAgent, self).process_router_add(ri)
if self.conf.enable_metadata_proxy:
self._spawn_metadata_proxy(ri)
@periodic_task.periodic_task
@lockutils.synchronized('l3-agent', 'neutron-')
def _sync_routers_task(self, context):
+ if self.services_sync:
+ super(L3NATAgent, self).process_services_sync(context)
if not self.fullsync:
return
try:
"rule operation.")
+class FirewallInternalDriverError(qexception.NeutronException):
+ """Fwaas exception for all driver errors.
+
+ On any failure or exception in the driver, driver should log it and
+ raise this exception to the agent
+ """
+ message = _("%(driver)s: Internal driver error.")
+
+
fw_valid_protocol_values = [None, constants.TCP, constants.UDP, constants.ICMP]
fw_valid_action_values = [constants.FWAAS_ALLOW, constants.FWAAS_DENY]
--- /dev/null
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2013 OpenStack Foundation.
+# All Rights Reserved.
+#
+# 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.
--- /dev/null
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+#
+# Copyright (c) 2013 OpenStack Foundation
+# All Rights Reserved.
+#
+# 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.
+#
+# @author: Sumit Naiksatam, sumitnaiksatam@gmail.com, Big Switch Networks, Inc.
+# @author: Sridar Kandaswamy, skandasw@cisco.com, Cisco Systems, Inc.
+# @author: Dan Florea, dflorea@cisco.com, Cisco Systems, Inc.
+
+from oslo.config import cfg
+
+from neutron.openstack.common import log as logging
+from neutron.openstack.common.rpc import proxy
+
+LOG = logging.getLogger(__name__)
+
+FWaaSOpts = [
+ cfg.StrOpt(
+ 'driver',
+ default=('neutron.services.firewall.agents.firewall_agent_api.'
+ 'NoopFwaasDriver'),
+ help=_("Name of the FWaaS Driver")),
+ cfg.BoolOpt(
+ 'enabled',
+ default=False,
+ help=_("Enable FWaaS")),
+]
+cfg.CONF.register_opts(FWaaSOpts, 'fwaas')
+
+
+class FWaaSPluginApiMixin(proxy.RpcProxy):
+ """Agent side of the FWaaS agent to FWaaS Plugin RPC API."""
+
+ RPC_API_VERSION = '1.0'
+
+ def __init__(self, topic, host):
+ super(FWaaSPluginApiMixin,
+ self).__init__(topic=topic,
+ default_version=self.RPC_API_VERSION)
+ self.host = host
+
+ def set_firewall_status(self, context, firewall_id, status):
+ """Make a RPC to set the status of a firewall."""
+ return self.call(context,
+ self.make_msg('set_firewall_status', host=self.host,
+ firewall_id=firewall_id, status=status),
+ topic=self.topic)
+
+ def firewall_deleted(self, context, firewall_id):
+ """Make a RPC to indicate that the firewall resources are deleted."""
+ return self.call(context,
+ self.make_msg('firewall_deleted', host=self.host,
+ firewall_id=firewall_id),
+ topic=self.topic)
+
+
+class FWaaSAgentRpcCallbackMixin(object):
+ """Mixin for FWaaS agent Implementations."""
+
+ def __init__(self, host):
+
+ super(FWaaSAgentRpcCallbackMixin, self).__init__(host)
+
+ def create_firewall(self, context, firewall, host):
+ """Handle RPC cast from plugin to create a firewall."""
+ pass
+
+ def update_firewall(self, context, firewall, host):
+ """Handle RPC cast from plugin to update a firewall."""
+ pass
+
+ def delete_firewall(self, context, firewall, host):
+ """Handle RPC cast from plugin to delete a firewall."""
+ pass
+
+
+class NoopFwaasDriver(object):
+ """Noop Fwaas Driver.
+
+ Firewall driver which does nothing.
+ This driver is for disabling the firewall functionality.
+ Put in temporarily until Driver changes are integrated when
+ this will come in from there.
+ """
+
+ def create_firewall(self, apply_list, firewall):
+ pass
+
+ def delete_firewall(self, apply_list, firewall):
+ pass
+
+ def update_firewall(self, apply_list, firewall):
+ pass
+
+ def apply_default_policy(self, apply_list):
+ pass
--- /dev/null
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2013 OpenStack Foundation.
+# All Rights Reserved.
+#
+# 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.
--- /dev/null
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+#
+# Copyright (c) 2013 OpenStack Foundation.
+# All Rights Reserved.
+#
+# 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.
+#
+# @author: Sumit Naiksatam, sumitnaiksatam@gmail.com, Big Switch Networks, Inc.
+# @author: Sridar Kandaswamy, skandasw@cisco.com, Cisco Systems, Inc.
+# @author: Dan Florea, dflorea@cisco.com, Cisco Systems, Inc.
+
+from oslo.config import cfg
+
+from neutron.agent.common import config
+from neutron.agent.linux import ip_lib
+from neutron.common import topics
+from neutron import context
+from neutron.extensions import firewall as fw_ext
+from neutron.openstack.common import importutils
+from neutron.openstack.common import log as logging
+from neutron.plugins.common import constants
+from neutron.services.firewall.agents import firewall_agent_api as api
+
+LOG = logging.getLogger(__name__)
+
+
+class FWaaSL3PluginApi(api.FWaaSPluginApiMixin):
+ """Agent side of the FWaaS agent to FWaaS Plugin RPC API."""
+
+ def __init__(self, topic, host):
+ super(FWaaSL3PluginApi, self).__init__(topic, host)
+
+ def get_firewalls_for_tenant(self, context, **kwargs):
+ """Get the Firewalls with rules from the Plugin to send to driver."""
+ LOG.debug(_("Retrieve Firewall with rules from Plugin"))
+
+ return self.call(context,
+ self.make_msg('get_firewalls_for_tenant',
+ host=self.host),
+ topic=self.topic)
+
+ def get_tenants_with_firewalls(self, context, **kwargs):
+ """Get all Tenants that have Firewalls configured from plugin."""
+ LOG.debug(_("Retrieve Tenants with Firewalls configured from Plugin"))
+
+ return self.call(context,
+ self.make_msg('get_tenants_with_firewalls',
+ host=self.host),
+ topic=self.topic)
+
+
+class FWaaSL3AgentRpcCallback(api.FWaaSAgentRpcCallbackMixin):
+ """FWaaS Agent support to be used by Neutron L3 agent."""
+
+ def __init__(self, conf):
+ LOG.debug(_("Initializing firewall agent"))
+ self.conf = conf
+ fwaas_driver_class_path = cfg.CONF.fwaas.driver
+ self.fwaas_enabled = cfg.CONF.fwaas.enabled
+ try:
+ self.fwaas_driver = importutils.import_object(
+ fwaas_driver_class_path)
+ LOG.debug(_("FWaaS Driver Loaded: '%s'"), fwaas_driver_class_path)
+ except ImportError:
+ msg = _('Error importing FWaaS device driver: %s')
+ raise ImportError(msg % fwaas_driver_class_path)
+ self.services_sync = False
+ self.root_helper = config.get_root_helper(conf)
+ # setup RPC to msg fwaas plugin
+ self.fwplugin_rpc = FWaaSL3PluginApi(topics.FIREWALL_PLUGIN,
+ conf.host)
+ super(FWaaSL3AgentRpcCallback, self).__init__(host=conf.host)
+
+ def _get_router_info_list_for_tenant(self, routers, tenant_id):
+ """Returns the list of router info objects on which to apply the fw."""
+ root_ip = ip_lib.IPWrapper(self.root_helper)
+ # Get the routers for the tenant
+ router_ids = [
+ router['id']
+ for router in routers
+ if router['tenant_id'] == tenant_id]
+ local_ns_list = root_ip.get_namespaces(self.root_helper)
+
+ router_info_list = []
+ # Pick up namespaces for Tenant Routers
+ for rid in router_ids:
+ if self.router_info[rid].use_namespaces:
+ router_ns = self.router_info[rid].ns_name()
+ if router_ns in local_ns_list:
+ router_info_list.append(self.router_info[rid])
+ else:
+ router_info_list.append(self.router_info[rid])
+ return router_info_list
+
+ def _invoke_driver_for_plugin_api(self, context, fw, func_name):
+ """Invoke driver method for plugin API and provide status back."""
+ LOG.debug(_("%(func_name)s from agent for fw: %(fwid)s"),
+ {'func_name': func_name, 'fwid': fw['id']})
+ try:
+ routers = self.plugin_rpc.get_routers(context)
+ router_info_list = self._get_router_info_list_for_tenant(
+ routers,
+ fw['tenant_id'])
+ if not router_info_list:
+ LOG.debug(_('No Routers on tenant: %s'), fw['tenant_id'])
+ return
+ LOG.debug(_("Apply fw on Router List: '%s'"),
+ [ri.router['id'] for ri in router_info_list])
+ # call into the driver
+ try:
+ self.fwaas_driver.__getattribute__(func_name)(
+ router_info_list,
+ fw)
+ status = constants.ACTIVE
+ except fw_ext.FirewallInternalDriverError:
+ LOG.error(_("Firewall Driver Error for %(func_name)s "
+ "for fw: %(fwid)s"),
+ {'func_name': func_name, 'fwid': fw['id']})
+ status = constants.ERROR
+ # delete needs different handling
+ if func_name == 'delete_firewall':
+ if status == constants.ACTIVE:
+ self.fwplugin_rpc.firewall_deleted(context, fw['id'])
+ else:
+ self.fwplugin_rpc.set_firewall_status(
+ context,
+ fw['id'],
+ status)
+ except Exception:
+ LOG.exception(
+ _("FWaaS RPC failure in %(func_name)s for fw: %(fwid)s"),
+ {'func_name': func_name, 'fwid': fw['id']})
+ self.services_sync = True
+ return
+
+ def _invoke_driver_for_sync_from_plugin(self, ctx, router_info_list, fw):
+ """Invoke the delete driver method for status of PENDING_DELETE and
+ update method for all other status to (re)apply on driver which is
+ Idempotent.
+ """
+ if fw['status'] == constants.PENDING_DELETE:
+ try:
+ self.fwaas_driver.delete_firewall(router_info_list, fw)
+ self.fwplugin_rpc.firewall_deleted(
+ ctx,
+ fw['id'])
+ except fw_ext.FirewallInternalDriverError:
+ LOG.error(_("Firewall Driver Error on fw state %(fwmsg)s "
+ "for fw: %(fwid)s"),
+ {'fwmsg': fw['status'], 'fwid': fw['id']})
+ self.fwplugin_rpc.set_firewall_status(
+ ctx,
+ fw['id'],
+ constants.ERROR)
+ else:
+ # PENDING_UPDATE, PENDING_CREATE, ...
+ try:
+ self.fwaas_driver.update_firewall(router_info_list, fw)
+ status = constants.ACTIVE
+ except fw_ext.FirewallInternalDriverError:
+ LOG.error(_("Firewall Driver Error on fw state %(fwmsg)s "
+ "for fw: %(fwid)s"),
+ {'fwmsg': fw['status'], 'fwid': fw['id']})
+ status = constants.ERROR
+
+ self.fwplugin_rpc.set_firewall_status(
+ ctx,
+ fw['id'],
+ status)
+
+ def _process_router_add(self, ri):
+ """On router add, get fw with rules from plugin and update driver."""
+ LOG.debug(_("Process router add, router_id: '%s'"), ri.router['id'])
+ routers = []
+ routers.append(ri.router)
+ router_info_list = self._get_router_info_list_for_tenant(
+ routers,
+ ri.router['tenant_id'])
+ if router_info_list:
+ # Get the firewall with rules
+ # for the tenant the router is on.
+ ctx = context.Context('', ri.router['tenant_id'])
+ fw_list = self.fwplugin_rpc.get_firewalls_for_tenant(ctx)
+ LOG.debug(_("Process router add, fw_list: '%s'"),
+ [fw['id'] for fw in fw_list])
+ for fw in fw_list:
+ self._invoke_driver_for_sync_from_plugin(
+ ctx,
+ router_info_list,
+ fw)
+
+ def process_router_add(self, ri):
+ """On router add, get fw with rules from plugin and update driver."""
+ # avoid msg to plugin when fwaas is not configured
+ if not self.fwaas_enabled:
+ return
+ try:
+ self._process_router_add(ri)
+ except Exception:
+ LOG.exception(
+ _("FWaaS RPC info call failed for '%s'."),
+ ri.router['id'])
+ self.services_sync = True
+
+ def process_services_sync(self, ctx):
+ """On RPC issues sync with plugin and apply the sync data."""
+ try:
+ # get all routers
+ routers = self.plugin_rpc.get_routers(ctx)
+ # get the list of tenants with firewalls configured
+ # from the plugin
+ tenant_ids = self.fwplugin_rpc.get_tenants_with_firewalls(ctx)
+ LOG.debug(_("Tenants with Firewalls: '%s'"), tenant_ids)
+ for tenant_id in tenant_ids:
+ ctx = context.Context('', tenant_id)
+ fw_list = self.fwplugin_rpc.get_firewalls_for_tenant(ctx)
+ if fw_list:
+ # if fw present on tenant
+ router_info_list = self._get_router_info_list_for_tenant(
+ routers,
+ tenant_id)
+ if router_info_list:
+ LOG.debug(_("Router List: '%s'"),
+ [ri.router['id'] for ri in router_info_list])
+ LOG.debug(_("fw_list: '%s'"),
+ [fw['id'] for fw in fw_list])
+ # apply sync data on fw for this tenant
+ for fw in fw_list:
+ # fw, routers present on this host for tenant
+ # install
+ LOG.debug(_("Apply fw on Router List: '%s'"),
+ [ri.router['id']
+ for ri in router_info_list])
+ # no need to apply sync data for ACTIVE fw
+ if fw['status'] != constants.ACTIVE:
+ self._invoke_driver_for_sync_from_plugin(
+ ctx,
+ router_info_list,
+ fw)
+ self.services_sync = False
+ except Exception:
+ LOG.exception(_("Failed fwaas process services sync"))
+ self.services_sync = True
+
+ def create_firewall(self, context, firewall, host):
+ """Handle Rpc from plugin to create a firewall."""
+ return self._invoke_driver_for_plugin_api(
+ context,
+ firewall,
+ 'create_firewall')
+
+ def update_firewall(self, context, firewall, host):
+ """Handle Rpc from plugin to update a firewall."""
+ return self._invoke_driver_for_plugin_api(
+ context,
+ firewall,
+ 'update_firewall')
+
+ def delete_firewall(self, context, firewall, host):
+ """Handle Rpc from plugin to delete a firewall."""
+ return self._invoke_driver_for_plugin_api(
+ context,
+ firewall,
+ 'delete_firewall')
from neutron.common import rpc as q_rpc
from neutron.common import topics
+from neutron import context as neutron_context
from neutron.db import api as qdbapi
from neutron.db.firewall import firewall_db
from neutron.extensions import firewall as fw_ext
fw_list = [fw for fw in self.plugin.get_firewalls(context)]
return fw_list
+ def get_tenants_with_firewalls(self, context, **kwargs):
+ """Agent uses this to get all tenants that have firewalls."""
+ LOG.debug(_("get_tenants_with_firewalls() called"))
+ ctx = neutron_context.get_admin_context()
+ fw_list = self.plugin.get_firewalls(ctx)
+ fw_tenant_list = list(set(fw['tenant_id'] for fw in fw_list))
+ return fw_tenant_list
+
class FirewallAgentApi(proxy.RpcProxy):
"""Plugin side of plugin to agent RPC API."""
--- /dev/null
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2013 OpenStack Foundation
+#
+# 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.
--- /dev/null
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2013 OpenStack Foundation
+#
+# 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.
--- /dev/null
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+#
+# Copyright (c) 2013 OpenStack Foundation
+# All Rights Reserved.
+#
+# 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.
+#
+# @author: Sumit Naiksatam, sumitnaiksatam@gmail.com, Big Switch Networks, Inc.
+# @author: Sridar Kandaswamy, skandasw@cisco.com, Cisco Systems, Inc.
+# @author: Dan Florea, dflorea@cisco.com, Cisco Systems, Inc.
+
+import contextlib
+import mock
+from oslo.config import cfg
+
+from neutron.agent.common import config as agent_config
+from neutron.common import config as base_config
+from neutron.services.firewall.agents.l3reference import firewall_l3_agent
+from neutron.tests import base
+
+
+class FWaasHelper(object):
+ def __init__(self, host):
+ pass
+
+
+class FWaasAgent(firewall_l3_agent.FWaaSL3AgentRpcCallback, FWaasHelper):
+ def __init__(self, conf=None):
+ super(FWaasAgent, self).__init__(conf)
+
+
+class TestFwaasL3AgentRpcCallback(base.BaseTestCase):
+ def setUp(self):
+ super(TestFwaasL3AgentRpcCallback, self).setUp()
+ self.addCleanup(mock.patch.stopall)
+
+ self.conf = cfg.ConfigOpts()
+ self.conf.register_opts(base_config.core_opts)
+ agent_config.register_root_helper(self.conf)
+ self.conf.root_helper = 'sudo'
+ self.api = FWaasAgent(self.conf)
+
+ def test_create_firewall(self):
+ fake_firewall = {'id': 0}
+ with mock.patch.object(
+ self.api,
+ '_invoke_driver_for_plugin_api'
+ ) as mock_driver:
+ self.assertEqual(
+ self.api.create_firewall(
+ mock.sentinel.context,
+ fake_firewall,
+ 'host'),
+ mock_driver.return_value)
+
+ def test_update_firewall(self):
+ fake_firewall = {'id': 0}
+ with mock.patch.object(
+ self.api,
+ '_invoke_driver_for_plugin_api'
+ ) as mock_driver:
+ self.assertEqual(
+ self.api.update_firewall(
+ mock.sentinel.context,
+ fake_firewall,
+ 'host'),
+ mock_driver.return_value)
+
+ def test_delete_firewall(self):
+ fake_firewall = {'id': 0}
+ with mock.patch.object(
+ self.api,
+ '_invoke_driver_for_plugin_api'
+ ) as mock_driver:
+ self.assertEqual(
+ self.api.delete_firewall(
+ mock.sentinel.context,
+ fake_firewall,
+ 'host'),
+ mock_driver.return_value)
+
+ def test_invoke_driver_for_plugin_api(self):
+ fake_firewall = {'id': 0, 'tenant_id': 001}
+ self.api.plugin_rpc = mock.Mock()
+ with contextlib.nested(
+ mock.patch.object(self.api.plugin_rpc, 'get_routers'),
+ mock.patch.object(self.api, '_get_router_info_list_for_tenant'),
+ mock.patch.object(self.api.fwaas_driver, 'create_firewall'),
+ mock.patch.object(self.api.fwplugin_rpc, 'set_firewall_status')
+ ) as (
+ mock_get_routers,
+ mock_get_router_info_list_for_tenant,
+ mock_driver_create_firewall,
+ mock_set_firewall_status):
+
+ mock_driver_create_firewall.return_value = True
+ self.api.create_firewall(
+ context=mock.sentinel.context,
+ firewall=fake_firewall, host='host')
+
+ mock_get_routers.assert_called_once_with(
+ mock.sentinel.context)
+
+ mock_get_router_info_list_for_tenant.assert_called_once_with(
+ mock_get_routers.return_value, fake_firewall['tenant_id'])
+
+ mock_set_firewall_status.assert_called_once_with(
+ mock.sentinel.context,
+ fake_firewall['id'],
+ 'ACTIVE')
--- /dev/null
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+#
+# Copyright (c) 2013 OpenStack Foundation
+# All Rights Reserved.
+#
+# 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.
+#
+# @author: Sumit Naiksatam, sumitnaiksatam@gmail.com, Big Switch Networks, Inc.
+# @author: Sridar Kandaswamy, skandasw@cisco.com, Cisco Systems, Inc.
+# @author: Dan Florea, dflorea@cisco.com, Cisco Systems, Inc.
+
+import contextlib
+import mock
+
+from neutron.services.firewall.agents import firewall_agent_api as api
+from neutron.tests import base
+
+
+class TestFWaaSAgentApi(base.BaseTestCase):
+ def setUp(self):
+ super(TestFWaaSAgentApi, self).setUp()
+ self.addCleanup(mock.patch.stopall)
+
+ self.api = api.FWaaSPluginApiMixin(
+ 'topic',
+ 'host')
+
+ def test_init(self):
+ self.assertEqual(self.api.host, 'host')
+
+ def test_set_firewall_status(self):
+ with contextlib.nested(
+ mock.patch.object(self.api, 'make_msg'),
+ mock.patch.object(self.api, 'call')
+ ) as (mock_make_msg, mock_call):
+
+ self.assertEqual(
+ self.api.set_firewall_status(
+ mock.sentinel.context,
+ 'firewall_id',
+ 'status'),
+ mock_call.return_value)
+
+ mock_make_msg.assert_called_once_with(
+ 'set_firewall_status',
+ host='host',
+ firewall_id='firewall_id',
+ status='status')
+
+ mock_call.assert_called_once_with(
+ mock.sentinel.context,
+ mock_make_msg.return_value,
+ topic='topic')
+
+ def test_firewall_deleted(self):
+ with contextlib.nested(
+ mock.patch.object(self.api, 'make_msg'),
+ mock.patch.object(self.api, 'call')
+ ) as (mock_make_msg, mock_call):
+
+ self.assertEqual(
+ self.api.firewall_deleted(
+ mock.sentinel.context,
+ 'firewall_id'),
+ mock_call.return_value)
+
+ mock_make_msg.assert_called_once_with(
+ 'firewall_deleted',
+ host='host',
+ firewall_id='firewall_id')
+
+ mock_call.assert_called_once_with(
+ mock.sentinel.context,
+ mock_make_msg.return_value,
+ topic='topic')
Code Review / openstack-build / neutron-build.git / commitdiff