openstack
/
tap-as-a-service
Code Issues Proposed changes

Support service framework to separate plugin logic 

This fix supports service framework in order to separate
the plugin code into backend specific and generic part.

Change-Id: I66d0f4a9ce5453ec7ca0bf0fe52f24a3a2e3c8ea
Closes-Bug: #1567770
This commit is contained in:
Yoichiro Iura 2016-04-19 09:33:45 +00:00
parent a7d933dec0
commit 499cdfc7e5
9 changed files with 344 additions and 200 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
devstack/README.rst
View File

@ -8,3 +8,5 @@ A `local.conf` recipe to enable tap-as-a-service::
enable_plugin tap-as-a-service https://github.com/openstack/tap-as-a-service
enable_service taas
enable_service taas_openvswitch_agent
Q_PLUGIN_EXTRA_CONF_PATH=/etc/neutron
Q_PLUGIN_EXTRA_CONF_FILES=(taas_plugin.ini)

20
devstack/plugin.sh
View File

@ -17,24 +17,23 @@
# This script is meant to be sourced from devstack. It is a wrapper of
# devmido scripts that allows proper exporting of environment variables.
ABSOLUTE_PATH=$(cd `dirname "${BASH_SOURCE[0]}"` && pwd)
PLUGIN_PATH=$ABSOLUTE_PATH/..
TAAS_OVS_AGENT_BINARY="$NEUTRON_BIN_DIR/neutron-taas-openvswitch-agent"
TAAS_OVS_AGENT_CONF_FILE="/etc/neutron/taas.ini"
function install_taas {
pip_install --no-deps --editable $PLUGIN_PATH
pip_install --no-deps --editable $TAAS_PLUGIN_PATH
}
function configure_taas_plugin {
if [ ! -d $NEUTRON_CONF_DIR ]; then
_create_neutron_conf_dir
fi
cp -p $TAAS_PLUGIN_PATH/etc/taas_plugin.ini $TAAS_PLUGIN_CONF_FILE
_neutron_service_plugin_class_add taas
}
function configure_taas_openvswitch_agent {
local conf=$TAAS_OVS_AGENT_CONF_FILE
cp $PLUGIN_PATH/etc/taas.ini $conf
cp $TAAS_PLUGIN_PATH/etc/taas.ini $conf
iniset $conf taas driver neutron_taas.services.taas.drivers.linux.ovs_taas.OvsTaasDriver
iniset $conf taas enabled True
iniset $conf taas vlan_range_start 3000
@ -54,6 +53,13 @@ if is_service_enabled taas; then
configure_taas_plugin
elif [[ "$2" == "post-config" ]]; then
neutron-db-manage --subproject tap-as-a-service upgrade head
if is_service_enabled q-svc; then
echo "Configuring taas"
if [ "$TAAS_SERVICE_DRIVER" ]; then
inicomment $TAAS_PLUGIN_CONF_FILE service_providers service_provider
iniadd $TAAS_PLUGIN_CONF_FILE service_providers service_provider $TAAS_SERVICE_DRIVER
fi
fi
elif [[ "$2" == "extra" ]]; then
:
fi

9
devstack/settings Normal file
View File

@ -0,0 +1,9 @@
# Devstack settings
ABSOLUTE_PATH=$(cd `dirname "${BASH_SOURCE[0]}"` && pwd)
TAAS_PLUGIN_PATH=$ABSOLUTE_PATH/..
TAAS_PLUGIN_CONF_FILE="/etc/neutron/taas_plugin.ini"
TAAS_OVS_AGENT_BINARY="$NEUTRON_BIN_DIR/neutron-taas-openvswitch-agent"
TAAS_OVS_AGENT_CONF_FILE="/etc/neutron/taas.ini"
Q_PLUGIN_EXTRA_CONF_PATH=/etc/neutron
Q_PLUGIN_EXTRA_CONF_FILES=(taas_plugin.ini)
#TAAS_SERVICE_DRIVER=

7
etc/taas_plugin.ini Normal file
View File

@ -0,0 +1,7 @@
[DEFAULT]
[service_providers]
# Defines providers for advanced services using the format:
# <service_type>:<name>:<driver>[:default] (multi valued)
service_provider = TAAS:TAAS:neutron_taas.services.taas.service_drivers.taas_rpc.TaasRpcDriver:default

48
neutron_taas/services/taas/service_drivers/__init__.py Normal file
View File

@ -0,0 +1,48 @@
# Copyright (C) 2016 Midokura SARL.
# 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.
import abc
from oslo_log import log as logging
import six
LOG = logging.getLogger(__name__)
@six.add_metaclass(abc.ABCMeta)
class TaasDriver(object):
def __init__(self, service_plugin, validator=None):
self.service_plugin = service_plugin
@property
def service_type(self):
pass
@abc.abstractmethod
def create_tap_service(self, tap_service):
pass
@abc.abstractmethod
def delete_tap_service(self, tap_service):
pass
@abc.abstractmethod
def create_tap_flow(self, tap_service):
pass
@abc.abstractmethod
def delete_tap_flow(self, tap_service):
pass

81
neutron_taas/services/taas/service_drivers/taas_agent_api.py Normal file
View File

@ -0,0 +1,81 @@
# Copyright (C) 2016 Midokura SARL.
# Copyright (C) 2015 Ericsson AB
# Copyright (c) 2015 Gigamon
#
# 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.
from neutron.common import rpc as n_rpc
from oslo_log import log as logging
import oslo_messaging as messaging
LOG = logging.getLogger(__name__)
class TaasCallbacks(object):
"""Currently there are no callbacks to the Taas Plugin."""
def __init__(self, plugin):
super(TaasCallbacks, self).__init__()
self.plugin = plugin
return
class TaasAgentApi(object):
"""RPC calls to agent APIs"""
def __init__(self, topic, host):
self.host = host
target = messaging.Target(topic=topic, version='1.0')
self.client = n_rpc.get_client(target)
return
def create_tap_service(self, context, tap_service, host):
LOG.debug("In RPC Call for Create Tap Service: Host=%s, MSG=%s" %
(host, tap_service))
cctxt = self.client.prepare(fanout=True)
cctxt.cast(context, 'create_tap_service', tap_service=tap_service,
host=host)
return
def create_tap_flow(self, context, tap_flow_msg, host):
LOG.debug("In RPC Call for Create Tap Flow: Host=%s, MSG=%s" %
(host, tap_flow_msg))
cctxt = self.client.prepare(fanout=True)
cctxt.cast(context, 'create_tap_flow', tap_flow_msg=tap_flow_msg,
host=host)
return
def delete_tap_service(self, context, tap_service, host):
LOG.debug("In RPC Call for Delete Tap Service: Host=%s, MSG=%s" %
(host, tap_service))
cctxt = self.client.prepare(fanout=True)
cctxt.cast(context, 'delete_tap_service', tap_service=tap_service,
host=host)
return
def delete_tap_flow(self, context, tap_flow_msg, host):
LOG.debug("In RPC Call for Delete Tap Flow: Host=%s, MSG=%s" %
(host, tap_flow_msg))
cctxt = self.client.prepare(fanout=True)
cctxt.cast(context, 'delete_tap_flow', tap_flow_msg=tap_flow_msg,
host=host)
return

119
neutron_taas/services/taas/service_drivers/taas_rpc.py Normal file
View File

@ -0,0 +1,119 @@
# Copyright (C) 2016 Midokura SARL.
# Copyright (C) 2015 Ericsson AB
# Copyright (c) 2015 Gigamon
#
# 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.
from neutron.common import rpc as n_rpc
from neutron_taas.common import topics
from neutron_taas.extensions import taas as taas_ex
from neutron_taas.services.taas import service_drivers
from neutron_taas.services.taas.service_drivers import taas_agent_api
from oslo_config import cfg
from oslo_log import log as logging
LOG = logging.getLogger(__name__)
class TaasRpcDriver(service_drivers.TaasDriver):
"""Taas Rpc Service Driver class"""
def __init__(self, service_plugin, validator=None):
LOG.debug("Loading TaasRpcDriver.")
super(TaasRpcDriver, self).__init__(service_plugin, validator)
self.endpoints = [taas_agent_api.TaasCallbacks(service_plugin)]
self.conn = n_rpc.create_connection()
self.conn.create_consumer(topics.TAAS_PLUGIN,
self.endpoints, fanout=False)
self.conn.consume_in_threads()
self.agent_rpc = taas_agent_api.TaasAgentApi(
topics.TAAS_AGENT,
cfg.CONF.host
)
return
def get_taas_id(self, context, tf):
ts = self.service_plugin.get_tap_service(context,
tf['tap_service_id'])
taas_id = (self.service_plugin.get_tap_id_association(
context,
tap_service_id=ts['id'])['taas_id'] +
cfg.CONF.taas.vlan_range_start)
return taas_id
def create_tap_service(self, context, ts, tap_id_association):
taas_vlan_id = (tap_id_association['taas_id'] +
cfg.CONF.taas.vlan_range_start)
port = self.service_plugin._get_port_details(context, ts['port_id'])
host = port['binding:host_id']
if taas_vlan_id > cfg.CONF.taas.vlan_range_end:
raise taas_ex.TapServiceLimitReached()
rpc_msg = {'tap_service': ts,
'taas_id': taas_vlan_id,
'port': port}
self.agent_rpc.create_tap_service(context, rpc_msg, host)
return
def delete_tap_service(self, context, ts, tap_id_association):
"""Execute delete tap service Rpc to agent"""
taas_vlan_id = (tap_id_association['taas_id'] +
cfg.CONF.taas.vlan_range_start)
port = self.service_plugin._get_port_details(context, ts['port_id'])
host = port['binding:host_id']
rpc_msg = {'tap_service': ts,
'taas_id': taas_vlan_id,
'port': port}
self.agent_rpc.delete_tap_service(context, rpc_msg, host)
return
def create_tap_flow(self, context, tf):
taas_id = self.get_taas_id(context, tf)
# Extract the host where the source port is located
port = self.service_plugin._get_port_details(context,
tf['source_port'])
host = port['binding:host_id']
port_mac = port['mac_address']
# Send RPC message to both the source port host and
# tap service(destination) port host
rpc_msg = {'tap_flow': tf,
'port_mac': port_mac,
'taas_id': taas_id,
'port': port}
self.agent_rpc.create_tap_flow(context, rpc_msg, host)
def delete_tap_flow(self, context, tf):
taas_id = self.get_taas_id(context, tf)
# Extract the host where the source port is located
port = self.service_plugin._get_port_details(context,
tf['source_port'])
host = port['binding:host_id']
port_mac = port['mac_address']
# Send RPC message to both the source port host and
# tap service(destination) port host
rpc_msg = {'tap_flow': tf,
'port_mac': port_mac,
'taas_id': taas_id,
'port': port}
self.agent_rpc.delete_tap_flow(context, rpc_msg, host)

167
neutron_taas/services/taas/taas_plugin.py
View File

@ -1,3 +1,4 @@
# Copyright (C) 2016 Midokura SARL.
# Copyright (C) 2015 Ericsson AB
# Copyright (c) 2015 Gigamon
#
@ -13,76 +14,23 @@
# License for the specific language governing permissions and limitations
# under the License.
from oslo_config import cfg
import oslo_messaging as messaging
from neutron.common import rpc as n_rpc
from neutron_taas.common import topics
from neutron.common import exceptions as n_exc
from neutron.db import servicetype_db as st_db
from neutron.services import provider_configuration as pconf
from neutron.services import service_base
from neutron_taas.common import constants
from neutron_taas.db import taas_db
from neutron_taas.extensions import taas as taas_ex
from oslo_log import log as logging
LOG = logging.getLogger(__name__)
class TaasCallbacks(object):
"""Currently there are no callbacks to the Taas Plugin."""
def __init__(self, plugin):
super(TaasCallbacks, self).__init__()
self.plugin = plugin
return
class TaasAgentApi(object):
"""RPC calls to agent APIs"""
def __init__(self, topic, host):
self.host = host
target = messaging.Target(topic=topic, version='1.0')
self.client = n_rpc.get_client(target)
return
def create_tap_service(self, context, tap_service, host):
LOG.debug("In RPC Call for Create Tap Service: Host=%s, MSG=%s" %
(host, tap_service))
cctxt = self.client.prepare(fanout=True)
cctxt.cast(context, 'create_tap_service', tap_service=tap_service,
host=host)
return
def create_tap_flow(self, context, tap_flow_msg, host):
LOG.debug("In RPC Call for Create Tap Flow: Host=%s, MSG=%s" %
(host, tap_flow_msg))
cctxt = self.client.prepare(fanout=True)
cctxt.cast(context, 'create_tap_flow', tap_flow_msg=tap_flow_msg,
host=host)
return
def delete_tap_service(self, context, tap_service, host):
LOG.debug("In RPC Call for Delete Tap Service: Host=%s, MSG=%s" %
(host, tap_service))
cctxt = self.client.prepare(fanout=True)
cctxt.cast(context, 'delete_tap_service', tap_service=tap_service,
host=host)
return
def delete_tap_flow(self, context, tap_flow_msg, host):
LOG.debug("In RPC Call for Delete Tap Flow: Host=%s, MSG=%s" %
(host, tap_flow_msg))
cctxt = self.client.prepare(fanout=True)
cctxt.cast(context, 'delete_tap_flow', tap_flow_msg=tap_flow_msg,
host=host)
return
def add_provider_configuration(type_manager, service_type):
type_manager.add_provider_configuration(
service_type,
pconf.ProviderConfiguration('neutron_taas'))
class TaasPlugin(taas_db.Tass_db_Mixin):
@ -93,20 +41,24 @@ class TaasPlugin(taas_db.Tass_db_Mixin):
def __init__(self):
LOG.debug("TAAS PLUGIN INITIALIZED")
self.endpoints = [TaasCallbacks(self)]
self.conn = n_rpc.create_connection(new=True)
self.conn.create_consumer(
topics.TAAS_PLUGIN, self.endpoints, fanout=False)
self.conn.consume_in_threads()
self.agent_rpc = TaasAgentApi(
topics.TAAS_AGENT,
cfg.CONF.host
)
self.service_type_manager = st_db.ServiceTypeManager.get_instance()
add_provider_configuration(self.service_type_manager, constants.TAAS)
self._load_drivers()
self.driver = self._get_driver_for_provider(self.default_provider)
return
def _load_drivers(self):
"""Loads plugin-drivers specified in configuration."""
self.drivers, self.default_provider = service_base.load_drivers(
'TAAS', self)
def _get_driver_for_provider(self, provider):
if provider in self.drivers:
return self.drivers[provider]
raise n_exc.Invalid("Error retrieving driver for provider %s" %
provider)
def create_tap_service(self, context, tap_service):
LOG.debug("create_tap_service() called")
@ -136,15 +88,8 @@ class TaasPlugin(taas_db.Tass_db_Mixin):
context,
tap_service_id=ts['id'])
taas_vlan_id = (tap_id_association['taas_id'] +
cfg.CONF.taas.vlan_range_start)
if taas_vlan_id > cfg.CONF.taas.vlan_range_end:
raise taas_ex.TapServiceLimitReached()
rpc_msg = {'tap_service': ts, 'taas_id': taas_vlan_id, 'port': port}
self.agent_rpc.create_tap_service(context, rpc_msg, host)
# call service_driver
self.driver.create_tap_service(context, ts, tap_id_association)
return ts
@ -167,23 +112,10 @@ class TaasPlugin(taas_db.Tass_db_Mixin):
for t_f in t_f_collection:
self.delete_tap_flow(context, t_f['id'])
# Get the port and the host that it is on
port_id = ts['port_id']
port = self._get_port_details(context, port_id)
host = port['binding:host_id']
super(TaasPlugin, self).delete_tap_service(context, id)
taas_vlan_id = (tap_id_association['taas_id'] +
cfg.CONF.taas.vlan_range_start)
rpc_msg = {'tap_service': ts,
'taas_id': taas_vlan_id,
'port': port}
self.agent_rpc.delete_tap_service(context, rpc_msg, host)
# call service_driver
self.driver.delete_tap_service(context, ts, tap_id_association)
return ts
@ -199,30 +131,14 @@ class TaasPlugin(taas_db.Tass_db_Mixin):
ts = self.get_tap_service(context, t_f['tap_service_id'])
ts_tenant_id = ts['tenant_id']
taas_id = (self.get_tap_id_association(
context,
tap_service_id=ts['id'])['taas_id'] +
cfg.CONF.taas.vlan_range_start)
if tenant_id != ts_tenant_id:
raise taas_ex.TapServiceNotBelongToTenant()
# Extract the host where the source port is located
port = self._get_port_details(context, t_f['source_port'])
host = port['binding:host_id']
port_mac = port['mac_address']
# create tap flow in the db model
tf = super(TaasPlugin, self).create_tap_flow(context, tap_flow)
# Send RPC message to both the source port host and
# tap service(destination) port host
rpc_msg = {'tap_flow': tf,
'port_mac': port_mac,
'taas_id': taas_id,
'port': port}
self.agent_rpc.create_tap_flow(context, rpc_msg, host)
# call service_driver
self.driver.create_tap_flow(context, tf)
return tf
@ -231,24 +147,9 @@ class TaasPlugin(taas_db.Tass_db_Mixin):
tf = self.get_tap_flow(context, id)
taas_id = (self.get_tap_id_association(
context,
tf['tap_service_id'])['taas_id'] +
cfg.CONF.taas.vlan_range_start)
port = self._get_port_details(context, tf['source_port'])
host = port['binding:host_id']
port_mac = port['mac_address']
super(TaasPlugin, self).delete_tap_flow(context, id)
# Send RPC message to both the source port host and
# tap service(destination) port host
rpc_msg = {'tap_flow': tf,
'port_mac': port_mac,
'taas_id': taas_id,
'port': port}
self.agent_rpc.delete_tap_flow(context, rpc_msg, host)
# call service_driver
self.driver.delete_tap_flow(context, tf)
return tf

