def test_checksum_disabled(self):
    # Disable binlog_checksum to make sure we can also talk to a server without
    # checksums enabled, in case they are enabled by default.
    start_position = mysql_flavor().master_position(dst_replica)
    logging.debug('test_checksum_disabled: starting @ %s', start_position)

    # For flavors that don't support checksums, this is a no-op.
    mysql_flavor().disable_binlog_checksum(dst_replica)

    # Insert something and make sure it comes through intact.
    sql = (
        "INSERT INTO test_table (id, keyspace_id, msg) "
        "VALUES (58812, 1, 'testing checksum disabled') "
        "/* vtgate:: keyspace_id:00000001 */")
    src_master.mquery(
        'vt_test_keyspace', sql, write=True)

    # Look for it using update stream to see if binlog streamer can talk to
    # dst_replica, which now has binlog_checksum disabled.
    stream = _get_update_stream(dst_replica)
    found = False
    for stream_event in stream.stream_update(start_position):
      if stream_event.category == update_stream.StreamEvent.POS:
        break
      if stream_event.sql == sql:
        found = True
        break
    stream.close()
    self.assertEqual(found, True, 'expected query not found in update stream')

  def test_checksum_enabled(self):
    start_position = mysql_flavor().master_position(dst_replica)
    logging.debug('test_checksum_enabled: starting @ %s', start_position)

    # Enable binlog_checksum, which will also force a log rotation that should
    # cause binlog streamer to notice the new checksum setting.
    if not mysql_flavor().enable_binlog_checksum(dst_replica):
      logging.debug('skipping checksum test on flavor without binlog_checksum setting')
      return

    # Insert something and make sure it comes through intact.
    sql = "INSERT INTO test_table (id, keyspace_id, msg) VALUES (19283, 1, 'testing checksum enabled') /* EMD keyspace_id:1 */"
    src_master.mquery("vt_test_keyspace",
        sql,
        write=True)

    # Look for it using update stream to see if binlog streamer can talk to
    # dst_replica, which now has binlog_checksum enabled.
    stream = _get_update_stream(dst_replica)
    stream.dial()
    data = stream.stream_start(start_position)
    found = False
    while data:
      if data['Category'] == 'POS':
        break
      if data['Sql'] == sql:
        found = True
        break
      data = stream.stream_next()
    stream.close()
    self.assertEqual(found, True, 'expected query not found in update stream')

    def test_no_mysql_healthcheck(self):
        """This test starts a vttablet with no mysql port, while mysql is down.
    It makes sure vttablet will start properly and be unhealthy.
    Then we start mysql, and make sure vttablet becomes healthy.
    """
        # we need replication to be enabled, so the slave tablet can be healthy.
        for t in tablet_62344, tablet_62044:
            t.create_db("vt_test_keyspace")
        pos = mysql_flavor().master_position(tablet_62344)
        changeMasterCmds = mysql_flavor().change_master_commands(utils.hostname, tablet_62344.mysql_port, pos)
        tablet_62044.mquery("", ["RESET MASTER", "RESET SLAVE"] + changeMasterCmds + ["START SLAVE"])

        # now shutdown all mysqld
        shutdown_procs = [tablet_62344.shutdown_mysql(), tablet_62044.shutdown_mysql()]
        utils.wait_procs(shutdown_procs)

        # start the tablets, wait for them to be NOT_SERVING (mysqld not there)
        tablet_62344.init_tablet("master", "test_keyspace", "0")
        tablet_62044.init_tablet("spare", "test_keyspace", "0", include_mysql_port=False)
        for t in tablet_62344, tablet_62044:
            t.start_vttablet(
                wait_for_state=None, target_tablet_type="replica", full_mycnf_args=True, include_mysql_port=False
            )
        for t in tablet_62344, tablet_62044:
            t.wait_for_vttablet_state("NOT_SERVING")
            self.check_healthz(t, False)

        # restart mysqld
        start_procs = [tablet_62344.start_mysql(), tablet_62044.start_mysql()]
        utils.wait_procs(start_procs)

        # the master should still be healthy
        utils.run_vtctl(["RunHealthCheck", tablet_62344.tablet_alias, "replica"], auto_log=True)
        self.check_healthz(tablet_62344, True)

        # the slave won't be healthy at first, as replication is not running
        utils.run_vtctl(["RunHealthCheck", tablet_62044.tablet_alias, "replica"], auto_log=True)
        self.check_healthz(tablet_62044, False)
        tablet_62044.wait_for_vttablet_state("NOT_SERVING")

        # restart replication
        tablet_62044.mquery("", ["START SLAVE"])

        # wait for the tablet to become healthy and fix its mysql port
        utils.run_vtctl(["RunHealthCheck", tablet_62044.tablet_alias, "replica"], auto_log=True)
        tablet_62044.wait_for_vttablet_state("SERVING")
        self.check_healthz(tablet_62044, True)

        for t in tablet_62344, tablet_62044:
            # wait for mysql port to show up
            timeout = 10
            while True:
                ti = utils.run_vtctl_json(["GetTablet", t.tablet_alias])
                if "mysql" in ti["Portmap"]:
                    break
                timeout = utils.wait_step("mysql port in tablet record", timeout)
            self.assertEqual(ti["Portmap"]["mysql"], t.mysql_port)

        # all done
        tablet.kill_tablets([tablet_62344, tablet_62044])

  def test_checksum_enabled(self):
    start_position = mysql_flavor().master_position(dst_replica)
    logging.debug('test_checksum_enabled: starting @ %s', start_position)

    # Enable binlog_checksum, which will also force a log rotation that should
    # cause binlog streamer to notice the new checksum setting.
    if not mysql_flavor().enable_binlog_checksum(dst_replica):
      logging.debug(
          'skipping checksum test on flavor without binlog_checksum setting')
      return

    # Insert something and make sure it comes through intact.
    sql = (
        "INSERT INTO test_table (id, keyspace_id, msg) "
        "VALUES (19283, 1, 'testing checksum enabled') "
        "/* vtgate:: keyspace_id:00000001 */")
    src_master.mquery('vt_test_keyspace', sql, write=True)

    # Look for it using update stream to see if binlog streamer can talk to
    # dst_replica, which now has binlog_checksum enabled.
    stream = _get_update_stream(dst_replica)
    found = False
    for stream_event in stream.stream_update(start_position):
      if stream_event.category == update_stream.StreamEvent.POS:
        break
      if stream_event.sql == sql:
        found = True
        break
    stream.close()
    self.assertEqual(found, True, 'expected query not found in update stream')

  def test_checksum_disabled(self):
    # Disable binlog_checksum to make sure we can also talk to a server without
    # checksums enabled, in case they are enabled by default.
    start_position = mysql_flavor().master_position(dst_replica)
    logging.debug('test_checksum_disabled: starting @ %s', start_position)

    # For flavors that don't support checksums, this is a no-op.
    mysql_flavor().disable_binlog_checksum(dst_replica)

    # Insert something and make sure it comes through intact.
    sql = "INSERT INTO test_table (id, keyspace_id, msg) VALUES (58812, 1, 'testing checksum disabled') /* EMD keyspace_id:1 */"
    src_master.mquery("vt_test_keyspace",
        sql,
        write=True)

    # Look for it using update stream to see if binlog streamer can talk to
    # dst_replica, which now has binlog_checksum disabled.
    stream = _get_update_stream(dst_replica)
    stream.dial()
    data = stream.stream_start(start_position)
    found = False
    while data:
      if data['Category'] == 'POS':
        break
      if data['Sql'] == sql:
        found = True
        break
      data = stream.stream_next()
    stream.close()
    self.assertEqual(found, True, 'expected query not found in update stream')

 def init_mysql(self, extra_my_cnf=None):
   if self.use_mysqlctld:
     return self.mysqlctld(
         ['-bootstrap_archive', mysql_flavor().bootstrap_archive()],
         extra_my_cnf=extra_my_cnf)
   else:
     return self.mysqlctl(
         ['init', '-bootstrap_archive', mysql_flavor().bootstrap_archive()],
         extra_my_cnf=extra_my_cnf, with_ports=True)

  def test_no_mysql_healthcheck(self):
    """This test starts a vttablet with no mysql port, while mysql is down.
    It makes sure vttablet will start properly and be unhealthy.
    Then we start mysql, and make sure vttablet becomes healthy.
    """
    # we need replication to be enabled, so the slave tablet can be healthy.
    for t in tablet_62344, tablet_62044:
      t.create_db('vt_test_keyspace')
    pos = mysql_flavor().master_position(tablet_62344)
    changeMasterCmds = mysql_flavor().change_master_commands(
                            utils.hostname,
                            tablet_62344.mysql_port,
                            pos)
    tablet_62044.mquery('', ['RESET MASTER', 'RESET SLAVE'] +
                        changeMasterCmds +
                        ['START SLAVE'])

    # now shutdown all mysqld
    shutdown_procs = [
        tablet_62344.shutdown_mysql(),
        tablet_62044.shutdown_mysql(),
        ]
    utils.wait_procs(shutdown_procs)

    # start the tablets, wait for them to be NOT_SERVING (mysqld not there)
    tablet_62344.init_tablet('master', 'test_keyspace', '0')
    tablet_62044.init_tablet('spare', 'test_keyspace', '0',
                             include_mysql_port=False)
    for t in tablet_62344, tablet_62044:
      t.start_vttablet(wait_for_state=None,
                       target_tablet_type='replica',
                       full_mycnf_args=True, include_mysql_port=False)
    for t in tablet_62344, tablet_62044:
      t.wait_for_vttablet_state('NOT_SERVING')

    # restart mysqld
    start_procs = [
        tablet_62344.start_mysql(),
        tablet_62044.start_mysql(),
        ]
    utils.wait_procs(start_procs)

    # wait for the tablets to become healthy and fix their mysql port
    for t in tablet_62344, tablet_62044:
      t.wait_for_vttablet_state('SERVING')
    for t in tablet_62344, tablet_62044:
      # wait for mysql port to show up
      timeout = 10
      while True:
        ti = utils.run_vtctl_json(['GetTablet', t.tablet_alias])
        if 'mysql' in ti['Portmap']:
          break
        timeout = utils.wait_step('mysql port in tablet record', timeout)
      self.assertEqual(ti['Portmap']['mysql'], t.mysql_port)

    # all done
    tablet.kill_tablets([tablet_62344, tablet_62044])

  def test_charset(self):
    start_position = mysql_flavor().master_position(dst_replica)
    logging.debug('test_charset: starting @ %s', start_position)

    # Insert something that will replicate incorrectly if the charset is not
    # propagated through binlog streamer to the destination.
    #
    # Vitess tablets default to using utf8, so we insert something crazy and
    # pretend it's latin1. If the binlog player doesn't also pretend it's
    # latin1, it will be inserted as utf8, which will change its value.
    src_master.mquery("vt_test_keyspace",
        "INSERT INTO test_table (id, keyspace_id, msg) VALUES (41523, 1, 'Šṛ́rỏé') /* EMD keyspace_id:1 */",
        conn_params={'charset': 'latin1'}, write=True)

    # Wait for it to replicate.
    stream = _get_update_stream(dst_replica)
    stream.dial()
    data = stream.stream_start(start_position)
    while data:
      if data['Category'] == 'POS':
        break
      data = stream.stream_next()
    stream.close()

    # Check the value.
    data = dst_master.mquery("vt_test_keyspace",
        "SELECT id, keyspace_id, msg FROM test_table WHERE id=41523 LIMIT 1")
    self.assertEqual(len(data), 1, 'No data replicated.')
    self.assertEqual(len(data[0]), 3, 'Wrong number of columns.')
    self.assertEqual(data[0][2], 'Šṛ́rỏé', 'Data corrupted due to wrong charset.')

 def test_log_rotation(self):
   start_position = _get_master_current_position()
   logging.debug('test_log_rotation: starting @ %s', start_position)
   position = start_position
   master_tablet.mquery('vt_test_keyspace', 'flush logs')
   self._exec_vt_txn(self._populate_vt_a(15))
   self._exec_vt_txn(['delete from vt_a'])
   master_conn = self._get_vtgate_stream_conn()
   master_txn_count = 0
   logs_correct = False
   for event, _ in master_conn.update_stream(
       'test_keyspace', topodata_pb2.MASTER,
       event=query_pb2.EventToken(shard='0', position=start_position),
       shard='0'):
     if event.event_token.position:
       master_txn_count += 1
       position = event.event_token.position
       if mysql_flavor().position_after(position, start_position):
         logs_correct = True
         logging.debug('Log rotation correctly interpreted')
         break
       if master_txn_count == 2:
         self.fail('ran out of logs')
   if not logs_correct:
     self.fail("Flush logs didn't get properly interpreted")
   master_conn.close()

  def test_charset(self):
    start_position = mysql_flavor().master_position(dst_replica)
    logging.debug('test_charset: starting @ %s', start_position)

    # Insert something that will replicate incorrectly if the charset is not
    # propagated through binlog streamer to the destination.
    #
    # Vitess tablets default to using utf8, so we insert something crazy and
    # pretend it's latin1. If the binlog player doesn't also pretend it's
    # latin1, it will be inserted as utf8, which will change its value.
    src_master.mquery(
        'vt_test_keyspace',
        "INSERT INTO test_table (id, keyspace_id, msg) "
        "VALUES (41523, 1, 'Šṛ́rỏé') /* vtgate:: keyspace_id:00000001 */",
        conn_params={'charset': 'latin1'}, write=True)

    # Wait for it to replicate.
    stream = _get_update_stream(dst_replica)
    for event in stream.stream_update('test_keyspace', '-',
                                      topodata_pb2.REPLICA, start_position):
      if event.event_token.position:
        break
    stream.close()

    # Check the value.
    data = dst_master.mquery(
        'vt_test_keyspace',
        'SELECT id, keyspace_id, msg FROM test_table WHERE id=41523 LIMIT 1')
    self.assertEqual(len(data), 1, 'No data replicated.')
    self.assertEqual(len(data[0]), 3, 'Wrong number of columns.')
    self.assertEqual(data[0][2], 'Šṛ́rỏé',
                     'Data corrupted due to wrong charset.')

