diff --git a/.zuul.yaml b/.zuul.yaml
index 278d8a5..e34be89 100644
--- a/.zuul.yaml
+++ b/.zuul.yaml
@@ -101,6 +101,21 @@
       devstack_localrc:
         TEMPEST_PLUGINS: /opt/stack/tap-as-a-service-tempest-plugin
         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/releases/bionic/release/ubuntu-18.04-server-cloudimg-amd64.img"
+        DEFAULT_IMAGE_NAME: cirros-0.3.4-i386-disk
+        ADVANCED_IMAGE_NAME: ubuntu-18.04-server-cloudimg-amd64
+        BUILD_TIMEOUT: 784
+      devstack_local_conf:
+        test-config:
+          $TEMPEST_CONFIG:
+            taas_plugin_options:
+              advanced_image_ref: ubuntu-18.04-server-cloudimg-amd64
+              advanced_image_ssh_user: ubuntu
+              provider_physical_network: public
+              provider_segmentation_id: 100
+            image_feature_enabled:
+              api_v2: true
       devstack_plugins:
         neutron: git://opendev.org/openstack/neutron.git
         tap-as-a-service: git://opendev.org/x/tap-as-a-service.git
diff --git a/neutron_taas_tempest_plugin/config.py b/neutron_taas_tempest_plugin/config.py
index d6e02a3..7d8785b 100644
--- a/neutron_taas_tempest_plugin/config.py
+++ b/neutron_taas_tempest_plugin/config.py
@@ -34,4 +34,14 @@ TaaSPluginGroup = [
                default='',
                help='Comma separated list of VLANs to be mirrored '
                     'for a Tap-Flow.'),
+    cfg.StrOpt('advanced_image_ref',
+               default='',
+               help='Valid advanced image uuid to be used in tests. '
+                    'Must contain tcpdump preinstalled.'),
+    cfg.StrOpt('advanced_image_ssh_user',
+               default='ubuntu',
+               help='Name of ssh user to use with advanced image in tests.'),
+    cfg.StrOpt('advanced_image_flavor_ref',
+               default='m1.medium',
+               help='Valid flavor to use with advanced image in tests.'),
 ]
diff --git a/neutron_taas_tempest_plugin/tests/scenario/manager.py b/neutron_taas_tempest_plugin/tests/scenario/manager.py
index e3d857b..edd78a1 100644
--- a/neutron_taas_tempest_plugin/tests/scenario/manager.py
+++ b/neutron_taas_tempest_plugin/tests/scenario/manager.py
@@ -38,7 +38,7 @@ LOG = log.getLogger(__name__)
 class ScenarioTest(tempest.test.BaseTestCase):
     """Base class for scenario tests. Uses tempest own clients. """
 
-    credentials = ['primary']
+    credentials = ['primary', 'admin']
 
     @classmethod
     def setup_clients(cls):
@@ -51,10 +51,14 @@ class ScenarioTest(tempest.test.BaseTestCase):
         cls.ports_client = cls.os_primary.ports_client
         cls.routers_client = cls.os_primary.routers_client
         cls.subnets_client = cls.os_primary.subnets_client
+        cls.flavor_client = cls.os_primary.flavors_client
         cls.floating_ips_client = cls.os_primary.floating_ips_client
         cls.security_groups_client = cls.os_primary.security_groups_client
         cls.security_group_rules_client = (
             cls.os_primary.security_group_rules_client)
+        cls.admin_networks_client = cls.os_admin.networks_client
+        cls.admin_subnets_client = cls.os_admin.subnets_client
+        cls.admin_routers_client = cls.os_admin.routers_client
 
     # ## Test functions library
     #
@@ -76,13 +80,14 @@ class ScenarioTest(tempest.test.BaseTestCase):
                         client.delete_port, port['id'])
         return port
 
-    def create_keypair(self, client=None):
+    @classmethod
+    def create_keypair(cls, client=None):
         if not client:
-            client = self.keypairs_client
-        name = data_utils.rand_name(self.__class__.__name__)
+            client = cls.keypairs_client
+        name = data_utils.rand_name(cls.__class__.__name__)
         # We don't need to create a keypair by pubkey in scenario
         body = client.create_keypair(name=name)
