Browse Source

Firewall as a Service (FWaaS) Agent

blueprint: quantum-fwaas-agent

This is the first iteration of the FWaaS Agent with some basic
functionality to enable integration of Plugin - Agent - Driver.

An inheritance approach is taken with the L3 Agent to enable the
agent side messaging.

Unit tests, included, coverage being increased

Change-Id: Ib0970fdc4ad1ac53df66fba172a5a7f7d7ee3f1b
changes/64/34064/30
Sridar Kandaswamy 9 years ago
parent
commit
17ef53cda6
  1. 3
      etc/fwaas_driver.ini
  2. 9
      neutron/agent/l3_agent.py
  3. 9
      neutron/extensions/firewall.py
  4. 16
      neutron/services/firewall/agents/__init__.py
  5. 108
      neutron/services/firewall/agents/firewall_agent_api.py
  6. 16
      neutron/services/firewall/agents/l3reference/__init__.py
  7. 274
      neutron/services/firewall/agents/l3reference/firewall_l3_agent.py
  8. 9
      neutron/services/firewall/fwaas_plugin.py
  9. 15
      neutron/tests/unit/services/firewall/agents/__init__.py
  10. 15
      neutron/tests/unit/services/firewall/agents/l3reference/__init__.py
  11. 120
      neutron/tests/unit/services/firewall/agents/l3reference/test_firewall_l3_agent.py
  12. 85
      neutron/tests/unit/services/firewall/agents/test_firewall_agent_api.py

3
etc/fwaas_driver.ini

@ -0,0 +1,3 @@
[fwaas]
#driver = neutron.services.firewall.drivers.linux.iptables_fwaas.IptablesFwaasDriver
#enabled = True

9
neutron/agent/l3_agent.py

@ -43,7 +43,7 @@ from neutron.openstack.common.rpc import common as rpc_common
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-'
@ -138,7 +138,7 @@ class RouterInfo(object):
self._snat_action = None
class L3NATAgent(manager.Manager):
class L3NATAgent(firewall_l3_agent.FWaaSL3AgentRpcCallback, manager.Manager):
"""Manager for L3NatAgent
API version history:
@ -215,7 +215,7 @@ class L3NATAgent(manager.Manager):
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
@ -282,6 +282,7 @@ class L3NATAgent(manager.Manager):
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)
@ -700,6 +701,8 @@ class L3NATAgent(manager.Manager):
@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:

9
neutron/extensions/firewall.py

@ -89,6 +89,15 @@ class FirewallRuleInfoMissing(qexception.InvalidInput):
"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]

16
neutron/services/firewall/agents/__init__.py

@ -0,0 +1,16 @@
# 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.

108
neutron/services/firewall/agents/firewall_agent_api.py

@ -0,0 +1,108 @@
# 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

16
neutron/services/firewall/agents/l3reference/__init__.py

@ -0,0 +1,16 @@
# 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.

274
neutron/services/firewall/agents/l3reference/firewall_l3_agent.py

@ -0,0 +1,274 @@
# 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')

9
neutron/services/firewall/fwaas_plugin.py

@ -21,6 +21,7 @@ from oslo.config import cfg
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
@ -83,6 +84,14 @@ class FirewallCallbacks(object):
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."""

15
neutron/tests/unit/services/firewall/agents/__init__.py

@ -0,0 +1,15 @@
# 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.

15
neutron/tests/unit/services/firewall/agents/l3reference/__init__.py

@ -0,0 +1,15 @@
# 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.

120
neutron/tests/unit/services/firewall/agents/l3reference/test_firewall_l3_agent.py

@ -0,0 +1,120 @@
# 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')

85
neutron/tests/unit/services/firewall/agents/test_firewall_agent_api.py

@ -0,0 +1,85 @@
# 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')
Loading…
Cancel
Save