def wait_for_replication_pos(tablet_a, tablet_b, timeout=60.0):
  """Waits for tablet B to catch up to the replication position of tablet A.

  If the replication position does not catch up within timeout seconds, it will
  raise a TestError.
  """
  replication_pos_a = mysql_flavor().master_position(tablet_a)
  while True:
    replication_pos_b = mysql_flavor().master_position(tablet_b)
    if mysql_flavor().position_at_least(replication_pos_b, replication_pos_a):
      break
    timeout = wait_step(
      "%s's replication position to catch up %s's; currently at: %s, waiting to catch up to: %s" % (
        tablet_b.tablet_alias, tablet_a.tablet_alias, replication_pos_b, replication_pos_a),
      timeout, sleep_time=0.1
    )

  def test_charset(self):
    start_position = mysql_flavor().master_position(dst_replica)
    logging.debug('test_charset: starting @ %s', start_position)

    # Insert something that will replicate incorrectly if the charset is not
    # propagated through binlog streamer to the destination.
    #
    # Vitess tablets default to using utf8, so we insert something crazy and
    # pretend it's latin1. If the binlog player doesn't also pretend it's
    # latin1, it will be inserted as utf8, which will change its value.
    src_master.mquery(
        'vt_test_keyspace',
        "INSERT INTO test_table (id, keyspace_id, msg) "
        "VALUES (41523, 1, 'Šṛ́rỏé') /* vtgate:: keyspace_id:00000001 */",
        conn_params={'charset': 'latin1'}, write=True)

    # Wait for it to replicate.
    event = utils.run_vtctl_json(['VtTabletUpdateStream',
                                  '-position', start_position,
                                  '-count', '1',
                                  dst_replica.tablet_alias])
    self.assertIn('event_token', event)
    self.assertIn('timestamp', event['event_token'])

    # Check the value.
    data = dst_master.mquery(
        'vt_test_keyspace',
        'SELECT id, keyspace_id, msg FROM test_table WHERE id=41523 LIMIT 1')
    self.assertEqual(len(data), 1, 'No data replicated.')
    self.assertEqual(len(data[0]), 3, 'Wrong number of columns.')
    self.assertEqual(data[0][2], 'Šṛ́rỏé',
                     'Data corrupted due to wrong charset.')

def get_all_extra_my_cnf(extra_my_cnf):
  all_extra_my_cnf = []
  flavor_my_cnf = mysql_flavor().extra_my_cnf()
  if flavor_my_cnf:
    all_extra_my_cnf.append(flavor_my_cnf)
  if extra_my_cnf:
    all_extra_my_cnf.append(extra_my_cnf)
  return all_extra_my_cnf

def get_all_extra_my_cnf(extra_my_cnf):
  all_extra_my_cnf = [environment.vttop + '/config/mycnf/default-fast.cnf']
  flavor_my_cnf = mysql_flavor().extra_my_cnf()
  if flavor_my_cnf:
    all_extra_my_cnf.append(flavor_my_cnf)
  if extra_my_cnf:
    all_extra_my_cnf.append(extra_my_cnf)
  return all_extra_my_cnf

  def test_checksum_disabled(self):
    # Disable binlog_checksum to make sure we can also talk to a server without
    # checksums enabled, in case they are enabled by default.
    start_position = mysql_flavor().master_position(dst_replica)
    logging.debug('test_checksum_disabled: starting @ %s', start_position)

    # For flavors that don't support checksums, this is a no-op.
    mysql_flavor().disable_binlog_checksum(dst_replica)

    # Insert something and make sure it comes through intact.
    sql = (
        "INSERT INTO test_table (id, keyspace_id, msg) "
        "VALUES (58812, 1, 'testing checksum disabled') "
        "/* vtgate:: keyspace_id:00000001 */")
    src_master.mquery(
        'vt_test_keyspace', sql, write=True)

    # Look for it using update stream to see if binlog streamer can talk to
    # dst_replica, which now has binlog_checksum disabled.
    self._wait_for_replica_event(start_position, sql)

  def external_reparent(self):
    # Demote master.
    start = time.time()
    master.mquery('', mysql_flavor().demote_master_commands(), log_query=True)
    if master.semi_sync_enabled():
      master.set_semi_sync_enabled(master=False)

    # Wait for replica to catch up to master.
    utils.wait_for_replication_pos(master, replica)

    # Wait for at least one second to articially prolong the failover and give
    # the buffer a chance to observe it.
    d = time.time() - start
    min_unavailability_s = 1
    if d < min_unavailability_s:
      w = min_unavailability_s - d
      logging.debug('Waiting for %.1f seconds because the failover was too fast'
                    ' (took only %.3f seconds)', w, d)
      time.sleep(w)

    # Promote replica to new master.
    replica.mquery('', mysql_flavor().promote_slave_commands(),
                   log_query=True)
    if replica.semi_sync_enabled():
      replica.set_semi_sync_enabled(master=True)
    old_master = master
    new_master = replica

    # Configure old master to use new master.
    new_pos = mysql_flavor().master_position(new_master)
    logging.debug('New master position: %s', str(new_pos))
    # Use 'localhost' as hostname because Travis CI worker hostnames
    # are too long for MySQL replication.
    change_master_cmds = mysql_flavor().change_master_commands(
        'localhost', new_master.mysql_port, new_pos)
    old_master.mquery('', ['RESET SLAVE'] + change_master_cmds +
                      ['START SLAVE'], log_query=True)

    # Notify the new vttablet master about the reparent.
    utils.run_vtctl(['TabletExternallyReparented', new_master.tablet_alias],
                    auto_log=True)

  def test_checksum_enabled(self):
    start_position = mysql_flavor().master_position(dst_replica)
    logging.debug('test_checksum_enabled: starting @ %s', start_position)

    # Enable binlog_checksum, which will also force a log rotation that should
    # cause binlog streamer to notice the new checksum setting.
    if not mysql_flavor().enable_binlog_checksum(dst_replica):
      logging.debug(
          'skipping checksum test on flavor without binlog_checksum setting')
      return

    # Insert something and make sure it comes through intact.
    sql = (
        "INSERT INTO test_table (id, keyspace_id, msg) "
        "VALUES (19283, 1, 'testing checksum enabled') "
        "/* vtgate:: keyspace_id:00000001 */")
    src_master.mquery('vt_test_keyspace', sql, write=True)

    # Look for it using update stream to see if binlog streamer can talk to
    # dst_replica, which now has binlog_checksum enabled.
    self._wait_for_replica_event(start_position, sql)

 def test_log_rotation(self):
     start_position = _get_master_current_position()
     position = start_position
     master_tablet.mquery("vt_test_keyspace", "flush logs")
     self._exec_vt_txn(self._populate_vt_a(15))
     self._exec_vt_txn(["delete from vt_a"])
     master_conn = self._get_master_stream_conn()
     master_txn_count = 0
     logs_correct = False
     for stream_event in master_conn.stream_update(start_position):
         if stream_event.category == update_stream.StreamEvent.POS:
             master_txn_count += 1
             position = mysql_flavor().position_append(position, stream_event.transaction_id)
             if mysql_flavor().position_after(position, start_position):
                 logs_correct = True
                 logging.debug("Log rotation correctly interpreted")
                 break
             if master_txn_count == 2:
                 self.fail("ran out of logs")
     if not logs_correct:
         self.fail("Flush logs didn't get properly interpreted")

 def test_log_rotation(self):
   start_position = _get_master_current_position()
   position = start_position
   master_tablet.mquery('vt_test_keyspace', 'flush logs')
   self._exec_vt_txn(self._populate_vt_a(15))
   self._exec_vt_txn(['delete from vt_a'])
   master_conn = self._get_master_stream_conn()
   master_conn.dial()
   data = master_conn.stream_start(start_position)
   master_txn_count = 0
   logs_correct = False
   while master_txn_count <= 2:
     data = master_conn.stream_next()
     if data['Category'] == 'POS':
       master_txn_count += 1
       position = mysql_flavor().position_append(position, data['GTIDField'])
       if mysql_flavor().position_after(position, start_position):
         logs_correct = True
         logging.debug('Log rotation correctly interpreted')
         break
   if not logs_correct:
     self.fail("Flush logs didn't get properly interpreted")

def wait_for_replication_pos(tablet_a, tablet_b, timeout=60.0):
  """Waits for tablet B to catch up to the replication position of tablet A.

  Args:
    tablet_a: tablet Object for tablet A.
    tablet_b: tablet Object for tablet B.
    timeout: Timeout in seconds.

  Raises:
    TestError: replication position did not catch up within timeout seconds.
  """
  replication_pos_a = mysql_flavor().master_position(tablet_a)
  while True:
    replication_pos_b = mysql_flavor().master_position(tablet_b)
    if mysql_flavor().position_at_least(replication_pos_b, replication_pos_a):
      break
    timeout = wait_step(
        "%s's replication position to catch up %s's; "
        'currently at: %s, waiting to catch up to: %s' % (
            tablet_b.tablet_alias, tablet_a.tablet_alias, replication_pos_b,
            replication_pos_a),
        timeout, sleep_time=0.1)