def _test_sanity(self):
    # Start up a master mysql and vttablet
    utils.run_vtctl('CreateKeyspace -force test_keyspace')
    utils.run_vtctl('CreateShard -force test_keyspace/0')
    tablet_62344.init_tablet('master', 'test_keyspace', '0', parent=False)
    utils.run_vtctl('RebuildKeyspaceGraph test_keyspace')
    utils.validate_topology()

    # if these statements don't run before the tablet it will wedge waiting for the
    # db to become accessible. this is more a bug than a feature.
    tablet_62344.populate('vt_test_keyspace', self._create_vt_select_test,
                          self._populate_vt_select_test)

    tablet_62344.start_vttablet()

    # make sure the query service is started right away
    result, _ = utils.run_vtctl('Query test_nj test_keyspace "select * from vt_select_test"', trap_output=True)
    rows = result.splitlines()
    self.assertEqual(len(rows), 5, "expected 5 rows in vt_select_test: %s %s" % (str(rows), result))

    # check Pings
    utils.run_vtctl('Ping ' + tablet_62344.tablet_alias)
    utils.run_vtctl('RpcPing ' + tablet_62344.tablet_alias)

    utils.validate_topology()
    utils.run_vtctl('ValidateKeyspace test_keyspace')
    # not pinging tablets, as it enables replication checks, and they
    # break because we only have a single master, no slaves
    utils.run_vtctl('ValidateShard -ping-tablets=false test_keyspace/0')

def setup_tablets():
  # Start up a master mysql and vttablet
  logging.debug("Setting up tablets")
  utils.run_vtctl('CreateKeyspace test_keyspace')
  master_tablet.init_tablet('master', 'test_keyspace', '0')
  replica_tablet.init_tablet('replica', 'test_keyspace', '0')
  utils.run_vtctl('RebuildShardGraph test_keyspace/0')
  utils.validate_topology()
  master_tablet.create_db('vt_test_keyspace')
  replica_tablet.create_db('vt_test_keyspace')

  utils.run_vtctl('RebuildKeyspaceGraph test_keyspace')
  zkocc_server = utils.zkocc_start()

  master_tablet.start_vttablet()
  replica_tablet.start_vttablet()
  utils.run_vtctl('SetReadWrite ' + master_tablet.tablet_alias)
  utils.check_db_read_write(62344)

  for t in [master_tablet, replica_tablet]:
    t.reset_replication()
  utils.run_vtctl('ReparentShard -force test_keyspace/0 ' + master_tablet.tablet_alias, auto_log=True)

  # reset counter so tests don't assert
  tablet.Tablet.tablets_running = 0
  setup_schema()
  master_tablet.vquery("set vt_schema_reload_time=86400", path="test_keyspace/0")
  replica_tablet.vquery("set vt_schema_reload_time=86400", path="test_keyspace/0")

    def setUpClass(klass):
        utils.run_vtctl("CreateKeyspace test_keyspace")

        shard_0_master.init_tablet("master", "test_keyspace", "-80")
        shard_0_replica.init_tablet("replica", "test_keyspace", "-80")
        shard_0_spare.init_tablet("spare", "test_keyspace", "-80")
        shard_1_master.init_tablet("master", "test_keyspace", "80-")
        shard_1_replica.init_tablet("replica", "test_keyspace", "80-")
        idle.init_tablet("idle")
        scrap.init_tablet("idle")

        utils.run_vtctl("RebuildShardGraph /zk/global/vt/keyspaces/test_keyspace/shards/*", auto_log=True)
        utils.run_vtctl("RebuildKeyspaceGraph /zk/global/vt/keyspaces/*", auto_log=True)

        for t in assigned:
            t.create_db("vt_test_keyspace")
            t.start_vttablet()

        for t in scrap, idle, shard_0_spare:
            t.start_vttablet(wait_for_state="NOT_SERVING")

        scrap.scrap()

        for t in [shard_0_master, shard_0_replica, shard_0_spare, shard_1_master, shard_1_replica, idle, scrap]:
            t.reset_replication()
        utils.run_vtctl("ReparentShard -force test_keyspace/-80 " + shard_0_master.tablet_alias, auto_log=True)
        utils.run_vtctl("ReparentShard -force test_keyspace/80- " + shard_1_master.tablet_alias, auto_log=True)

        # run checks now before we start the tablets
        utils.validate_topology()

  def test_actions_and_timeouts(self):
    # Start up a master mysql and vttablet
    utils.run_vtctl(['CreateKeyspace', 'test_keyspace'])

    tablet_62344.init_tablet('master', 'test_keyspace', '0')
    utils.validate_topology()
    tablet_62344.create_db('vt_test_keyspace')
    tablet_62344.start_vttablet()

    utils.run_vtctl(['Ping', tablet_62344.tablet_alias])

    # schedule long action in the background, sleep a little bit to make sure
    # it started to run
    args = (environment.binary_args('vtctl') +
            environment.topo_server().flags() +
            ['-tablet_manager_protocol',
             protocols_flavor().tablet_manager_protocol(),
             '-tablet_protocol', protocols_flavor().tabletconn_protocol(),
             '-log_dir', environment.vtlogroot,
             'Sleep', tablet_62344.tablet_alias, '10s'])
    bg = utils.run_bg(args)
    time.sleep(3)

    # try a frontend RefreshState that should timeout as the tablet is busy
    # running the other one
    _, stderr = utils.run_vtctl(
        ['-wait-time', '3s', 'RefreshState', tablet_62344.tablet_alias],
        expect_fail=True)
    self.assertIn(protocols_flavor().rpc_timeout_message(), stderr)

    # wait for the background vtctl
    bg.wait()

    tablet_62344.kill_vttablet()

