Added test for reattached security groups

We had a bug for OSP13 with openvswitch firewall driver that the established
connection can't be resumed after the security group has been removed from
the port and than added back. Need to test this behavior.

In order to keep the connection open there is a new StatefulConnection
class

Related-Bug: #1915530
Change-Id: I3c2f037180b35dbbd254d8b4ce69852d31391a9a

neutron_tempest_plugin/common/utils.py

@@ -136,3 +136,74 @@ def spawn_http_server(ssh_client, port, message):
 def call_url_remote(ssh_client, url):
     cmd = "curl %s --retry 3 --connect-timeout 2" % url
     return ssh_client.exec_command(cmd)
+
+
+class StatefulConnection:
+    """Class to test connection that should remain opened
+
+    Can be used to perform some actions while the initiated connection
+    remain opened
+    """
+
+    def __init__(self, client_ssh, server_ssh, target_ip, target_port):
+        self.client_ssh = client_ssh
+        self.server_ssh = server_ssh
+        self.ip = target_ip
+        self.port = target_port
+        self.connection_started = False
+        self.test_attempt = 0
+
+    def __enter__(self):
+        return self
+
+    @property
+    def test_str(self):
+        return 'attempt_{}'.format(str(self.test_attempt).zfill(3))
+
+    def _start_connection(self):
+        self.server_ssh.exec_command(
+                'echo "{}" > input.txt'.format(self.test_str))
+        self.server_ssh.exec_command('tail -f input.txt | nc -lp '
+                '{} &> output.txt &'.format(self.port))
+        self.client_ssh.exec_command(
+                'echo "{}" > input.txt'.format(self.test_str))
+        self.client_ssh.exec_command('tail -f input.txt | nc {} {} &>'
+                'output.txt &'.format(self.ip, self.port))
+
+    def _test_connection(self):
+        if not self.connection_started:
+            self._start_connection()
+        else:
+            self.server_ssh.exec_command(
+                    'echo "{}" >> input.txt'.format(self.test_str))
+            self.client_ssh.exec_command(
+                    'echo "{}" >> input.txt & sleep 1'.format(self.test_str))
+        try:
+            self.server_ssh.exec_command(
+                    'grep {} output.txt'.format(self.test_str))
+            self.client_ssh.exec_command(
+                    'grep {} output.txt'.format(self.test_str))
+            if not self.should_pass:
+                return False
+            else:
+                if not self.connection_started:
+                    self.connection_started = True
+                return True
+        except exceptions.SSHExecCommandFailed:
+            if self.should_pass:
+                return False
+            else:
+                return True
+        finally:
+            self.test_attempt += 1
+
+    def test_connection(self, should_pass=True, timeout=10, sleep_timer=1):
+        self.should_pass = should_pass
+        wait_until_true(
+                self._test_connection, timeout=timeout, sleep=sleep_timer)
+
+    def __exit__(self, type, value, traceback):
+        self.server_ssh.exec_command('sudo killall nc || killall nc')
+        self.server_ssh.exec_command('sudo killall tail || killall tail')
+        self.client_ssh.exec_command('sudo killall nc || killall nc')
+        self.client_ssh.exec_command('sudo killall tail || killall tail')

neutron_tempest_plugin/scenario/test_security_groups.py

@@ -277,6 +277,50 @@ class NetworkSecGroupTest(base.BaseTempestTestCase):
                       'remote_ip_prefix': cidr}]
         self._test_ip_prefix(rule_list, should_succeed=False)
 
+    @decorators.idempotent_id('01f0ddca-b049-47eb-befd-82acb502c9ec')
+    def test_established_tcp_session_after_re_attachinging_sg(self):
+        """Test existing connection remain open after sg has been re-attached
+
+        Verifies that new packets can pass over the existing connection when
+        the security group has been removed from the server and then added
+        back
+        """
+
+        ssh_sg = self.create_security_group()
+        self.create_loginable_secgroup_rule(secgroup_id=ssh_sg['id'])
+        vm_ssh, fips, vms = self.create_vm_testing_sec_grp(
+                security_groups=[{'name': ssh_sg['name']}])
+        sg = self.create_security_group()
+        nc_rule = [{'protocol': constants.PROTO_NUM_TCP,
+                    'direction': constants.INGRESS_DIRECTION,
+                    'port_range_min': 6666,
+                    'port_range_max': 6666}]
+        self.create_secgroup_rules(nc_rule, secgroup_id=sg['id'])
+        srv_port = self.client.list_ports(network_id=self.network['id'],
+                device_id=vms[1]['server']['id'])['ports'][0]
+        srv_ip = srv_port['fixed_ips'][0]['ip_address']
+        with utils.StatefulConnection(
+                vm_ssh[0], vm_ssh[1], srv_ip, 6666) as con:
+            self.client.update_port(srv_port['id'],
+                    security_groups=[ssh_sg['id'], sg['id']])
+            con.test_connection()
+        with utils.StatefulConnection(
+                vm_ssh[0], vm_ssh[1], srv_ip, 6666) as con:
+            self.client.update_port(
+                    srv_port['id'], security_groups=[ssh_sg['id']])
+            con.test_connection(should_pass=False)
+        with utils.StatefulConnection(
+                vm_ssh[0], vm_ssh[1], srv_ip, 6666) as con:
+            self.client.update_port(srv_port['id'],
+                    security_groups=[ssh_sg['id'], sg['id']])
+            con.test_connection()
+            self.client.update_port(srv_port['id'],
+                    security_groups=[ssh_sg['id']])
+            con.test_connection(should_pass=False)
+            self.client.update_port(srv_port['id'],
+                    security_groups=[ssh_sg['id'], sg['id']])
+            con.test_connection()
+
     @decorators.idempotent_id('7ed39b86-006d-40fb-887a-ae46693dabc9')
     def test_remote_group(self):
         # create a new sec group

zuul.d/master_jobs.yaml

@@ -206,7 +206,11 @@
       network_available_features: *available_features
       # TODO(slaweq): remove trunks subport_connectivity test from blacklist
       # when bug https://bugs.launchpad.net/neutron/+bug/1838760 will be fixed
-      tempest_exclude_regex: "(^neutron_tempest_plugin.scenario.test_trunk.TrunkTest.test_subport_connectivity)"
+      # TODO(akatz): remove established tcp session verification test when the
+      # bug https://bugzilla.redhat.com/show_bug.cgi?id=1965036 will be fixed
+      tempest_exclude_regex: "\
+          (^neutron_tempest_plugin.scenario.test_trunk.TrunkTest.test_subport_connectivity)|\
+          (^neutron_tempest_plugin.scenario.test_security_groups.NetworkSecGroupTest.test_established_tcp_session_after_re_attachinging_sg)"
       devstack_localrc:
         Q_AGENT: openvswitch
         Q_ML2_TENANT_NETWORK_TYPE: vxlan