diff --git a/neutron_tempest_plugin/config.py b/neutron_tempest_plugin/config.py
index a40fb0d..aea79ad 100644
--- a/neutron_tempest_plugin/config.py
+++ b/neutron_tempest_plugin/config.py
@@ -220,6 +220,25 @@
 CONF.register_group(sfc_group)
 CONF.register_opts(SfcGroup, group="sfc")
 
+
+TaasGroup = [
+    cfg.StrOpt('provider_physical_network',
+               default='',
+               help='Physical network to be used for creating SRIOV network.'),
+    cfg.StrOpt('provider_segmentation_id',
+               default='',
+               help='Segmentation-id to be used for creating SRIOV network.'),
+    cfg.StrOpt('vlan_filter',
+               default='',
+               help='Comma separated list of VLANs to be mirrored '
+                    'for a Tap-Flow.'),
+]
+taas_group = cfg.OptGroup(name='taas',
+                          title='TaaS Tempest Options')
+CONF.register_group(taas_group)
+CONF.register_opts(TaasGroup, group="taas")
+
+
 config_opts_translator = {
     'project_network_cidr': 'tenant_network_cidr',
     'project_network_v6_cidr': 'tenant_network_v6_cidr',
diff --git a/neutron_tempest_plugin/scenario/base.py b/neutron_tempest_plugin/scenario/base.py
index cbe5df6..de3d8fa 100644
--- a/neutron_tempest_plugin/scenario/base.py
+++ b/neutron_tempest_plugin/scenario/base.py
@@ -615,3 +615,38 @@
         result = shell.execute_local_command(cmd)
         self.assertEqual(0, result.exit_status)
         return result.stdout
+
+    def _ensure_public_router(self, client=None, tenant_id=None):
+        """Retrieve a router for the given tenant id.
+
+        If a public router has been configured, it will be returned.
+
+        If a public router has not been configured, but a public
+        network has, a tenant router will be created and returned that
+        routes traffic to the public network.
+        """
+        if not client:
+            client = self.client
+        if not tenant_id:
+            tenant_id = client.tenant_id
+        router_id = CONF.network.public_router_id
+        network_id = CONF.network.public_network_id
+        if router_id:
+            body = client.show_router(router_id)
+            return body['router']
+        elif network_id:
+            router = self.create_router_by_client()
+            self.addCleanup(test_utils.call_and_ignore_notfound_exc,
+                            client.delete_router, router['id'])
+            kwargs = {'external_gateway_info': dict(network_id=network_id)}
+            router = client.update_router(router['id'], **kwargs)['router']
+            return router
+        else:
+            raise Exception("Neither of 'public_router_id' or "
+                            "'public_network_id' has been defined.")
+
+    def _update_router_admin_state(self, router, admin_state_up):
+        kwargs = dict(admin_state_up=admin_state_up)
+        router = self.client.update_router(
+            router['id'], **kwargs)['router']
+        self.assertEqual(admin_state_up, router['admin_state_up'])
diff --git a/neutron_tempest_plugin/tap_as_a_service/scenario/__init__.py b/neutron_tempest_plugin/tap_as_a_service/scenario/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/neutron_tempest_plugin/tap_as_a_service/scenario/__init__.py
diff --git a/neutron_tempest_plugin/tap_as_a_service/scenario/manager.py b/neutron_tempest_plugin/tap_as_a_service/scenario/manager.py
new file mode 100644
index 0000000..80389c1
--- /dev/null
+++ b/neutron_tempest_plugin/tap_as_a_service/scenario/manager.py
@@ -0,0 +1,293 @@
+# Copyright 2012 OpenStack Foundation
+# Copyright 2013 IBM Corp.
+# 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 netaddr
+from oslo_log import log
+from oslo_utils import netutils
+
+from tempest import config
+from tempest.lib.common.utils import data_utils
+from tempest.lib.common.utils import test_utils
+from tempest.lib import exceptions as lib_exc
+
+from neutron_tempest_plugin.scenario import base
+from neutron_tempest_plugin.tap_as_a_service.services import taas_client
+
+CONF = config.CONF
+
+LOG = log.getLogger(__name__)
+
+
+class BaseTaasScenarioTests(base.BaseTempestTestCase):
+
+    credentials = ['primary', 'admin']
+
+    @classmethod
+    def setup_clients(cls):
+        super(BaseTaasScenarioTests, cls).setup_clients()
+
+        cls.client = cls.os_primary.network_client
+        cls.admin_network_client = cls.os_admin.network_client
+
+        # Setup taas clients
+        cls.tap_services_client = taas_client.TapServicesClient(
+            cls.os_primary.auth_provider,
+            CONF.network.catalog_type,
+            CONF.network.region or CONF.identity.region,
+            endpoint_type=CONF.network.endpoint_type,
+            build_interval=CONF.network.build_interval,
+            build_timeout=CONF.network.build_timeout,
+            **cls.os_primary.default_params)
+        cls.tap_flows_client = taas_client.TapFlowsClient(
+            cls.os_primary.auth_provider,
+            CONF.network.catalog_type,
+            CONF.network.region or CONF.identity.region,
+            endpoint_type=CONF.network.endpoint_type,
+            build_interval=CONF.network.build_interval,
+            build_timeout=CONF.network.build_timeout,
+            **cls.os_primary.default_params)
+
+    def _create_subnet(self, network, subnets_client=None,
+                       namestart='subnet-smoke', **kwargs):
+        """Create a subnet for the given network
+
+        within the cidr block configured for tenant networks.
+        """
+        if not subnets_client:
+            subnets_client = self.client
+
+        def cidr_in_use(cidr, tenant_id):
+            """Check cidr existence
+
+            :returns: True if subnet with cidr already exist in tenant
+                  False else
+            """
+            cidr_in_use = self.os_admin.network_client.list_subnets(
+                tenant_id=tenant_id, cidr=cidr)['subnets']
+            return len(cidr_in_use) != 0
+
+        ip_version = kwargs.pop('ip_version', 4)
+
+        if ip_version == 6:
+            tenant_cidr = netaddr.IPNetwork(
+                CONF.network.project_network_v6_cidr)
+            num_bits = CONF.network.project_network_v6_mask_bits
+        else:
+            tenant_cidr = netaddr.IPNetwork(CONF.network.project_network_cidr)
+            num_bits = CONF.network.project_network_mask_bits
+
+        result = None
+        str_cidr = None
+        # Repeatedly attempt subnet creation with sequential cidr
+        # blocks until an unallocated block is found.
+        for subnet_cidr in tenant_cidr.subnet(num_bits):
+            str_cidr = str(subnet_cidr)
+            if cidr_in_use(str_cidr, tenant_id=network['tenant_id']):
+                continue
+
+            subnet = dict(
+                name=data_utils.rand_name(namestart),
+                network_id=network['id'],
+                tenant_id=network['tenant_id'],
+                cidr=str_cidr,
+                ip_version=ip_version,
+                **kwargs
+            )
+            try:
+                result = subnets_client.create_subnet(**subnet)
+                break
+            except lib_exc.Conflict as e:
+                is_overlapping_cidr = 'overlaps with another subnet' in str(e)
+                if not is_overlapping_cidr:
+                    raise
+        assert result is not None, 'Unable to allocate tenant network'
+
+        subnet = result['subnet']
+        assert subnet['cidr'] == str_cidr
+
+        self.addCleanup(test_utils.call_and_ignore_notfound_exc,
+                       subnets_client.delete_subnet, subnet['id'])
+
+        return subnet
+
+    def _get_server_port_id_and_ip4(self, server, ip_addr=None):
+        ports = self.os_admin.network_client.list_ports(
+            device_id=server['id'], fixed_ip=ip_addr)['ports']
+        # A port can have more than one IP address in some cases.
+        # If the network is dual-stack (IPv4 + IPv6), this port is associated
+        # with 2 subnets
+        p_status = ['ACTIVE']
+        # NOTE(vsaienko) With Ironic, instances live on separate hardware
+        # servers. Neutron does not bind ports for Ironic instances, as a
+        # result the port remains in the DOWN state.
+        # TODO(vsaienko) remove once bug: #1599836 is resolved.
+        if getattr(CONF.service_available, 'ironic', False):
+            p_status.append('DOWN')
+        port_map = [(p["id"], fxip["ip_address"])
+                    for p in ports
+                    for fxip in p["fixed_ips"]
+                    if netutils.is_valid_ipv4(fxip["ip_address"]) and
+                    p['status'] in p_status]
+        inactive = [p for p in ports if p['status'] != 'ACTIVE']
+        if inactive:
+            LOG.warning("Instance has ports that are not ACTIVE: %s", inactive)
+
+        self.assertNotEqual(0, len(port_map),
+                            "No IPv4 addresses found in: %s" % ports)
+        self.assertEqual(len(port_map), 1,
+                         "Found multiple IPv4 addresses: %s. "
+                         "Unable to determine which port to target."
+                         % port_map)
+        return port_map[0]
+
+    def _get_network_by_name(self, network_name):
+        net = self.os_admin.network_client.list_networks(
+            name=network_name)['networks']
+        self.assertNotEqual(len(net), 0,
+                            "Unable to get network by name: %s" % network_name)
+        return net[0]
+
+    def _run_in_background(self, sshclient, cmd):
+        runInBg = "nohup %s 2>&1 &" % cmd
+        sshclient.exec_command(runInBg)
+
+    def create_networks(self, networks_client=None,
+                        routers_client=None, subnets_client=None,
+                        dns_nameservers=None, port_security_enabled=True):
+        """Create a network with a subnet connected to a router.
+
+        The baremetal driver is a special case since all nodes are
+        on the same shared network.
+
+        :param dns_nameservers: list of dns servers to send to subnet.
+        :returns: network, subnet, router
+        """
+        if CONF.network.shared_physical_network:
+            # NOTE(Shrews): This exception is for environments where tenant
+            # credential isolation is available, but network separation is
+            # not (the current baremetal case). Likely can be removed when
+            # test account mgmt is reworked:
+            # https://blueprints.launchpad.net/tempest/+spec/test-accounts
+            if not CONF.compute.fixed_network_name:
+                m = 'fixed_network_name must be specified in config'
+                raise lib_exc.InvalidConfiguration(m)
+            network = self._get_network_by_name(
+                CONF.compute.fixed_network_name)
+            router = None
+            subnet = None
+        else:
+            network = self.create_network(
+                client=networks_client,
+                port_security_enabled=port_security_enabled)
+            router = self._ensure_public_router(client=routers_client)
+            subnet_kwargs = dict(network=network,
+                                 subnets_client=subnets_client)
+            # use explicit check because empty list is a valid option
+            if dns_nameservers is not None:
+                subnet_kwargs['dns_nameservers'] = dns_nameservers
+            subnet = self._create_subnet(**subnet_kwargs)
+            if not routers_client:
+                routers_client = self.client
+            router_id = router['id']
+            routers_client.add_router_interface_with_subnet_id(
+                router_id=router_id, subnet_id=subnet['id'])
+
+            # save a cleanup job to remove this association between
+            # router and subnet
+            self.addCleanup(
+                test_utils.call_and_ignore_notfound_exc,
+                routers_client.remove_router_interface_with_subnet_id,
+                router_id=router_id, subnet_id=subnet['id'])
+        return network, subnet, router
+
+    def _create_server_with_floatingip(self, use_taas_cloud_image=False,
+                                       provider_net=False, **kwargs):
+        network = self.network
+        if use_taas_cloud_image:
+            image = CONF.neutron_plugin_options.advanced_image_ref
+            flavor = CONF.neutron_plugin_options.advanced_image_flavor_ref
+        else:
+            flavor = CONF.compute.flavor_ref
+            image = CONF.compute.image_ref
+
+        if provider_net:
+            network = self.provider_network
+
+        port = self.create_port(
+            network=network, security_groups=[self.secgroup['id']], **kwargs)
+        self.addCleanup(test_utils.call_and_ignore_notfound_exc,
+                        self.client.delete_port, port['id'])
+
+        params = {
+            'flavor_ref': flavor,
+            'image_ref': image,
+            'key_name': self.keypair['name']
+        }
+        vm = self.create_server(networks=[{'port': port['id']}], **params)
+        self.wait_for_server_active(vm['server'])
+        self.wait_for_guest_os_ready(vm['server'])
+
+        fip = self.create_and_associate_floatingip(
+            port_id=port['id'])
+
+        return port, fip
+
+    def _setup_provider_network(self):
+        net = self._create_provider_network()
+        self._create_provider_subnet(net["id"])
+        return net
+
+    def _create_provider_network(self):
+        network_kwargs = {
+            "admin_state_up": True,
+            "shared": True,
+            "provider:network_type": "vlan",
+            "provider:physical_network":
+                CONF.taas.provider_physical_network,
+        }
+
+        segmentation_id = CONF.taas.provider_segmentation_id
+        if segmentation_id and segmentation_id == "0":
+            network_kwargs['provider:network_type'] = 'flat'
+        elif segmentation_id:
+            network_kwargs['provider:segmentation_id'] = segmentation_id
+
+        network = self.admin_network_client.create_network(
+            **network_kwargs)['network']
+        self.addCleanup(test_utils.call_and_ignore_notfound_exc,
+                        self.admin_network_client.delete_network,
+                        network['id'])
+
+        return network
+
+    def _create_provider_subnet(self, net_id):
+        subnet = dict(
+            network_id=net_id,
+            cidr="172.25.100.0/24",
+            ip_version=4,
+        )
+        result = self.admin_network_client.create_subnet(**subnet)
+        self.addCleanup(
+            test_utils.call_and_ignore_notfound_exc,
+            self.admin_network_client.delete_subnet, result['subnet']['id'])
+
+        self.admin_network_client.add_router_interface_with_subnet_id(
+            self.router['id'], subnet_id=result['subnet']['id'])
+
+        self.addCleanup(
+            test_utils.call_and_ignore_notfound_exc,
+            self.admin_network_client.remove_router_interface_with_subnet_id,
+            self.router['id'], subnet_id=result['subnet']['id'])
diff --git a/neutron_tempest_plugin/tap_as_a_service/scenario/test_taas.py b/neutron_tempest_plugin/tap_as_a_service/scenario/test_taas.py
new file mode 100644
index 0000000..5598fbe
--- /dev/null
+++ b/neutron_tempest_plugin/tap_as_a_service/scenario/test_taas.py
@@ -0,0 +1,249 @@
+# Copyright (c) 2015 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.
+
+from oslo_log import log as logging
+from tempest.common import utils
+from tempest import config
+from tempest.lib.common.utils import data_utils
+from tempest.lib.common.utils import test_utils
+from tempest.lib import decorators
+from tempest.lib import exceptions as lib_exc
+import testtools
+
+from neutron_tempest_plugin.tap_as_a_service.scenario import manager
+
+CONF = config.CONF
+LOG = logging.getLogger(__name__)
+
+
+# pylint: disable=too-many-ancestors
+class TestTaaS(manager.BaseTaasScenarioTests):
+    """Config Requirement in tempest.conf:
+
+    - project_network_cidr_bits- specifies the subnet range for each network
+    - project_network_cidr
+    - public_network_id.
+    """
+
+    @classmethod
+    @utils.requires_ext(extension='taas', service='network')
+    @utils.requires_ext(extension='security-group', service='network')
+    @utils.requires_ext(extension='router', service='network')
+    def skip_checks(cls):
+        super(TestTaaS, cls).skip_checks()
+
+    @classmethod
+    def resource_setup(cls):
+        super(TestTaaS, cls).resource_setup()
+        cls.keypair = cls.create_keypair()
+        cls.secgroup = cls.create_security_group(
+            name=data_utils.rand_name('secgroup'))
+        cls.create_loginable_secgroup_rule(secgroup_id=cls.secgroup['id'])
+        LOG.debug("TaaSScenarioTest Setup done.")
+
+    def _create_server(self, network, security_group=None):
+        """Create a server
+
+        Creates a server having a port on given network and security group.
+        """
+        keys = self.create_keypair()
+        kwargs = {}
+        if security_group is not None:
+            kwargs['security_groups'] = [{'name': security_group['name']}]
+        server = self.create_server(
+            key_name=keys['name'],
+            networks=[{'uuid': network['id']}],
+            flavor_ref=CONF.compute.flavor_ref,
+            image_ref=CONF.compute.image_ref,
+            **kwargs)
+        self.wait_for_server_active(server['server'])
+        self.wait_for_guest_os_ready(server['server'])
+        return server, keys
+
+    @testtools.skipUnless(CONF.taas.provider_physical_network,
+                          'Provider physical network parameter not provided.')
+    @utils.requires_ext(extension="provider", service="network")
+    def _create_network_sriov(self, networks_client=None,
+                              tenant_id=None,
+                              namestart='network-smoke-sriov-',
+                              port_security_enabled=True):
+        if not networks_client:
+            networks_client = self.networks_client
+        if not tenant_id:
+            tenant_id = networks_client.tenant_id
+        name = data_utils.rand_name(namestart)
+        network_kwargs = dict(name=name, tenant_id=tenant_id)
+        # Neutron disables port security by default so we have to check the
+        # config before trying to create the network with
+        # port_security_enabled
+        if CONF.network_feature_enabled.port_security:
+            network_kwargs['port_security_enabled'] = port_security_enabled
+
+        if CONF.network.port_vnic_type and \
+                CONF.network.port_vnic_type == 'direct':
+            network_kwargs['provider:network_type'] = 'vlan'
+            if CONF.taas_plugin_options.provider_segmentation_id:
+                if CONF.taas_plugin_options.provider_segmentation_id == '0':
+                    network_kwargs['provider:network_type'] = 'flat'
+                else:
+                    network_kwargs['provider:segmentation_id'] = \
+                        CONF.taas_plugin_options.provider_segmentation_id
+
+            network_kwargs['provider:physical_network'] = \
+                CONF.taas_plugin_options.provider_physical_network
+
+        result = networks_client.create_network(**network_kwargs)
+        network = result['network']
+        self.assertEqual(network['name'], name)
+        self.addCleanup(test_utils.call_and_ignore_notfound_exc,
+                        networks_client.delete_network,
+                        network['id'])
+        return network
+
+    @testtools.skipUnless(CONF.taas.provider_physical_network,
+                          'Provider physical network parameter not provided.')
+    @utils.requires_ext(extension="provider", service="network")
+    def create_networks_sriov(self, networks_client=None,
+                              routers_client=None, subnets_client=None,
+                              tenant_id=None, dns_nameservers=None,
+                              port_security_enabled=True):
+        """Create a network with a subnet connected to a router.
+
+        The baremetal driver is a special case since all nodes are
+        on the same shared network.
+
+        :param tenant_id: id of tenant to create resources in.
+        :param dns_nameservers: list of dns servers to send to subnet.
+        :returns: network, subnet, router
+        """
+        router = None
+        if CONF.network.shared_physical_network:
+            # NOTE(Shrews): This exception is for environments where tenant
+            # credential isolation is available, but network separation is
+            # not (the current baremetal case). Likely can be removed when
+            # test account mgmt is reworked:
+            # https://blueprints.launchpad.net/tempest/+spec/test-accounts
+            if not CONF.compute.fixed_network_name:
+                msg = 'fixed_network_name must be specified in config'
+                raise lib_exc.InvalidConfiguration(msg)
+            network = self._get_network_by_name(
+                CONF.compute.fixed_network_name)
+            subnet = None
+        else:
+            network = self._create_network_sriov(
+                networks_client=networks_client,
+                tenant_id=tenant_id,
+                port_security_enabled=port_security_enabled)
+            subnet_kwargs = dict(network=network,
+                                 subnets_client=subnets_client,
+                                 routers_client=routers_client)
+            # use explicit check because empty list is a valid option
+            if dns_nameservers is not None:
+                subnet_kwargs['dns_nameservers'] = dns_nameservers
+            subnet = self._create_subnet(**subnet_kwargs)
+        return network, subnet, router
+
+    def _create_topology(self):
+        """Topology
+
+        +----------+             +----------+
+        | "server" |             | "server" |
+        |  VM-1    |             |  VM-2    |
+        |          |             |          |
+        +----+-----+             +----+-----+
+             |                        |
+             |                        |
+        +----+----+----+----+----+----+-----+
+                            |
+                            |
+                            |
+                     +------+------+
+                     | "server"    |
+                     | tap-service |
+                     +-------------+
+        """
+        LOG.debug('Starting Topology Creation')
+        resp = {}
+        # Create Network1 and Subnet1.
+        vnic_type = CONF.network.port_vnic_type
+        if vnic_type == 'direct':
+            self.network1, self.subnet1, self.router1 = \
+                self.create_networks_sriov()
+        else:
+            self.network1, self.subnet1, self.router1 = self.create_networks()
+        resp['network1'] = self.network1
+        resp['subnet1'] = self.subnet1
+        resp['router1'] = self.router1
+
+        # Create a security group allowing icmp and ssh traffic.
+        self.security_group = self.create_security_group(
+            name=data_utils.rand_name('secgroup'))
+        self.create_loginable_secgroup_rule(
+            secgroup_id=self.security_group['id'])
+
+        # Create 3 VMs and assign them a floating IP each.
+        port1, server_floating_ip_1 = self._create_server_with_floatingip()
+        port2, server_floating_ip_2 = self._create_server_with_floatingip()
+        port3, server_floating_ip_3 = self._create_server_with_floatingip()
+
+        # Store the received information to be used later
+        resp['port1'] = port1
+        resp['server_floating_ip_1'] = server_floating_ip_1
+
+        resp['port2'] = port2
+        resp['server_floating_ip_2'] = server_floating_ip_2
+
+        resp['port3'] = port3
+        resp['server_floating_ip_3'] = server_floating_ip_3
+
+        return resp
+
+    @utils.services('network')
+    @utils.requires_ext(extension="taas-vlan-filter", service="network")
+    @decorators.attr(type='slow')
+    @decorators.idempotent_id('40903cbd-0e3c-464d-b311-dc77d3894e65')
+    def test_tap_flow_data_mirroring(self):
+        """Create test topology and TaaS resources
+
+        Creates test topology consisting of 3 servers, one routable network,
+        ports and TaaS resources, i.e. tap-service and tap-flow using those
+        ports.
+        """
+        self.network, self.subnet, self.router = self.create_networks()
+        topology = self._create_topology()
+
+        # Create Tap-Service.
+        tap_service = self.tap_services_client.create_tap_service(
+            port_id=topology['port1']['id'])['tap_service']
+
+        LOG.debug('TaaS Config options: vlan-filter: %s',
+                  CONF.taas.vlan_filter)
+
+        # Create Tap-Flow.
+        vnic_type = CONF.network.port_vnic_type
+        vlan_filter = None
+        if vnic_type == 'direct':
+            vlan_filter = '108-117,126,135-144'
+            if CONF.taas.vlan_filter:
+                vlan_filter = CONF.taas.vlan_filter
+            elif topology['network1']['provider:segmentation_id'] != '0':
+                vlan_filter = topology['network1']['provider:segmentation_id']
+
+        tap_flow = self.tap_flows_client.create_tap_flow(
+            tap_service_id=tap_service['id'], direction='BOTH',
+            source_port=topology['port3']['id'],
+            vlan_filter=vlan_filter)['tap_flow']
+
+        self.assertEqual(tap_flow['vlan_filter'], vlan_filter)
diff --git a/neutron_tempest_plugin/tap_as_a_service/scenario/test_traffic_impact.py b/neutron_tempest_plugin/tap_as_a_service/scenario/test_traffic_impact.py
new file mode 100644
index 0000000..e2b14c7
--- /dev/null
+++ b/neutron_tempest_plugin/tap_as_a_service/scenario/test_traffic_impact.py
@@ -0,0 +1,261 @@
+# Copyright (c) 2019 AT&T
+# 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.
+
+from contextlib import contextmanager
+from oslo_log import log
+import testtools
+
+from tempest.common import utils
+from tempest import config
+from tempest.lib.common.utils import data_utils
+from tempest.lib.common.utils.linux import remote_client
+from tempest.lib.common.utils import test_utils
+
+from tempest.lib import decorators
+
+from neutron_tempest_plugin.tap_as_a_service.scenario import manager
+
+
+CONF = config.CONF
+LOG = log.getLogger(__name__)
+
+
+class TestTaaSTrafficScenarios(manager.BaseTaasScenarioTests):
+
+    @classmethod
+    @utils.requires_ext(extension='taas', service='network')
+    @utils.requires_ext(extension='security-group', service='network')
+    @utils.requires_ext(extension='router', service='network')
+    def skip_checks(cls):
+        super(TestTaaSTrafficScenarios, cls).skip_checks()
+
+    @classmethod
+    def resource_setup(cls):
+        super(TestTaaSTrafficScenarios, cls).resource_setup()
+        cls.provider_network = None
+        cls.keypair = cls.create_keypair()
+        cls.secgroup = cls.create_security_group(
+            name=data_utils.rand_name('secgroup'))
+        cls.create_loginable_secgroup_rule(secgroup_id=cls.secgroup['id'])
+        cls.create_pingable_secgroup_rule(secgroup_id=cls.secgroup['id'])
+
+    @contextmanager
+    def _setup_topology(self, taas=True, use_taas_cloud_image=False,
+                        provider_net=False):
+        """Setup topology for the test
+
+           +------------+
+           | monitor vm |
+           +-----+------+
+                 |
+           +-----v---+
+        +--+ network <--+
+        |  +----^----+  |
+        |       |       |
+        |  +----+-+ +---+--+
+        |  | vm 1 | | vm 2 |
+        |  +------+ +------+
+        |
+        |  +--------+
+        +--> router |
+           +-----+--+
+                 |
+           +-----v------+
+           | public net |
+           +------------+
+       """
+        self.network, self.subnet, self.router = self.create_networks()
+        LOG.debug('Setup topology sbunet details: %s ', self.subnet)
+        if provider_net:
+            if CONF.taas.provider_physical_network:
+                self.provider_network = self._setup_provider_network()
+            else:
+                msg = "provider_physical_network not provided"
+                raise self.skipException(msg)
+
+        self.mon_port, mon_fip = self._create_server_with_floatingip(
+            use_taas_cloud_image=use_taas_cloud_image,
+            provider_net=provider_net)
+        LOG.debug('Setup topology monitor port: %s  ###  monitor FIP: %s ',
+                  self.mon_port, mon_fip)
+        self.left_port, self.left_fip = self._create_server_with_floatingip(
+            provider_net=provider_net)
+        LOG.debug('Setup topology left port: %s  ###  left FIP: %s ',
+                  self.left_port, self.left_fip)
+        self.right_port, self.right_fip = self._create_server_with_floatingip(
+            provider_net=provider_net)
+        LOG.debug('Setup topology right port: %s  ###  right FIP: %s ',
+                  self.right_port, self.right_fip)
+
+        if taas:
+            LOG.debug("Create TAAS service")
+            tap_service = self.tap_services_client.create_tap_service(
+                port_id=self.mon_port['id'])['tap_service']
+            self.addCleanup(test_utils.call_and_ignore_notfound_exc,
+                            self.client.delete_tap_service, tap_service['id'])
+            tap_flow = self.tap_flows_client.create_tap_flow(
+                tap_service_id=tap_service['id'], direction='BOTH',
+                source_port=self.left_port['id'])['tap_flow']
+            self.addCleanup(test_utils.call_and_ignore_notfound_exc,
+                            self.client.delete_tap_flow, tap_flow['id'])
+            tap_flow = self.tap_flows_client.create_tap_flow(
+                tap_service_id=tap_service['id'], direction='BOTH',
+                source_port=self.right_port['id'])['tap_flow']
+            self.addCleanup(test_utils.call_and_ignore_notfound_exc,
+                            self.client.delete_tap_flow, tap_flow['id'])
+
+        user = CONF.validation.image_ssh_user
+        if use_taas_cloud_image:
+            user = CONF.neutron_plugin_options.advanced_image_ssh_user
+
+        self.monitor_client = remote_client.RemoteClient(
+            mon_fip['floating_ip_address'], user,
+            pkey=self.keypair['private_key'])
+        self.monitor_client.validate_authentication()
+        self.left_client = remote_client.RemoteClient(
+            self.left_fip['floating_ip_address'],
+            CONF.validation.image_ssh_user,
+            pkey=self.keypair['private_key'])
+        self.left_client.validate_authentication()
+        self.right_client = remote_client.RemoteClient(
+            self.right_fip['floating_ip_address'],
+            CONF.validation.image_ssh_user,
+            pkey=self.keypair['private_key'])
+        self.right_client.validate_authentication()
+        yield
+
+    def _check_icmp_traffic(self):
+        log_location = "/tmp/tcpdumplog"
+
+        right_ip = self.right_port['fixed_ips'][0]['ip_address']
+        left_ip = self.left_port['fixed_ips'][0]['ip_address']
+
+        # Run tcpdump in background
+        self._run_in_background(self.monitor_client,
+                                "sudo tcpdump -n -nn > %s" % log_location)
+
+        # Ensure tcpdump is up and running
+        psax = self.monitor_client.exec_command("ps -ax")
+        self.assertTrue("tcpdump" in psax)
+
+        # Run traffic from left_vm to right_vm
+        LOG.debug('Check ICMP traffic: ping %s ', right_ip)
+        # self.left_client.exec_command(
+        #     "ping -c 50 %s" % self.right_fip['floating_ip_address'])
+        self.check_remote_connectivity(self.left_client, right_ip,
+                                       ping_count=50)
+
+        # Collect tcpdump results
+        output = self.monitor_client.exec_command("cat %s" % log_location)
+        self.assertLess(0, len(output))
+
+        looking_for = ["IP %s > %s: ICMP echo request" % (left_ip, right_ip),
+                       "IP %s > %s: ICMP echo reply" % (right_ip, left_ip)]
+
+        results = []
+        for tcpdump_line in looking_for:
+            results.append(tcpdump_line in output)
+
+        return all(results)
+
+    def _test_taas_connectivity(self, use_provider_net=False):
+        """Ensure TAAS doesn't break connectivity
+
+        This test creates TAAS service between two servers and checks that
+        it doesn't break basic connectivity between them.
+        """
+        # Check uninterrupted traffic between VMs
+        with self._setup_topology(provider_net=use_provider_net):
+            # Left to right
+            self.check_remote_connectivity(
+                self.left_client,
+                self.right_port['fixed_ips'][0]['ip_address'])
+
+            # Right to left
+            self.check_remote_connectivity(
+                self.right_client,
+                self.left_port['fixed_ips'][0]['ip_address'])
+
+            # TAAS vm to right
+            self.check_remote_connectivity(
+                self.monitor_client,
+                self.right_port['fixed_ips'][0]['ip_address'])
+
+            # TAAS vm to left
+            self.check_remote_connectivity(
+                self.monitor_client,
+                self.left_port['fixed_ips'][0]['ip_address'])
+
+    @decorators.idempotent_id('ff414b7d-e81c-47f2-b6c8-53bc2f1e9b00')
+    @decorators.attr(type='slow')
+    @utils.services('compute', 'network')
+    def test_taas_provider_network_connectivity(self):
+        self._test_taas_connectivity(use_provider_net=True)
+
+    @decorators.idempotent_id('e3c52e91-7abf-4dfd-8687-f7c071cdd333')
+    @decorators.attr(type='slow')
+    @utils.services('compute', 'network')
+    def test_taas_network_connectivity(self):
+        self._test_taas_connectivity(use_provider_net=False)
+
+    @decorators.idempotent_id('fcb15ca3-ef61-11e9-9792-f45c89c47e11')
+    @testtools.skipUnless(CONF.neutron_plugin_options.advanced_image_ref,
+                          'Cloud image not found.')
+    @decorators.attr(type='slow')
+    @utils.services('compute', 'network')
+    def test_taas_forwarded_traffic_positive(self):
+        """Check that TAAS forwards traffic as expected"""
+
+        with self._setup_topology(use_taas_cloud_image=True):
+            # Check that traffic was forwarded to TAAS service
+            self.assertTrue(self._check_icmp_traffic())
+
+    @decorators.idempotent_id('6c54d9c5-075a-4a1f-bbe6-12c3c9abf1e2')
+    @testtools.skipUnless(CONF.neutron_plugin_options.advanced_image_ref,
+                          'Cloud image not found.')
+    @decorators.attr(type='slow')
+    @utils.services('compute', 'network')
+    def test_taas_forwarded_traffic_negative(self):
+        """Check that TAAS doesn't forward traffic"""
+
+        with self._setup_topology(taas=False, use_taas_cloud_image=True):
+            # Check that traffic was NOT forwarded to TAAS service
+            self.assertFalse(self._check_icmp_traffic())
+
+    @decorators.idempotent_id('fcb15ca3-ef61-11e9-9792-f45c89c47e12')
+    @testtools.skipUnless(CONF.neutron_plugin_options.advanced_image_ref,
+                          'Cloud image not found.')
+    @decorators.attr(type='slow')
+    @utils.services('compute', 'network')
+    def test_taas_forwarded_traffic_provider_net_positive(self):
+        """Check that TAAS forwards traffic as expected in provider network"""
+
+        with self._setup_topology(use_taas_cloud_image=True,
+                                  provider_net=True):
+            # Check that traffic was forwarded to TAAS service
+            self.assertTrue(self._check_icmp_traffic())
+
+    @decorators.idempotent_id('6c54d9c5-075a-4a1f-bbe6-12c3c9abf1e3')
+    @testtools.skipUnless(CONF.neutron_plugin_options.advanced_image_ref,
+                          'Cloud image not found.')
+    @decorators.attr(type='slow')
+    @utils.services('compute', 'network')
+    def test_taas_forwarded_traffic_provider_net_negative(self):
+        """Check that TAAS doesn't forward traffic in provider network"""
+
+        with self._setup_topology(taas=False, use_taas_cloud_image=True,
+                                  provider_net=True):
+            # Check that traffic was NOT forwarded to TAAS service
+            self.assertFalse(self._check_icmp_traffic())
diff --git a/zuul.d/master_jobs.yaml b/zuul.d/master_jobs.yaml
index e27b01d..26bf860 100644
--- a/zuul.d/master_jobs.yaml
+++ b/zuul.d/master_jobs.yaml
@@ -1091,24 +1091,28 @@
         - taas-vlan-filter
       devstack_localrc:
         NETWORK_API_EXTENSIONS: "{{ (network_api_extensions_common + network_api_extensions_tempest) | join(',') }}"
-        DOWNLOAD_DEFAULT_IMAGES: false
-        IMAGE_URLS: "http://download.cirros-cloud.net/0.3.4/cirros-0.3.4-i386-disk.img,https://cloud-images.ubuntu.com/minimal/releases/focal/release/ubuntu-20.04-minimal-cloudimg-amd64.img"
-        DEFAULT_IMAGE_NAME: cirros-0.3.4-i386-disk
-        ADVANCED_IMAGE_NAME: ubuntu-20.04-minimal-cloudimg-amd64
+        IMAGE_URLS: https://cloud-images.ubuntu.com/releases/bionic/release/ubuntu-18.04-server-cloudimg-amd64.img
+        ADVANCED_IMAGE_NAME: ubuntu-18.04-server-cloudimg-amd64
+        ADVANCED_INSTANCE_TYPE: ntp_image_384M
+        ADVANCED_INSTANCE_USER: ubuntu
+        CUSTOMIZE_IMAGE: false
         BUILD_TIMEOUT: 784
         Q_AGENT: openvswitch
-        Q_ML2_TENANT_NETWORK_TYPE: vxlan
+        Q_ML2_TENANT_NETWORK_TYPE: vxlan,vlan
         Q_ML2_PLUGIN_MECHANISM_DRIVERS: openvswitch
       devstack_local_conf:
         post-config:
           /$NEUTRON_CORE_PLUGIN_CONF:
             AGENT:
-              tunnel_types: vxlan,gre
+              tunnel_types: vxlan
+            ml2_type_vlan:
+              network_vlan_ranges: public
         test-config:
           $TEMPEST_CONFIG:
-            taas_plugin_options:
-              advanced_image_ref: ubuntu-20.04-minimal-cloudimg-amd64
-              advanced_image_ssh_user: ubuntu
+            neutron_plugin_options:
+              image_is_advanced: true
+              advanced_image_flavor_ref: d1
+            taas:
               provider_physical_network: public
               provider_segmentation_id: 100
             image_feature_enabled: