new_contents: from dtest import Tester, debug
  from tools.assertions import assert_all
  from tools.decorators import since
  
  
  class CompatibilityFlagTest(Tester):
      """
      Test 30 protocol compatibility flag
  
      @jira CASSANDRA-13004
      """
  
      def _compatibility_flag_off_with_30_node_test(self, from_version):
          """
          Test compatibility with 30 protocol version: if the flag is unset, schema agreement can not be reached
          """
  
          cluster = self.cluster
          cluster.populate(2)
          node1, node2 = cluster.nodelist()
          cluster.set_install_dir(version=from_version)
          cluster.start(wait_for_binary_proto=True)
  
          node1.drain()
          node1.watch_log_for("DRAINED")
          node1.stop(wait_other_notice=False)
          debug("Upgrading to current version")
          self.set_node_to_current_version(node1)
          node1.start(wait_for_binary_proto=True)
  
          node1.watch_log_for("Not pulling schema because versions match or shouldPullSchemaFrom returned false", filename='debug.log')
          node2.watch_log_for("Not pulling schema because versions match or shouldPullSchemaFrom returned false", filename='debug.log')
  
      def _compatibility_flag_on_with_30_test(self, from_version):
          """
          Test compatibility with 30 protocol version: if the flag is set, schema agreement can be reached
          """
  
          cluster = self.cluster
          cluster.populate(2)
          node1, node2 = cluster.nodelist()
          cluster.set_install_dir(version=from_version)
          cluster.start(wait_for_binary_proto=True)
  
          node1.drain()
          node1.watch_log_for("DRAINED")
          node1.stop(wait_other_notice=False)
          debug("Upgrading to current version")
          self.set_node_to_current_version(node1)
          node1.start(jvm_args=["-Dcassandra.force_3_0_protocol_version=true"], wait_for_binary_proto=True)
  
          session = self.patient_cql_connection(node1)
          self._run_test(session)
  
      def _compatibility_flag_on_3014_test(self):
          """
          Test compatibility between post-13004 nodes, one of which is in compatibility mode
          """
  
          cluster = self.cluster
          cluster.populate(2)
          node1, node2 = cluster.nodelist()
  
          node1.start(wait_for_binary_proto=True)
          node2.start(jvm_args=["-Dcassandra.force_3_0_protocol_version=true"], wait_for_binary_proto=True)
  
          session = self.patient_cql_connection(node1)
          self._run_test(session)
  
      def _compatibility_flag_off_3014_test(self):
          """
          Test compatibility between post-13004 nodes
          """
  
          cluster = self.cluster
          cluster.populate(2)
          node1, node2 = cluster.nodelist()
  
          node1.start(wait_for_binary_proto=True)
          node2.start(wait_for_binary_proto=True)
  
          session = self.patient_cql_connection(node1)
          self._run_test(session)
  
      def _run_test(self, session):
          # Make sure the system_auth table will get replicated to the node that we're going to replace
  
          session.execute("CREATE KEYSPACE test WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '2'} ;")
          session.execute("CREATE TABLE test.test (a text PRIMARY KEY, b text, c text);")
  
          for i in range(1, 6):
              session.execute("INSERT INTO test.test (a, b, c) VALUES ('{}', '{}', '{}');".format(i, i + 1, i + 2))
  
          assert_all(session,
                     "SELECT * FROM test.test",
                     [[str(i), str(i + 1), str(i + 2)] for i in range(1, 6)], ignore_order=True)
  
          assert_all(session,
                     "SELECT a,c FROM test.test",
                     [[str(i), str(i + 2)] for i in range(1, 6)], ignore_order=True)
  
  
  @since('3.0.14', max_version='3.0.x')
  class CompatibilityFlag30XTest(CompatibilityFlagTest):
  
      def compatibility_flag_off_with_30_node_test(self):
          self._compatibility_flag_off_with_30_node_test('3.0.12')
  
      def compatibility_flag_on_with_3_0_test(self):
          self._compatibility_flag_on_with_30_test('3.0.12')
  
      def compatibility_flag_on_3014_test(self):
          self._compatibility_flag_on_3014_test()
  
      def compatibility_flag_off_3014_test(self):
          self._compatibility_flag_off_3014_test()
  
  
  @since('3.11', max_version='4')
  class CompatibilityFlag3XTest(CompatibilityFlagTest):
  
      def compatibility_flag_off_with_30_node_test(self):
          self._compatibility_flag_off_with_30_node_test('3.10')
  
      def compatibility_flag_on_with_3_0_test(self):
          self._compatibility_flag_on_with_30_test('3.10')
  
      def compatibility_flag_on_3014_test(self):
          self._compatibility_flag_on_3014_test()
  
      def compatibility_flag_off_3014_test(self):
          self._compatibility_flag_off_3014_test()
  
subject: Add dtests for compatibility flag introduced in CASSANDRA-13004 (#1485)