-        self.addCleanup(client.delete_keypair, name)
+        cls.addClassResourceCleanup(client.delete_keypair, name)
         return body['keypair']
 
     def create_server(self, name=None, image_id=None, flavor=None,
@@ -353,12 +358,13 @@ class NetworkScenarioTest(ScenarioTest):
         if not CONF.service_available.neutron:
             raise cls.skipException('Neutron not available')
 
-    def _create_network(self, networks_client=None,
+    @classmethod
+    def _create_network(cls, networks_client=None,
                         tenant_id=None,
                         namestart='network-smoke-',
                         port_security_enabled=True):
         if not networks_client:
-            networks_client = self.networks_client
+            networks_client = cls.networks_client
         if not tenant_id:
             tenant_id = networks_client.tenant_id
         name = data_utils.rand_name(namestart)
@@ -370,13 +376,14 @@ class NetworkScenarioTest(ScenarioTest):
         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'])
+        assert network['name'] == name
+        cls.addClassResourceCleanup(test_utils.call_and_ignore_notfound_exc,
+                                    networks_client.delete_network,
+                                    network['id'])
         return network
 
-    def _create_subnet(self, network, subnets_client=None,
+    @classmethod
+    def _create_subnet(cls, network, subnets_client=None,
                        routers_client=None, namestart='subnet-smoke',
                        **kwargs):
         """Create a subnet for the given network
@@ -384,9 +391,9 @@ class NetworkScenarioTest(ScenarioTest):
         within the cidr block configured for tenant networks.
         """
         if not subnets_client:
-            subnets_client = self.subnets_client
+            subnets_client = cls.subnets_client
         if not routers_client:
-            routers_client = self.routers_client
+            routers_client = cls.routers_client
 
         def cidr_in_use(cidr, tenant_id):
             """Check cidr existence
@@ -394,7 +401,7 @@ class NetworkScenarioTest(ScenarioTest):
             :returns: True if subnet with cidr already exist in tenant
                   False else
             """
-            cidr_in_use = self.os_admin.subnets_client.list_subnets(
+            cidr_in_use = cls.os_admin.subnets_client.list_subnets(
                 tenant_id=tenant_id, cidr=cidr)['subnets']
             return len(cidr_in_use) != 0
 
@@ -432,13 +439,14 @@ class NetworkScenarioTest(ScenarioTest):
                 is_overlapping_cidr = 'overlaps with another subnet' in str(e)
                 if not is_overlapping_cidr:
                     raise
-        self.assertIsNotNone(result, 'Unable to allocate tenant network')
+        assert result is not None, 'Unable to allocate tenant network'
 
         subnet = result['subnet']
-        self.assertEqual(subnet['cidr'], str_cidr)
+        assert subnet['cidr'] == str_cidr
 
-        self.addCleanup(test_utils.call_and_ignore_notfound_exc,
-                        subnets_client.delete_subnet, subnet['id'])
+        cls.addClassResourceCleanup(
+            test_utils.call_and_ignore_notfound_exc,
+            subnets_client.delete_subnet, subnet['id'])
 
         return subnet
 
@@ -593,31 +601,33 @@ class NetworkScenarioTest(ScenarioTest):
                                           CONF.validation.ping_timeout,
                                           1)
 
-    def _create_security_group(self, security_group_rules_client=None,
+    @classmethod
+    def _create_security_group(cls, security_group_rules_client=None,
                                tenant_id=None,
                                namestart='secgroup-smoke',
                                security_groups_client=None):
         if security_group_rules_client is None:
-            security_group_rules_client = self.security_group_rules_client
+            security_group_rules_client = cls.security_group_rules_client
         if security_groups_client is None:
-            security_groups_client = self.security_groups_client
+            security_groups_client = cls.security_groups_client
         if tenant_id is None:
             tenant_id = security_groups_client.tenant_id
-        secgroup = self._create_empty_security_group(
+        secgroup = cls._create_empty_security_group(
             namestart=namestart, client=security_groups_client,
             tenant_id=tenant_id)
 
         # Add rules to the security group
-        rules = self._create_loginable_secgroup_rule(
+        rules = cls._create_loginable_secgroup_rule(
             security_group_rules_client=security_group_rules_client,
             secgroup=secgroup,
             security_groups_client=security_groups_client)
         for rule in rules:
-            self.assertEqual(tenant_id, rule['tenant_id'])
-            self.assertEqual(secgroup['id'], rule['security_group_id'])
+            assert tenant_id == rule['tenant_id']
+            assert secgroup['id'] == rule['security_group_id']
         return secgroup
 
-    def _create_empty_security_group(self, client=None, tenant_id=None,
+    @classmethod
+    def _create_empty_security_group(cls, client=None, tenant_id=None,
                                      namestart='secgroup-smoke'):
         """Create a security group without rules.
 
@@ -629,7 +639,7 @@ class NetworkScenarioTest(ScenarioTest):
         :returns: the created security group
         """
         if client is None:
-            client = self.security_groups_client
+            client = cls.security_groups_client
         if not tenant_id:
             tenant_id = client.tenant_id
         sg_name = data_utils.rand_name(namestart)
@@ -640,21 +650,23 @@ class NetworkScenarioTest(ScenarioTest):
         result = client.create_security_group(**sg_dict)
 
         secgroup = result['security_group']
-        self.assertEqual(secgroup['name'], sg_name)
-        self.assertEqual(tenant_id, secgroup['tenant_id'])
-        self.assertEqual(secgroup['description'], sg_desc)
+        assert secgroup['name'] == sg_name
+        assert tenant_id == secgroup['tenant_id']
+        assert secgroup['description'] == sg_desc
 
-        self.addCleanup(test_utils.call_and_ignore_notfound_exc,
-                        client.delete_security_group, secgroup['id'])
+        cls.addClassResourceCleanup(
+            test_utils.call_and_ignore_notfound_exc,
+            client.delete_security_group, secgroup['id'])
         return secgroup
 
-    def _default_security_group(self, client=None, tenant_id=None):
+    @classmethod
+    def _default_security_group(cls, client=None, tenant_id=None):
         """Get default secgroup for given tenant_id.
 
         :returns: default secgroup for given tenant
         """
         if client is None:
-            client = self.security_groups_client
+            client = cls.security_groups_client
         if not tenant_id:
             tenant_id = client.tenant_id
         sgs = [
@@ -662,10 +674,11 @@ class NetworkScenarioTest(ScenarioTest):
             if sg['tenant_id'] == tenant_id and sg['name'] == 'default'
         ]
         msg = "No default security group for tenant %s." % (tenant_id)
-        self.assertGreater(len(sgs), 0, msg)
+        assert len(sgs) > 0, msg
         return sgs[0]
 
-    def _create_security_group_rule(self, secgroup=None,
+    @classmethod
+    def _create_security_group_rule(cls, secgroup=None,
                                     sec_group_rules_client=None,
                                     tenant_id=None,
                                     security_groups_client=None, **kwargs):
@@ -687,13 +700,13 @@ class NetworkScenarioTest(ScenarioTest):
                     }
         """
         if sec_group_rules_client is None:
-            sec_group_rules_client = self.security_group_rules_client
+            sec_group_rules_client = cls.security_group_rules_client
         if security_groups_client is None:
-            security_groups_client = self.security_groups_client
+            security_groups_client = cls.security_groups_client
         if not tenant_id:
             tenant_id = security_groups_client.tenant_id
         if secgroup is None:
-            secgroup = self._default_security_group(
+            secgroup = cls._default_security_group(
                 client=security_groups_client, tenant_id=tenant_id)
 
         ruleset = dict(security_group_id=secgroup['id'],
@@ -703,12 +716,13 @@ class NetworkScenarioTest(ScenarioTest):
         sg_rule = sec_group_rules_client.create_security_group_rule(**ruleset)
         sg_rule = sg_rule['security_group_rule']
 
-        self.assertEqual(secgroup['tenant_id'], sg_rule['tenant_id'])
-        self.assertEqual(secgroup['id'], sg_rule['security_group_id'])
+        assert secgroup['tenant_id'] == sg_rule['tenant_id']
+        assert secgroup['id'] == sg_rule['security_group_id']
 
         return sg_rule
 
-    def _create_loginable_secgroup_rule(self, security_group_rules_client=None,
+    @classmethod
+    def _create_loginable_secgroup_rule(cls, security_group_rules_client=None,
                                         secgroup=None,
                                         security_groups_client=None):
         """Create loginable security group rule
@@ -721,9 +735,9 @@ class NetworkScenarioTest(ScenarioTest):
         """
 
         if security_group_rules_client is None:
-            security_group_rules_client = self.security_group_rules_client
+            security_group_rules_client = cls.security_group_rules_client
         if security_groups_client is None:
-            security_groups_client = self.security_groups_client
+            security_groups_client = cls.security_groups_client
         rules = []
         rulesets = [
             dict(
@@ -747,7 +761,7 @@ class NetworkScenarioTest(ScenarioTest):
             for r_direction in ['ingress', 'egress']:
                 ruleset['direction'] = r_direction
                 try:
-                    sg_rule = self._create_security_group_rule(
+                    sg_rule = cls._create_security_group_rule(
                         sec_group_rules_client=sec_group_rules_client,
                         secgroup=secgroup,
                         security_groups_client=security_groups_client,
@@ -758,12 +772,13 @@ class NetworkScenarioTest(ScenarioTest):
                     if msg not in ex._error_string:
                         raise ex
                 else:
-                    self.assertEqual(r_direction, sg_rule['direction'])
+                    assert r_direction == sg_rule['direction']
                     rules.append(sg_rule)
 
         return rules
 
-    def _get_router(self, client=None, tenant_id=None):
+    @classmethod
+    def _get_router(cls, client=None, tenant_id=None):
         """Retrieve a router for the given tenant id.
 
         If a public router has been configured, it will be returned.
@@ -773,7 +788,7 @@ class NetworkScenarioTest(ScenarioTest):
         routes traffic to the public network.
         """
         if not client:
-            client = self.routers_client
+            client = cls.routers_client
         if not tenant_id:
             tenant_id = client.tenant_id
         router_id = CONF.network.public_router_id
@@ -782,7 +797,7 @@ class NetworkScenarioTest(ScenarioTest):
             body = client.show_router(router_id)
             return body['router']
         elif network_id:
-            router = self._create_router(client, tenant_id)
+            router = cls._create_router(client, tenant_id)
             kwargs = {'external_gateway_info': dict(network_id=network_id)}
             router = client.update_router(router['id'], **kwargs)['router']
             return router
@@ -790,10 +805,11 @@ class NetworkScenarioTest(ScenarioTest):
             raise Exception("Neither of 'public_router_id' or "
                             "'public_network_id' has been defined.")
 
-    def _create_router(self, client=None, tenant_id=None,
+    @classmethod
+    def _create_router(cls, client=None, tenant_id=None,
                        namestart='router-smoke'):
         if not client:
-            client = self.routers_client
+            client = cls.routers_client
         if not tenant_id:
             tenant_id = client.tenant_id
         name = data_utils.rand_name(namestart)
@@ -801,10 +817,10 @@ class NetworkScenarioTest(ScenarioTest):
                                       admin_state_up=True,
                                       tenant_id=tenant_id)
         router = result['router']
-        self.assertEqual(router['name'], name)
-        self.addCleanup(test_utils.call_and_ignore_notfound_exc,
-                        client.delete_router,
-                        router['id'])
+        assert router['name'] == name
+        cls.addClassResourceCleanup(test_utils.call_and_ignore_notfound_exc,
+                                    client.delete_router,
+                                    router['id'])
         return router
 
     def _update_router_admin_state(self, router, admin_state_up):
@@ -813,7 +829,8 @@ class NetworkScenarioTest(ScenarioTest):
             router['id'], **kwargs)['router']
         self.assertEqual(admin_state_up, router['admin_state_up'])
 
-    def create_networks(self, networks_client=None,
+    @classmethod
+    def create_networks(cls, networks_client=None,
                         routers_client=None, subnets_client=None,
                         tenant_id=None, dns_nameservers=None,
                         port_security_enabled=True):
@@ -835,33 +852,149 @@ class NetworkScenarioTest(ScenarioTest):
             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(
+            network = cls._get_network_by_name(
                 CONF.compute.fixed_network_name)
             router = None
             subnet = None
         else:
-            network = self._create_network(
+            network = cls._create_network(
                 networks_client=networks_client,
                 tenant_id=tenant_id,
                 port_security_enabled=port_security_enabled)
-            router = self._get_router(client=routers_client,
-                                      tenant_id=tenant_id)
+            router = cls._get_router(client=routers_client,
+                                     tenant_id=tenant_id)
             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)
+            subnet = cls._create_subnet(**subnet_kwargs)
             if not routers_client:
-                routers_client = self.routers_client
+                routers_client = cls.routers_client
             router_id = router['id']
             routers_client.add_router_interface(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, router_id,
-                            subnet_id=subnet['id'])
+            cls.addClassResourceCleanup(
+                test_utils.call_and_ignore_notfound_exc,
+                routers_client.remove_router_interface, router_id,
+                subnet_id=subnet['id'])
         return network, subnet, router
+
+    def _create_server_and_wait(self, source_port, image_id=None,
+                                flavor_id=None):
+        image_id = image_id if image_id else CONF.compute.image_ref
+        flavor_id = flavor_id if flavor_id else CONF.compute.flavor_ref
+
+        vm = self.create_server(
+            flavor=flavor_id,
+            image_id=image_id,
+            key_name=self.keypair['name'],
+            wait_until='ACTIVE',
+            networks=[{'port': source_port['id']}])
+
+        return vm
+
+    def _create_server_with_floatingip(self, use_taas_cloud_image=False,
+                                       provider_net=False, **kwargs):
+        net_id = self.network['id']
+        flavor = CONF.compute.flavor_ref
+        image = CONF.compute.image_ref
+        if use_taas_cloud_image:
+            image = self._get_real_image_id(
+                CONF.taas_plugin_options.advanced_image_ref)
+            flavor = self._get_real_flavor_id(
+                CONF.taas_plugin_options.advanced_image_flavor_ref)
+
+        if provider_net:
+            net_id = self.provider_network['id']
+
+        port = self._create_port(
+            net_id, security_groups=[self.secgroup['id']], **kwargs)
+        self._create_server_and_wait(port, image, flavor)
+
+        fip = self.create_floating_ip(
+            port, port_id=port['id'],
+            external_network_id=CONF.network.public_network_id)
+
+        return port, fip
+
+    def _run_in_background(self, sshclient, cmd):
+        runInBg = "nohup %s 2>&1 &" % cmd
+        sshclient.exec_command(runInBg)
+
+    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_plugin_options.provider_physical_network,
+        }
+
+        segmentation_id = CONF.taas_plugin_options.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_networks_client.create_network(
+            **network_kwargs)['network']
+        self.addCleanup(test_utils.call_and_ignore_notfound_exc,
+                        self.admin_networks_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_subnets_client.create_subnet(**subnet)
+        self.addCleanup(
+            test_utils.call_and_ignore_notfound_exc,
+            self.admin_subnets_client.delete_subnet, result['subnet']['id'])
+
+        self.admin_routers_client.add_router_interface(
+            self.router['id'], subnet_id=result['subnet']['id'])
+
+        self.addCleanup(
+            test_utils.call_and_ignore_notfound_exc,
+            self.admin_routers_client.remove_router_interface,
+            self.router['id'], subnet_id=result['subnet']['id'])
+
+    def _get_real_image_id(self, image_id):
+        try:
+            id = self.image_client.show_image(
+                image_id=image_id)["id"]
+        except AssertionError:
+            images = self.image_client.list_images(
+                params={"name": image_id})["images"]
+            if len(images) == 0:
+                self.skip_checks()
+            id = images[0]["id"]
+        return id
+
+    def _get_real_flavor_id(self, flavor_id):
+        id = None
+        try:
+            id = self.flavor_client.show_flavor(
+                flavor_id=flavor_id)["flavor"]["id"]
+        except AssertionError:
+            flavors = self.flavor_client.list_flavors()["flavors"]
+            for f in flavors:
+                if f["name"] == flavor_id:
+                    id = f["id"]
+        if id is None:
+            self.skip_checks()
+        return id
diff --git a/neutron_taas_tempest_plugin/tests/scenario/test_traffic_impact.py b/neutron_taas_tempest_plugin/tests/scenario/test_traffic_impact.py
new file mode 100644
index 0000000..12dd273
--- /dev/null
+++ b/neutron_taas_tempest_plugin/tests/scenario/test_traffic_impact.py
@@ -0,0 +1,245 @@
+# 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.linux import remote_client
+from tempest.lib import decorators
+
+from neutron_taas_tempest_plugin.tests.scenario import base
+
+
+CONF = config.CONF
+LOG = log.getLogger(__name__)
+
+
+class TestTaaSTrafficScenarios(base.TaaSScenarioTest):
+
+    @classmethod
+    def setup_clients(cls):
+        super(TestTaaSTrafficScenarios, cls).setup_clients()
+
+        if CONF.image_feature_enabled.api_v1:
+            cls.image_client = cls.os_primary.image_client
+        elif CONF.image_feature_enabled.api_v2:
+            cls.image_client = cls.os_primary.image_client_v2
+        else:
+            raise cls.skipException(
+                'Either api_v1 or api_v2 must be True in '
+                '[image-feature-enabled].')
+
+    @classmethod
+    @utils.requires_ext(extension="router", service="network")
+    def resource_setup(cls):
+        super(TestTaaSTrafficScenarios, cls).resource_setup()
+        for ext in ['taas']:
+            if not utils.is_extension_enabled(ext, 'network'):
+                msg = "%s Extension not enabled." % ext
+                raise cls.skipException(msg)
+
+        cls.network, cls.subnet, cls.router = cls.create_networks()
+        cls.provider_network = None
+        cls.keypair = cls.create_keypair()
+        cls.secgroup = cls._create_security_group()
+
+    @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 |
+           +------------+
+       """
+        if provider_net:
+            if CONF.taas_plugin_options.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)
+        self.left_port, left_fip = self._create_server_with_floatingip(
+            provider_net=provider_net)
+        self.right_port, right_fip = self._create_server_with_floatingip(
+            provider_net=provider_net)
+
+        if taas:
+            LOG.debug("Create TAAS service")
+            tap_service = self.create_tap_service(port_id=self.mon_port['id'])
+            self.create_tap_flow(tap_service_id=tap_service['id'],
+                                 direction='BOTH',
+                                 source_port=self.left_port['id'])
+            self.create_tap_flow(tap_service_id=tap_service['id'],
+                                 direction='BOTH',
+                                 source_port=self.right_port['id'])
+
+        user = CONF.validation.image_ssh_user
+        if use_taas_cloud_image:
+            user = CONF.taas_plugin_options.advanced_image_ssh_user
+
+        self.monitor_client = remote_client.RemoteClient(
+            mon_fip['floating_ip_address'], user,
+            pkey=self.keypair['private_key'])
+        self.left_client = remote_client.RemoteClient(
+            left_fip['floating_ip_address'], CONF.validation.image_ssh_user,
+            pkey=self.keypair['private_key'])
+        self.right_client = remote_client.RemoteClient(
+            right_fip['floating_ip_address'], CONF.validation.image_ssh_user,
+            pkey=self.keypair['private_key'])
+        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
+        self.left_client.exec_command("ping -c 50 %s" % right_ip)
+
+        # Collect tcpdump results
+        output = self.monitor_client.exec_command("cat %s" % log_location)
+        self.assertTrue(len(output) > 0)
+
+        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.taas_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.taas_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.taas_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.taas_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())