def test_multisnapshot_vtctl():
  populate = sum([[
    "insert into vt_insert_test_%s (msg) values ('test %s')" % (i, x)
    for x in xrange(4)] for i in range(6)], [])
  create = ['''create table vt_insert_test_%s (
id bigint auto_increment,
msg varchar(64),
primary key (id)
) Engine=InnoDB''' % i for i in range(6)]

  utils.zk_wipe()

  # Start up a master mysql and vttablet
  utils.run_vtctl('CreateKeyspace -force test_keyspace')

  tablet_62344.init_tablet('master', 'test_keyspace', '0')
  utils.run_vtctl('RebuildShardGraph test_keyspace/0')
  utils.validate_topology()

  tablet_62344.populate('vt_test_keyspace', create,
                        populate)

  tablet_62344.start_vttablet()

  utils.run_vtctl('MultiSnapshot --force --tables=vt_insert_test_1,vt_insert_test_2,vt_insert_test_3 --spec=-0000000000000003- %s id' % tablet_62344.tablet_alias)

  # if err != 0:
  #   raise utils.TestError('mysqlctl multisnapshot failed')
  if os.path.exists(os.path.join(utils.vtdataroot, 'snapshot/vt_0000062344/data/vt_test_keyspace-,0000000000000003/vt_insert_test_4.0.csv.gz')):
    raise utils.TestError("Table vt_insert_test_4 wasn't supposed to be dumped.")
  for kr in 'vt_test_keyspace-,0000000000000003', 'vt_test_keyspace-0000000000000003,':
    path = os.path.join(utils.vtdataroot, 'snapshot/vt_0000062344/data/', kr, 'vt_insert_test_1.0.csv.gz')
    with gzip.open(path) as f:
      if len(f.readlines()) != 2:
        raise utils.TestError("Data looks wrong in %s" % path)

def setup_tablets():
    # Start up a master mysql and vttablet
    utils.debug("Setting up tablets")
    utils.run_vtctl("CreateKeyspace test_keyspace")
    master_tablet.init_tablet("master", "test_keyspace", "0")
    utils.run_vtctl("RebuildShardGraph test_keyspace/0")
    utils.run_vtctl("RebuildKeyspaceGraph test_keyspace")
    utils.validate_topology()

    setup_schema()
    replica_tablet.create_db("vt_test_keyspace")
    # master_tablet.start_vttablet(auth=True)
    master_tablet.start_vttablet()

    replica_tablet.init_tablet("idle", "test_keyspace", start=True)
    snapshot_dir = os.path.join(utils.vtdataroot, "snapshot")
    utils.run("mkdir -p " + snapshot_dir)
    utils.run("chmod +w " + snapshot_dir)
    utils.run_vtctl("Clone -force %s %s" % (master_tablet.tablet_alias, replica_tablet.tablet_alias))

    utils.run_vtctl("Ping test_nj-0000062344")
    utils.run_vtctl("SetReadWrite " + master_tablet.tablet_alias)
    utils.check_db_read_write(62344)

    utils.validate_topology()
    utils.run_vtctl("Ping test_nj-0000062345")

    # reset counter so tests don't assert
    tablet.Tablet.tablets_running = 0

  def setUpClass(klass):
    utils.run_vtctl('CreateKeyspace test_keyspace')

    shard_0_master.init_tablet( 'master',  'test_keyspace', '-80')
    shard_0_replica.init_tablet('replica', 'test_keyspace', '-80')
    shard_0_spare.init_tablet('spare', 'test_keyspace', '-80')
    shard_1_master.init_tablet( 'master',  'test_keyspace', '80-')
    shard_1_replica.init_tablet('replica', 'test_keyspace', '80-')
    idle.init_tablet('idle')
    scrap.init_tablet('idle')

    utils.run_vtctl('RebuildShardGraph /zk/global/vt/keyspaces/test_keyspace/shards/*', auto_log=True)
    utils.run_vtctl('RebuildKeyspaceGraph /zk/global/vt/keyspaces/*', auto_log=True)

    for t in assigned:
      t.create_db('vt_test_keyspace')
      t.start_vttablet()

    for t in scrap, idle, shard_0_spare:
      t.start_vttablet(wait_for_state='NOT_SERVING')

    scrap.scrap()

    utils.run_vtctl('ReparentShard -force test_keyspace/-80 ' + shard_0_master.tablet_alias, auto_log=True)
    utils.run_vtctl('ReparentShard -force test_keyspace/80- ' + shard_1_master.tablet_alias, auto_log=True)


  # run checks now before we start the tablets
    utils.validate_topology()