91
neutron_taas/tests/unit/services/taas/test_taas_plugin.py
View File

@ -18,7 +18,6 @@ import contextlib
import mock
import testtools
from oslo_config import cfg
from oslo_utils import uuidutils
import neutron.common.rpc as n_rpc
@ -28,6 +27,7 @@ from neutron.tests.unit import testlib_api
import neutron_taas.db.taas_db # noqa
import neutron_taas.extensions.taas as taas_ext
from neutron_taas.services.taas.service_drivers import taas_agent_api
from neutron_taas.services.taas import taas_plugin
@ -35,8 +35,16 @@ class TestTaasPlugin(testlib_api.SqlTestCase):
def setUp(self):
super(TestTaasPlugin, self).setUp()
mock.patch.object(n_rpc, 'create_connection', auto_spec=True).start()
mock.patch.object(taas_plugin, 'TaasCallbacks', auto_spec=True).start()
mock.patch.object(taas_plugin, 'TaasAgentApi', auto_spec=True).start()
mock.patch.object(taas_agent_api,
'TaasCallbacks', auto_spec=True).start()
mock.patch.object(taas_agent_api,
'TaasAgentApi', auto_spec=True).start()
self.driver = mock.MagicMock()
mock.patch('neutron.services.service_base.load_drivers',
return_value=({'dummy_provider': self.driver},
'dummy_provider')).start()
mock.patch('neutron.db.servicetype_db.ServiceTypeManager.get_instance',
return_value=mock.MagicMock()).start()
self._plugin = taas_plugin.TaasPlugin()
self._context = context.get_admin_context()
@ -74,14 +82,10 @@ class TestTaasPlugin(testlib_api.SqlTestCase):
return_value=self._port_details):
yield self._plugin.create_tap_service(self._context, req)
self._tap_service['id'] = mock.ANY
expected_msg = {
'tap_service': self._tap_service,
'taas_id': mock.ANY,
'port': self._port_details,
}
self._plugin.agent_rpc.assert_has_calls([
mock.call.create_tap_service(self._context, expected_msg,
self._host_id),
self.driver.assert_has_calls([
mock.call.create_tap_service(self._context, self._tap_service,
mock.ANY),
])
@contextlib.contextmanager
@ -97,15 +101,9 @@ class TestTaasPlugin(testlib_api.SqlTestCase):
yield self._plugin.create_tap_flow(self._context, req)
self._tap_flow['id'] = mock.ANY
self._tap_service['id'] = mock.ANY
expected_msg = {
'tap_flow': self._tap_flow,
'port_mac': self._port_details['mac_address'],
'taas_id': mock.ANY,
'port': self._port_details,
}
self._plugin.agent_rpc.assert_has_calls([
mock.call.create_tap_flow(self._context, expected_msg,
self._host_id),
self._plugin.driver.assert_has_calls([
mock.call.create_tap_flow(self._context, self._tap_flow),
])
def test_create_tap_service(self):
@ -117,51 +115,31 @@ class TestTaasPlugin(testlib_api.SqlTestCase):
with testtools.ExpectedException(taas_ext.PortDoesNotBelongToTenant), \
self.tap_service():
pass
self.assertEqual([], self._plugin.agent_rpc.mock_calls)
self.assertEqual([], self.driver.mock_calls)
def test_create_tap_service_reach_limit(self):
cfg.CONF.set_override('vlan_range_end', 3900, 'taas') # same as start
with testtools.ExpectedException(taas_ext.TapServiceLimitReached), \
self.tap_service():
pass
self.assertEqual([], self._plugin.agent_rpc.mock_calls)
# TODO(Yoichiro):Need to move this test to taas_rpc test
pass
def test_delete_tap_service(self):
with self.tap_service() as ts:
self._plugin.delete_tap_service(self._context, ts['id'])
expected_msg = {
'tap_service': self._tap_service,
'taas_id': mock.ANY,
'port': self._port_details,
}
self._plugin.agent_rpc.assert_has_calls([
mock.call.delete_tap_service(self._context, expected_msg,
self._host_id),
self.driver.assert_has_calls([
mock.call.delete_tap_service(self._context, self._tap_service,
mock.ANY),
])
def test_delete_tap_service_with_flow(self):
with self.tap_service() as ts, \
self.tap_flow(tap_service=ts['id']) as tf:
self._plugin.delete_tap_service(self._context, ts['id'])
expected_msg = {
'tap_service': self._tap_service,
'taas_id': mock.ANY,
'port': self._port_details,
}
self._plugin.agent_rpc.assert_has_calls([
mock.call.delete_tap_service(self._context, expected_msg,
self._host_id),
self._plugin.driver.assert_has_calls([
mock.call.delete_tap_service(self._context, self._tap_service,
mock.ANY),
])
self._tap_flow['id'] = tf['id']
expected_msg = {
'tap_flow': self._tap_flow,
'taas_id': mock.ANY,
'port_mac': self._port_details['mac_address'],
'port': self._port_details,
}
self._plugin.agent_rpc.assert_has_calls([
mock.call.delete_tap_flow(self._context, expected_msg,
self._host_id),
self._plugin.driver.assert_has_calls([
mock.call.delete_tap_flow(self._context, self._tap_flow),
])
def test_delete_tap_service_non_existent(self):
@ -183,13 +161,6 @@ class TestTaasPlugin(testlib_api.SqlTestCase):
self.tap_flow(tap_service=ts['id']) as tf:
self._plugin.delete_tap_flow(self._context, tf['id'])
self._tap_flow['id'] = tf['id']
expected_msg = {
'tap_flow': self._tap_flow,
'taas_id': mock.ANY,
'port_mac': self._port_details['mac_address'],
'port': self._port_details,
}
self._plugin.agent_rpc.assert_has_calls([
mock.call.delete_tap_flow(self._context, expected_msg,
self._host_id),
self._plugin.driver.assert_has_calls([
mock.call.delete_tap_flow(self._context, self._tap_flow),
])