def setup_tablets():
  # Start up a master mysql and vttablet
  utils.debug("Setting up tablets")
  utils.run_vtctl('CreateKeyspace test_keyspace')
  master_tablet.init_tablet('master', 'test_keyspace', '0')
  utils.run_vtctl('RebuildShardGraph test_keyspace/0')
  utils.run_vtctl('RebuildKeyspaceGraph test_keyspace')
  utils.validate_topology()

  setup_schema()
  replica_tablet.create_db('vt_test_keyspace')
  master_tablet.start_vttablet(memcache=True)

  replica_tablet.init_tablet('idle', 'test_keyspace', start=True, memcache=True)
  snapshot_dir = os.path.join(utils.vtdataroot, 'snapshot')
  utils.run("mkdir -p " + snapshot_dir)
  utils.run("chmod +w " + snapshot_dir)
  utils.run_vtctl('Clone -force %s %s' %
                  (master_tablet.tablet_alias, replica_tablet.tablet_alias))

  utils.run_vtctl('Ping test_nj-0000062344')
  utils.run_vtctl('SetReadWrite ' + master_tablet.tablet_alias)
  utils.check_db_read_write(62344)

  utils.validate_topology()
  utils.run_vtctl('Ping test_nj-0000062345')
  utils.run_vtctl('ChangeSlaveType test_nj-0000062345 replica')

  def test_restart_during_action(self):
    # Start up a master mysql and vttablet
    utils.run_vtctl(['CreateKeyspace', 'test_keyspace'])

    tablet_62344.init_tablet('master', 'test_keyspace', '0')
    utils.run_vtctl(['RebuildShardGraph', 'test_keyspace/0'])
    utils.validate_topology()
    srvShard = utils.run_vtctl_json(['GetSrvShard', 'test_nj', 'test_keyspace/0'])
    self.assertEqual(srvShard['MasterCell'], 'test_nj')
    tablet_62344.create_db('vt_test_keyspace')
    tablet_62344.start_vttablet()

    utils.run_vtctl(['Ping', tablet_62344.tablet_alias])

    # schedule long action
    utils.run_vtctl(['-no-wait', 'Sleep', tablet_62344.tablet_alias, '15s'], stdout=utils.devnull)
    # ping blocks until the sleep finishes unless we have a schedule race
    action_path, _ = utils.run_vtctl(['-no-wait', 'Ping', tablet_62344.tablet_alias], trap_output=True)
    action_path = action_path.strip()

    # kill agent leaving vtaction running
    tablet_62344.kill_vttablet()

    # restart agent
    tablet_62344.start_vttablet()

    # we expect this action with a short wait time to fail. this isn't the best
    # and has some potential for flakiness.
    utils.run_vtctl(['-wait-time', '2s', 'WaitForAction', action_path],
                    expect_fail=True)

    # wait until the background sleep action is done, otherwise there will be
    # a leftover vtaction whose result may overwrite running actions
    # NOTE(alainjobart): Yes, I've seen it happen, it's a pain to debug:
    # the zombie Sleep clobbers the Clone command in the following tests
    utils.run_vtctl(['-wait-time', '20s', 'WaitForAction', action_path],
                    auto_log=True)

    if environment.topo_server_implementation == 'zookeeper':
      # extra small test: we ran for a while, get the states we were in,
      # make sure they're accounted for properly
      # first the query engine States
      v = utils.get_vars(tablet_62344.port)
      logging.debug("vars: %s" % str(v))

      # then the Zookeeper connections
      if v['ZkMetaConn']['test_nj']['Current'] != 'Connected':
        self.fail('invalid zk test_nj state: %s' %
                  v['ZkMetaConn']['test_nj']['Current'])
      if v['ZkMetaConn']['global']['Current'] != 'Connected':
        self.fail('invalid zk global state: %s' %
                  v['ZkMetaConn']['global']['Current'])
      if v['ZkMetaConn']['test_nj']['DurationConnected'] < 10e9:
        self.fail('not enough time in Connected state: %u',
                  v['ZkMetaConn']['test_nj']['DurationConnected'])
      if v['TabletType'] != 'master':
        self.fail('TabletType not exported correctly')

    tablet_62344.kill_vttablet()

  def _test_sanity(self):
    # Start up a master mysql and vttablet
    utils.run_vtctl(['CreateKeyspace', '-force', 'test_keyspace'])
    utils.run_vtctl(['createshard', '-force', 'test_keyspace/0'])
    tablet_62344.init_tablet('master', 'test_keyspace', '0', parent=False)
    utils.run_vtctl(
        ['RebuildKeyspaceGraph', '-rebuild_srv_shards', 'test_keyspace'])
    utils.validate_topology()
    self._check_srv_shard()

    # if these statements don't run before the tablet it will wedge
    # waiting for the db to become accessible. this is more a bug than
    # a feature.
    tablet_62344.populate('vt_test_keyspace', self._create_vt_select_test,
                          self._populate_vt_select_test)

    tablet_62344.start_vttablet()

    # make sure the query service is started right away
    qr = tablet_62344.execute('select * from vt_select_test')
    self.assertEqual(len(qr['rows']), 4,
                     'expected 4 rows in vt_select_test: %s' % str(qr))

    # make sure direct dba queries work
    query_result = utils.run_vtctl_json(
        ['ExecuteFetchAsDba', '-json', tablet_62344.tablet_alias,
         'select * from vt_test_keyspace.vt_select_test'])
    self.assertEqual(
        len(query_result['rows']), 4,
        'expected 4 rows in vt_select_test: %s' % str(query_result))
    self.assertEqual(
        len(query_result['fields']), 2,
        'expected 2 fields in vt_select_test: %s' % str(query_result))

    # check Ping / RefreshState
    utils.run_vtctl(['Ping', tablet_62344.tablet_alias])
    utils.run_vtctl(['RefreshState', tablet_62344.tablet_alias])

    # Quickly check basic actions.
    utils.run_vtctl(['SetReadOnly', tablet_62344.tablet_alias])
    utils.wait_db_read_only(62344)

    utils.run_vtctl(['SetReadWrite', tablet_62344.tablet_alias])
    utils.check_db_read_write(62344)

    utils.run_vtctl(['DemoteMaster', tablet_62344.tablet_alias])
    utils.wait_db_read_only(62344)

    utils.validate_topology()
    utils.run_vtctl(['ValidateKeyspace', 'test_keyspace'])
    # not pinging tablets, as it enables replication checks, and they
    # break because we only have a single master, no slaves
    utils.run_vtctl(['ValidateShard', '-ping-tablets=false',
                     'test_keyspace/0'])
    self._check_srv_shard()

    tablet_62344.kill_vttablet()

    def test_actions_and_timeouts(self):
        # Start up a master mysql and vttablet
        utils.run_vtctl(["CreateKeyspace", "test_keyspace"])

        tablet_62344.init_tablet("master", "test_keyspace", "0")
        utils.run_vtctl(["RebuildShardGraph", "test_keyspace/0"])
        utils.validate_topology()
        self._check_srv_shard()
        tablet_62344.create_db("vt_test_keyspace")
        tablet_62344.start_vttablet()

        utils.run_vtctl(["Ping", tablet_62344.tablet_alias])

        # schedule long action in the background, sleep a little bit to make sure
        # it started to run
        args = (
            environment.binary_args("vtctl")
            + environment.topo_server().flags()
            + [
                "-tablet_manager_protocol",
                protocols_flavor().tablet_manager_protocol(),
                "-tablet_protocol",
                protocols_flavor().tabletconn_protocol(),
                "-log_dir",
                environment.vtlogroot,
                "Sleep",
                tablet_62344.tablet_alias,
                "10s",
            ]
        )
        bg = utils.run_bg(args)
        time.sleep(3)

        # try a frontend RefreshState that should timeout as the tablet is busy
        # running the other one
        _, stderr = utils.run_vtctl(["-wait-time", "3s", "RefreshState", tablet_62344.tablet_alias], expect_fail=True)
        self.assertIn(protocols_flavor().rpc_timeout_message(), stderr)

        # wait for the background vtctl
        bg.wait()

        if environment.topo_server().flavor() == "zookeeper":
            # extra small test: we ran for a while, get the states we were in,
            # make sure they're accounted for properly
            # first the query engine States
            v = utils.get_vars(tablet_62344.port)
            logging.debug("vars: %s", v)

            # then the Zookeeper connections
            if v["ZkCachedConn"]["test_nj"] != "Connected":
                self.fail("invalid zk test_nj state: %s" % v["ZkCachedConn"]["test_nj"])
            if v["ZkCachedConn"]["global"] != "Connected":
                self.fail("invalid zk global state: %s" % v["ZkCachedConn"]["global"])
            if v["TabletType"] != "master":
                self.fail("TabletType not exported correctly")

        tablet_62344.kill_vttablet()

def run_test_mysqlctl_split():
  utils.zk_wipe()

  # Start up a master mysql and vttablet
  utils.run_vtctl('CreateKeyspace -force test_keyspace')

  tablet_62344.init_tablet('master', 'test_keyspace', '0')
  utils.run_vtctl('RebuildShardGraph test_keyspace/0')
  utils.validate_topology()

  tablet_62344.populate('vt_test_keyspace', create_vt_insert_test,
                        populate_vt_insert_test)

  tablet_62344.start_vttablet()

  err = tablet_62344.mysqlctl('-port %u -mysql-port %u partialsnapshot -end=0000000000000003 vt_test_keyspace id' % (tablet_62344.port, tablet_62344.mysql_port)).wait()
  if err != 0:
    raise utils.TestError('mysqlctl partialsnapshot failed')


  utils.pause("partialsnapshot finished")

  tablet_62044.mquery('', 'stop slave')
  tablet_62044.create_db('vt_test_keyspace')
  call(["touch", "/tmp/vtSimulateFetchFailures"])
  err = tablet_62044.mysqlctl('-port %u -mysql-port %u partialrestore %s/snapshot/vt_0000062344/data/vt_test_keyspace-,0000000000000003/partial_snapshot_manifest.json' % (tablet_62044.port, tablet_62044.mysql_port, utils.vtdataroot)).wait()
  if err != 0:
    raise utils.TestError('mysqlctl partialrestore failed')

  tablet_62044.assert_table_count('vt_test_keyspace', 'vt_insert_test', 2)

  # change/add two values on the master, one in range, one out of range, make
  # sure the right one propagate and not the other
  utils.run_vtctl('SetReadWrite ' + tablet_62344.tablet_alias)
  tablet_62344.mquery('vt_test_keyspace', "insert into vt_insert_test (id, msg) values (5, 'test should not propagate')", write=True)
  tablet_62344.mquery('vt_test_keyspace', "update vt_insert_test set msg='test should propagate' where id=2", write=True)

  utils.pause("look at db now!")

  # wait until value that should have been changed is here
  timeout = 10
  while timeout > 0:
    result = tablet_62044.mquery('vt_test_keyspace', 'select msg from vt_insert_test where id=2')
    if result[0][0] == "test should propagate":
      break
    timeout -= 1
    time.sleep(1)
  if timeout == 0:
    raise utils.TestError("expected propagation to happen", result)

  # test value that should not propagate
  # this part is disabled now, as the replication pruning is only enabled
  # for row-based replication, but the mysql server is statement based.
  # will re-enable once we get statement-based pruning patch into mysql.
#  tablet_62044.assert_table_count('vt_test_keyspace', 'vt_insert_test', 0, 'where id=5')

  tablet_62344.kill_vttablet()

def test_multisnapshot_and_restore_vtctl():
    tables = ["vt_insert_test", "vt_insert_test1"]
    create_template = """create table %s (
id bigint auto_increment,
msg varchar(64),
primary key (id)
) Engine=InnoDB"""
    insert_template = "insert into %s (id, msg) values (%s, 'test %s')"
    utils.zk_wipe()

    # Start up a master mysql and vttablet
    utils.run_vtctl("CreateKeyspace -force test_keyspace")

    # Start three tablets for three different shards. At this point the
    # sharding schema is not really important, as long as it is
    # consistent.
    new_spec = "-0000000000000028-"
    old_tablets = [tablet_62044, tablet_41983, tablet_31981]
    for i, tablet in enumerate(old_tablets):
        tablet.init_tablet("master", "test_keyspace", str(i))
        utils.run_vtctl("RebuildShardGraph test_keyspace/%s" % i)
    utils.validate_topology()

    for i, tablet in enumerate(old_tablets):
        tablet.populate(
            "vt_test_keyspace",
            [create_template % table for table in tables],
            sum([[insert_template % (table, 10 * j + i, 10 * j + i) for j in range(1, 8)] for table in tables], []),
        )
        tablet.start_vttablet()
        utils.run_vtctl(
            "MultiSnapshot -force -maximum-file-size=1 -spec=%s %s id" % (new_spec, tablet.tablet_alias),
            trap_output=True,
        )

    utils.run_vtctl("CreateKeyspace -force test_keyspace_new")
    tablet_62344.init_tablet("master", "test_keyspace_new", "-0000000000000028", dbname="not_vt_test_keyspace")
    utils.run_vtctl("RebuildShardGraph test_keyspace_new/-0000000000000028")
    utils.validate_topology()
    tablet_62344.mquery("", "DROP DATABASE IF EXISTS not_vt_test_keyspace")
    tablet_62344.start_vttablet(wait_for_state="CONNECTING")  # db not created

    # 0x28 = 40
    source_aliases = " ".join(t.tablet_alias for t in old_tablets)
    utils.run_vtctl(
        "MultiRestore %s %s" % (tablet_62344.tablet_alias, source_aliases), auto_log=True, raise_on_error=True
    )
    time.sleep(1)
    for table in tables:
        rows = tablet_62344.mquery("not_vt_test_keyspace", "select id from %s" % table)
        if len(rows) == 0:
            raise utils.TestError("There are no rows in the restored database.")
        for row in rows:
            if row[0] > 32:
                raise utils.TestError("Bad row: %s" % row)
    for tablet in tablet_62044, tablet_41983, tablet_31981, tablet_62344:
        tablet.kill_vttablet()

  def test_vtgate(self):
    # Start up a master mysql and vttablet
    utils.run_vtctl('CreateKeyspace -force test_keyspace')
    utils.run_vtctl('CreateShard -force test_keyspace/0')
    tablet_62344.init_tablet('master', 'test_keyspace', '0', parent=False)
    utils.run_vtctl('RebuildShardGraph test_keyspace/0')
    utils.run_vtctl('RebuildKeyspaceGraph test_keyspace')
    utils.validate_topology()

    # if these statements don't run before the tablet it will wedge waiting for the
    # db to become accessible. this is more a bug than a feature.
    tablet_62344.mquery("", ["set global read_only = off"])
    tablet_62344.populate('vt_test_keyspace', self._create_vt_select_test,
                          self._populate_vt_select_test)

    tablet_62344.start_vttablet()
    gate_proc = utils.vtgate_start()

    try:
      conn = vtgate.connect("localhost:%s"%(utils.vtgate_port_base), "master", "test_keyspace", "0", 2.0)

      # _execute
      (result, count, lastrow, fields) = conn._execute("select * from vt_select_test", {})
      self.assertEqual(count, 4, "want 4, got %d" % (count))
      self.assertEqual(len(fields), 2, "want 2, got %d" % (len(fields)))

      # _stream_execute
      (result, count, lastrow, fields) = conn._stream_execute("select * from vt_select_test", {})
      self.assertEqual(len(fields), 2, "want 2, got %d" % (len(fields)))
      count = 0
      while 1:
        r = conn._stream_next()
        if not r:
          break
        count += 1
      self.assertEqual(count, 4, "want 4, got %d" % (count))

      # begin-rollback
      conn.begin()
      conn._execute("insert into vt_select_test values(:id, :msg)", {"id": 5, "msg": "test4"})
      conn.rollback()
      (result, count, lastrow, fields) = conn._execute("select * from vt_select_test", {})
      self.assertEqual(count, 4, "want 4, got %d" % (count))

      # begin-commit
      conn.begin()
      conn._execute("insert into vt_select_test values(:id, :msg)", {"id": 5, "msg": "test4"})
      conn.commit()
      (result, count, lastrow, fields) = conn._execute("select * from vt_select_test", {})
      self.assertEqual(count, 5, "want 5, got %d" % (count))

      # close
      conn.close()
    finally:
      utils.vtgate_kill(gate_proc)
      tablet_62344.kill_vttablet()

def setUpModule():
  global vtgate_server
  global vtgate_port
  global vtgate_socket_file
  global master_start_position

  try:
    environment.topo_server_setup()

    # start mysql instance external to the test
    setup_procs = [master_tablet.init_mysql(),
                   replica_tablet.init_mysql()
                   ]
    utils.wait_procs(setup_procs)

    # Start up a master mysql and vttablet
    logging.debug("Setting up tablets")
    utils.run_vtctl(['CreateKeyspace', 'test_keyspace'])
    master_tablet.init_tablet('master', 'test_keyspace', '0')
    replica_tablet.init_tablet('replica', 'test_keyspace', '0')
    utils.run_vtctl(['RebuildShardGraph', 'test_keyspace/0'])
    utils.validate_topology()
    master_tablet.create_db('vt_test_keyspace')
    master_tablet.create_db('other_database')
    replica_tablet.create_db('vt_test_keyspace')
    replica_tablet.create_db('other_database')

    utils.run_vtctl(['RebuildKeyspaceGraph', 'test_keyspace'])

    vtgate_socket_file = environment.tmproot + '/vtgate.sock'
    vtgate_server, vtgate_port = utils.vtgate_start(socket_file=vtgate_socket_file)

    master_tablet.start_vttablet()
    replica_tablet.start_vttablet()
    utils.run_vtctl(['SetReadWrite', master_tablet.tablet_alias])
    utils.check_db_read_write(master_tablet.tablet_uid)

    for t in [master_tablet, replica_tablet]:
      t.reset_replication()
    utils.run_vtctl(['ReparentShard', '-force', 'test_keyspace/0',
                     master_tablet.tablet_alias], auto_log=True)

    # reset counter so tests don't assert
    tablet.Tablet.tablets_running = 0

    master_start_position = _get_master_current_position()
    master_tablet.mquery('vt_test_keyspace', _create_vt_insert_test)
    master_tablet.mquery('vt_test_keyspace', _create_vt_a)
    master_tablet.mquery('vt_test_keyspace', _create_vt_b)

    utils.run_vtctl(['ReloadSchema', master_tablet.tablet_alias])
    utils.run_vtctl(['ReloadSchema', replica_tablet.tablet_alias])

  except:
    tearDownModule()
    raise

  def test_actions_and_timeouts(self):
    # Start up a master mysql and vttablet
    utils.run_vtctl(['CreateKeyspace', 'test_keyspace'])

    tablet_62344.init_tablet('master', 'test_keyspace', '0')
    utils.run_vtctl(['RebuildShardGraph', 'test_keyspace/0'])
    utils.validate_topology()
    self._check_srv_shard()
    tablet_62344.create_db('vt_test_keyspace')
    tablet_62344.start_vttablet()

    utils.run_vtctl(['Ping', tablet_62344.tablet_alias])

    # schedule long action in the background, sleep a little bit to make sure
    # it started to run
    args = (environment.binary_args('vtctl') +
            environment.topo_server().flags() +
            ['-tablet_manager_protocol',
             protocols_flavor().tablet_manager_protocol(),
             '-tablet_protocol', protocols_flavor().tabletconn_protocol(),
             '-log_dir', environment.vtlogroot,
             'Sleep', tablet_62344.tablet_alias, '10s'])
    bg = utils.run_bg(args)
    time.sleep(3)

    # try a frontend RefreshState that should timeout as the tablet is busy
    # running the other one
    stdout, stderr = utils.run_vtctl(
        ['-wait-time', '3s', 'RefreshState', tablet_62344.tablet_alias],
        expect_fail=True)
    self.assertIn(protocols_flavor().rpc_timeout_message(), stderr)

    # wait for the background vtctl
    bg.wait()

    if environment.topo_server().flavor() == 'zookeeper':
      # extra small test: we ran for a while, get the states we were in,
      # make sure they're accounted for properly
      # first the query engine States
      v = utils.get_vars(tablet_62344.port)
      logging.debug('vars: %s', v)

      # then the Zookeeper connections
      if v['ZkMetaConn']['test_nj']['Current'] != 'Connected':
        self.fail('invalid zk test_nj state: %s' %
                  v['ZkMetaConn']['test_nj']['Current'])
      if v['ZkMetaConn']['global']['Current'] != 'Connected':
        self.fail('invalid zk global state: %s' %
                  v['ZkMetaConn']['global']['Current'])
      if v['ZkMetaConn']['test_nj']['DurationConnected'] < 10e9:
        self.fail('not enough time in Connected state: %d',
                  v['ZkMetaConn']['test_nj']['DurationConnected'])
      if v['TabletType'] != 'master':
        self.fail('TabletType not exported correctly')

    tablet_62344.kill_vttablet()

  def test_reparent_lag_slave(self, shard_id='0'):
    utils.run_vtctl('CreateKeyspace test_keyspace')

    # create the database so vttablets start, as they are serving
    tablet_62344.create_db('vt_test_keyspace')
    tablet_62044.create_db('vt_test_keyspace')
    tablet_41983.create_db('vt_test_keyspace')
    tablet_31981.create_db('vt_test_keyspace')

    # Start up a master mysql and vttablet
    tablet_62344.init_tablet('master', 'test_keyspace', shard_id, start=True, wait_for_start=False)

    # Create a few slaves for testing reparenting.
    tablet_62044.init_tablet('replica', 'test_keyspace', shard_id, start=True, wait_for_start=False)
    tablet_31981.init_tablet('replica', 'test_keyspace', shard_id, start=True, wait_for_start=False)
    tablet_41983.init_tablet('lag', 'test_keyspace', shard_id, start=True, wait_for_start=False)

    # wait for all tablets to start
    for t in [tablet_62344, tablet_62044, tablet_31981]:
      t.wait_for_vttablet_state("SERVING")
    tablet_41983.wait_for_vttablet_state("NOT_SERVING")

    # Recompute the shard layout node - until you do that, it might not be valid.
    utils.run_vtctl('RebuildShardGraph test_keyspace/' + shard_id)
    utils.validate_topology()

    # Force the slaves to reparent assuming that all the datasets are identical.
    for t in [tablet_62344, tablet_62044, tablet_41983, tablet_31981]:
      t.reset_replication()
    utils.run_vtctl('ReparentShard -force test_keyspace/%s %s' % (shard_id, tablet_62344.tablet_alias))
    utils.validate_topology(ping_tablets=True)

    tablet_62344.create_db('vt_test_keyspace')
    tablet_62344.mquery('vt_test_keyspace', self._create_vt_insert_test)

    tablet_41983.mquery('', 'stop slave')
    for q in self._populate_vt_insert_test:
      tablet_62344.mquery('vt_test_keyspace', q, write=True)

    # Perform a graceful reparent operation.
    utils.run_vtctl('ReparentShard test_keyspace/%s %s' % (shard_id, tablet_62044.tablet_alias))

    tablet_41983.mquery('', 'start slave')
    time.sleep(1)

    utils.pause("check orphan")

    utils.run_vtctl('ReparentTablet %s' % tablet_41983.tablet_alias)

    result = tablet_41983.mquery('vt_test_keyspace', 'select msg from vt_insert_test where id=1')
    if len(result) != 1:
      self.fail('expected 1 row from vt_insert_test: %s' % str(result))

    utils.pause("check lag reparent")

    tablet.kill_tablets([tablet_62344, tablet_62044, tablet_41983, tablet_31981])

  def setUpClass(klass):
    utils.run_vtctl(['CreateKeyspace', 'test_keyspace'])
    utils.run_vtctl(['CreateKeyspace',
                     '--served-from', 'master:test_keyspace,replica:test_keyspace,rdonly:test_keyspace',
                     'redirected_keyspace'])

    shard_0_master.init_tablet( 'master',  'test_keyspace', '-80')
    shard_0_replica.init_tablet('spare',   'test_keyspace', '-80')
    shard_0_spare.init_tablet(  'spare',   'test_keyspace', '-80')
    shard_1_master.init_tablet( 'master',  'test_keyspace', '80-')
    shard_1_replica.init_tablet('replica', 'test_keyspace', '80-')
    idle.init_tablet('idle')
    scrap.init_tablet('idle')

    utils.run_vtctl(['RebuildKeyspaceGraph', 'test_keyspace'], auto_log=True)
    utils.run_vtctl(['RebuildKeyspaceGraph', 'redirected_keyspace'],
                    auto_log=True)

    # start running all the tablets
    for t in [shard_0_master, shard_1_master, shard_1_replica]:
      t.create_db('vt_test_keyspace')
      t.start_vttablet(wait_for_state=None,
                       extra_args=utils.vtctld.process_args())
    shard_0_replica.create_db('vt_test_keyspace')
    shard_0_replica.start_vttablet(extra_args=utils.vtctld.process_args(),
                                   target_tablet_type='replica',
                                   wait_for_state=None)

    for t in scrap, idle, shard_0_spare:
      t.start_vttablet(wait_for_state=None,
                       extra_args=utils.vtctld.process_args())

    # wait for the right states
    for t in [shard_0_master, shard_1_master, shard_1_replica]:
      t.wait_for_vttablet_state('SERVING')
    for t in [scrap, idle, shard_0_replica, shard_0_spare]:
      t.wait_for_vttablet_state('NOT_SERVING')

    scrap.scrap()

    for t in [shard_0_master, shard_0_replica, shard_0_spare,
              shard_1_master, shard_1_replica, idle, scrap]:
      t.reset_replication()
    utils.run_vtctl(['ReparentShard', '-force', 'test_keyspace/-80',
                     shard_0_master.tablet_alias], auto_log=True)
    utils.run_vtctl(['ReparentShard', '-force', 'test_keyspace/80-',
                     shard_1_master.tablet_alias], auto_log=True)
    shard_0_replica.wait_for_vttablet_state('SERVING')

    # run checks now before we start the tablets
    utils.validate_topology()

    # start a vtgate server too
    global vtgate_server, vtgate_port
    vtgate_server, vtgate_port = utils.vtgate_start(
        cache_ttl='0s', extra_args=utils.vtctld.process_args())

def run_test_reparent_lag_slave(shard_id="0"):
    utils.zk_wipe()

    utils.run_vtctl("CreateKeyspace -force test_keyspace")

    # create the database so vttablets start, as they are serving
    tablet_62344.create_db("vt_test_keyspace")
    tablet_62044.create_db("vt_test_keyspace")
    tablet_41983.create_db("vt_test_keyspace")
    tablet_31981.create_db("vt_test_keyspace")

    # Start up a master mysql and vttablet
    tablet_62344.init_tablet("master", "test_keyspace", shard_id, start=True)

    # Create a few slaves for testing reparenting.
    tablet_62044.init_tablet("replica", "test_keyspace", shard_id, start=True)
    tablet_31981.init_tablet("replica", "test_keyspace", shard_id, start=True)
    tablet_41983.init_tablet("lag", "test_keyspace", shard_id, start=True)

    # Recompute the shard layout node - until you do that, it might not be valid.
    utils.run_vtctl("RebuildShardGraph test_keyspace/" + shard_id)
    utils.validate_topology()

    # Force the slaves to reparent assuming that all the datasets are identical.
    for t in [tablet_62344, tablet_62044, tablet_41983, tablet_31981]:
        t.reset_replication()
    utils.run_vtctl("ReparentShard -force test_keyspace/%s %s" % (shard_id, tablet_62344.tablet_alias))
    utils.validate_topology(ping_tablets=True)

    tablet_62344.create_db("vt_test_keyspace")
    tablet_62344.mquery("vt_test_keyspace", create_vt_insert_test)

    tablet_41983.mquery("", "stop slave")
    for q in populate_vt_insert_test:
        tablet_62344.mquery("vt_test_keyspace", q, write=True)

    # Perform a graceful reparent operation.
    utils.run_vtctl("ReparentShard test_keyspace/%s %s" % (shard_id, tablet_62044.tablet_alias))

    tablet_41983.mquery("", "start slave")
    time.sleep(1)

    utils.pause("check orphan")

    utils.run_vtctl("ReparentTablet %s" % tablet_41983.tablet_alias)

    result = tablet_41983.mquery("vt_test_keyspace", "select msg from vt_insert_test where id=1")
    if len(result) != 1:
        raise utils.TestError("expected 1 row from vt_insert_test", result)

    utils.paus