def setUpModule():
    try:
        if utils.options.xvfb:
            try:
                # This will be killed automatically by utils.kill_sub_processes()
                utils.run_bg(["Xvfb", ":15", "-ac"])
                os.environ["DISPLAY"] = ":15"
            except OSError as err:
                # Despite running in background, utils.run_bg() will throw immediately
                # if the Xvfb binary is not found.
                logging.error("Can't start Xvfb (will try local DISPLAY instead): %s", err)
    except:
        tearDownModule()
        raise

 def start_memcache(self):
     self.memcache_path = os.path.join(self.tablet_dir, "memcache.sock")
     try:
       self.memcached = utils.run_bg(' '.join(["memcached", "-s", self.memcache_path]), stdout=utils.devnull)
     except Exception as e:
       print "Error: memcached couldn't start"
       raise

  def start_vttablet(self, port=None, auth=False, memcache=False, wait_for_state="SERVING", customrules=None, schema_override=None, cert=None, key=None, ca_cert=None, repl_extra_flags={}):
    """
    Starts a vttablet process, and returns it.
    The process is also saved in self.proc, so it's easy to kill as well.
    """
    environment.prog_compile('vtaction')
    args = [environment.binary_path('vttablet'),
            '-port', '%s' % (port or self.port),
            '-tablet-path', self.tablet_alias,
            '-log_dir', environment.vtlogroot]
    args.extend(environment.topo_server_flags())
    args.extend(environment.binlog_player_protocol_flags())

    dbconfigs = self._get_db_configs_file(repl_extra_flags)
    for key1 in dbconfigs:
      for key2 in dbconfigs[key1]:
        args.extend(["-db-config-"+key1+"-"+key2, dbconfigs[key1][key2]])

    if memcache:
      if os.path.exists(environment.vtroot + "/bin/memcached"):
        args.extend(["-rowcache-bin", environment.vtroot + "/bin/memcached"])
      else:
        args.extend(["-rowcache-bin", "memcached"])
      memcache_socket = os.path.join(self.tablet_dir, "memcache.sock")
      args.extend(["-rowcache-socket", memcache_socket])
      args.extend(["-enable-rowcache"])

    if auth:
      args.extend(['-auth-credentials', os.path.join(environment.vttop, 'test', 'test_data', 'authcredentials_test.json')])

    if customrules:
      args.extend(['-customrules', customrules])

    if schema_override:
      args.extend(['-schema-override', schema_override])

    if cert:
      self.secure_port = environment.reserve_ports(1)
      args.extend(['-secure-port', '%s' % self.secure_port,
                   '-cert', cert,
                   '-key', key])
      if ca_cert:
        args.extend(['-ca-cert', ca_cert])

    stderr_fd = open(os.path.join(self.tablet_dir, "vttablet.stderr"), "w")
    # increment count only the first time
    if not self.proc:
      Tablet.tablets_running += 1
    self.proc = utils.run_bg(args, stderr=stderr_fd)
    stderr_fd.close()

    # wait for zookeeper PID just to be sure we have it
    if environment.topo_server_implementation == 'zookeeper':
      utils.run(environment.binary_path('zk')+' wait -e ' + self.zk_pid, stdout=utils.devnull)

    # wait for query service to be in the right state
    if wait_for_state:
      self.wait_for_vttablet_state(wait_for_state, port=port)

    return self.proc

  def __init__(self, name):
    import environment
    import utils

    self.port_base = environment.reserve_ports(2)

    self.name = name
    self.hostname = 'localhost'
    self.client_port = self.port_base
    self.peer_port = self.port_base + 1
    self.client_addr = '%s:%u' % (self.hostname, self.client_port)
    self.peer_addr = '%s:%u' % (self.hostname, self.peer_port)
    self.api_url = 'http://%s/v2' % (self.client_addr)

    dirname = 'etcd_' + self.name
    self.data_dir = os.path.join(environment.vtdataroot, dirname)

    self.proc = utils.run_bg([
        'etcd', '-name', self.name, '-addr',
        self.client_addr, '-peer-addr',
        self.peer_addr, '-data-dir', self.data_dir],
                             stdout=open(os.path.join(
                                 environment.vtlogroot,
                                 dirname + '.stdout'),
                                         'w'),
                             stderr=open(os.path.join(
                                 environment.vtlogroot,
                                 dirname + '.stderr'),
                                         'w'),)

def setUpModule():
  global vtgateclienttest_process
  global vtgateclienttest_port
  global vtgateclienttest_grpc_port

  try:
    environment.topo_server().setup()

    vtgateclienttest_port = environment.reserve_ports(1)
    args = environment.binary_args('vtgateclienttest') + [
        '-log_dir', environment.vtlogroot,
        '-port', str(vtgateclienttest_port),
        ]

    if protocols_flavor().vtgate_python_protocol() == 'grpc':
      vtgateclienttest_grpc_port = environment.reserve_ports(1)
      args.extend(['-grpc_port', str(vtgateclienttest_grpc_port)])
    if protocols_flavor().service_map():
      args.extend(['-service_map', ','.join(protocols_flavor().service_map())])

    vtgateclienttest_process = utils.run_bg(args)
    utils.wait_for_vars('vtgateclienttest', vtgateclienttest_port)
  except:
    tearDownModule()
    raise

def run_test_sigterm():
  utils.zk_wipe()
  utils.run_vtctl('CreateKeyspace -force test_keyspace')

  # create the database so vttablets start, as it is serving
  tablet_62344.create_db('vt_test_keyspace')

  tablet_62344.init_tablet('master', 'test_keyspace', '0', start=True)

  # start a 'vtctl Sleep' command in the background
  sp = utils.run_bg(utils.vtroot+'/bin/vtctl -logfile=/dev/null Sleep %s 60s' %
                    tablet_62344.tablet_alias,
                    stdout=PIPE, stderr=PIPE)

  # wait for it to start, and let's kill it
  time.sleep(2.0)
  utils.run(['pkill', 'vtaction'])
  out, err = sp.communicate()

  # check the vtctl command got the right remote error back
  if "vtaction interrupted by signal" not in err:
    raise utils.TestError("cannot find expected output in error:", err)
  utils.debug("vtaction was interrupted correctly:\n" + err)

  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.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_vtgate_qps(self):
    # create the topology
    utils.run_vtctl('CreateKeyspace test_keyspace')
    t = tablet.Tablet(tablet_uid=1, cell="nj")
    t.init_tablet("master", "test_keyspace", "0")
    t.update_addrs()
    utils.run_vtctl('RebuildShardGraph test_keyspace/0', auto_log=True)
    utils.run_vtctl('RebuildKeyspaceGraph test_keyspace', auto_log=True)

    # start vtgate and the qps-er
    vtgate_proc, vtgate_port = utils.vtgate_start()
    qpser = utils.run_bg(utils.vtroot+'/bin/zkclient2 -server localhost:%u -mode qps2 test_nj test_keyspace' % vtgate_port)
    time.sleep(10)
    utils.kill_sub_process(qpser)

    # get the vtgate vars, make sure we have what we need
    v = utils.get_vars(vtgate_port)

    # some checks on performance / stats
    # a typical workstation will do 38-40k QPS, check we have more than 15k
    rpcCalls = v['TopoReaderRpcQueryCount']['test_nj']
    if rpcCalls < 150000:
      self.fail('QPS is too low: %u < 15000' % (rpcCalls / 10))
    else:
      logging.debug("Recorded qps: %u", rpcCalls / 10)
    utils.vtgate_kill(vtgate_proc)

  def test_zkocc_qps(self):
    # preload the test_nj cell
    zkocc_14850 = utils.zkocc_start()

    qpser = utils.run_bg(utils.vtroot+'/bin/zkclient2 -server localhost:%u -mode qps /zk/test_nj/vt/zkocc1/data1 /zk/test_nj/vt/zkocc1/data2' % utils.zkocc_port_base)
    time.sleep(10)
    utils.kill_sub_process(qpser)

    # get the zkocc vars, make sure we have what we need
    v = utils.get_vars(utils.zkocc_port_base)
    if v['ZkReader']['test_nj']['State'] != 'Connected':
      raise utils.TestError('invalid zk global state: ', v['ZkReader']['test_nj']['State'])

    # some checks on performance / stats
    # a typical workstation will do 45-47k QPS, check we have more than 15k
    rpcCalls = v['ZkReader']['RpcCalls']
    if rpcCalls < 150000:
      self.fail('QPS is too low: %u < 15000' % (rpcCalls / 10))
    else:
      logging.debug("Recorded qps: %u", rpcCalls / 10)
    cacheReads = v['ZkReader']['test_nj']['CacheReads']
    if cacheReads < 150000:
      self.fail('Cache QPS is too low: %u < 15000' % (cacheReads / 10))
    totalCacheReads = v['ZkReader']['total']['CacheReads']
    self.assertEqual(cacheReads, totalCacheReads, 'Rollup stats are wrong')
    self.assertEqual(v['ZkReader']['UnknownCellErrors'], 0, 'unexpected UnknownCellErrors')
    utils.zkocc_kill(zkocc_14850)

  def test_zkocc_qps(self):
    # preload the test_nj cell
    zkocc_14850 = utils.zkocc_start()

    qpser = utils.run_bg(environment.binary_argstr('zkclient2')+' -server localhost:%u -mode qps /zk/test_nj/vt/zkocc1/data1 /zk/test_nj/vt/zkocc1/data2' % environment.topo_server().zkocc_port_base)
    qpser.wait()

    # get the zkocc vars, make sure we have what we need
    v = utils.get_vars(environment.topo_server().zkocc_port_base)
    if v['ZkReader']['test_nj']['State'] != 'Connected':
      self.fail('invalid zk global state: ' + v['ZkReader']['test_nj']['State'])

    # some checks on performance / stats
    rpcCalls = v['ZkReader']['RpcCalls']
    if rpcCalls < MIN_QPS * 10:
      self.fail('QPS is too low: %u < %u' % (rpcCalls / 10, MIN_QPS))
    else:
      logging.debug("Recorded qps: %u", rpcCalls / 10)
    cacheReads = v['ZkReader']['test_nj']['CacheReads']
    if cacheReads < MIN_QPS * 10:
      self.fail('Cache QPS is too low: %u < %u' % (cacheReads, MIN_QPS * 10))
    totalCacheReads = v['ZkReader']['total']['CacheReads']
    self.assertEqual(cacheReads, totalCacheReads, 'Rollup stats are wrong')
    self.assertEqual(v['ZkReader']['UnknownCellErrors'], 0, 'unexpected UnknownCellErrors')
    utils.zkocc_kill(zkocc_14850)

def run_test_zkocc_qps():
  _populate_zk()

  # preload the test_nj cell
  zkocc_14850 = utils.zkocc_start()

  qpser = utils.run_bg(utils.vtroot+'/bin/zkclient2 -server localhost:%u -mode qps /zk/test_nj/zkocc1/data1 /zk/test_nj/zkocc1/data2' % utils.zkocc_port_base)
  time.sleep(10)
  utils.kill_sub_process(qpser)

  # get the zkocc vars, make sure we have what we need
  v = utils.get_vars(utils.zkocc_port_base)
  if v['ZkReader']['test_nj']['State']['Current'] != 'Connected':
    raise utils.TestError('invalid zk global state: ', v['ZkReader']['test_nj']['State']['Current'])
  if v['ZkReader']['test_nj']['State']['DurationConnected'] < 9e9:
    raise utils.TestError('not enough time in Connected state', v['ZkReader']['test_nj']['State']['DurationConnected'])

  # some checks on performance / stats
  # a typical workstation will do 15k QPS, check we have more than 3k
  rpcCalls = v['ZkReader']['RpcCalls']
  if rpcCalls < 30000:
    raise utils.TestError('QPS is too low: %u < 30000', rpcCalls / 10)
  cacheReads = v['ZkReader']['test_nj']['CacheReads']
  if cacheReads < 30000:
    raise utils.TestError('Cache QPS is too low: %u < 30000', cacheReads / 10)
  totalCacheReads = v['ZkReader']['total']['CacheReads']
  if cacheReads != totalCacheReads:
    raise utils.TestError('Rollup stats are wrong: %u != %u', cacheReads,
                          totalCacheReads)
  if v['ZkReader']['UnknownCellErrors'] != 0:
    raise utils.TestError('unexpected UnknownCellErrors', v['ZkReader']['UnknownCellErrors'])

  utils.zkocc_kill(zkocc_14850)

  def mysqlctl(self, cmd, extra_my_cnf=None, with_ports=False, verbose=False):
    """Runs a mysqlctl command.

    Args:
      cmd: the command to run.
      extra_my_cnf: list of extra mycnf files to use
      with_ports: if set, sends the tablet and mysql ports to mysqlctl.
      verbose: passed to mysqlctld.
    Returns:
      the result of run_bg.
    """
    extra_env = {}
    all_extra_my_cnf = get_all_extra_my_cnf(extra_my_cnf)
    if all_extra_my_cnf:
      extra_env['EXTRA_MY_CNF'] = ':'.join(all_extra_my_cnf)
    args = environment.binary_args('mysqlctl') + [
        '-log_dir', environment.vtlogroot,
        '-tablet_uid', str(self.tablet_uid)]
    if self.use_mysqlctld:
      args.extend(
          ['-mysqlctl_socket', os.path.join(self.tablet_dir, 'mysqlctl.sock')])
    if with_ports:
      args.extend(['-port', str(self.port),
                   '-mysql_port', str(self.mysql_port)])
    self._add_dbconfigs(args)
    if verbose:
      args.append('-alsologtostderr')
    args.extend(cmd)
    return utils.run_bg(args, extra_env=extra_env)

    def mysqlctld(self, cmd, extra_my_cnf=None, verbose=False):
        """Runs a mysqlctld command.

    Args:
      cmd: the command to run.
      extra_my_cnf: list of extra mycnf files to use
      verbose: passed to mysqlctld.
    Returns:
      the result of run_bg.
    """
        extra_env = {}
        all_extra_my_cnf = get_all_extra_my_cnf(extra_my_cnf)
        if all_extra_my_cnf:
            extra_env["EXTRA_MY_CNF"] = ":".join(all_extra_my_cnf)
        args = environment.binary_args("mysqlctld") + [
            "-log_dir",
            environment.vtlogroot,
            "-tablet_uid",
            str(self.tablet_uid),
            "-mysql_port",
            str(self.mysql_port),
            "-socket_file",
            os.path.join(self.tablet_dir, "mysqlctl.sock"),
        ]
        self._add_dbconfigs(args)
        if verbose:
            args.append("-alsologtostderr")
        args.extend(cmd)
        return utils.run_bg(args, extra_env=extra_env)

  def test_sigterm(self):
    utils.run_vtctl('CreateKeyspace test_keyspace')

    # create the database so vttablets start, as it is serving
    tablet_62344.create_db('vt_test_keyspace')

    tablet_62344.init_tablet('master', 'test_keyspace', '0', start=True)

    # start a 'vtctl Sleep' command in the background
    args = [environment.binary_path('vtctl'),
            '-log_dir', environment.vtlogroot,
            '--alsologtostderr']
    args.extend(environment.topo_server_flags())
    args.extend(environment.tablet_manager_protocol_flags())
    args.extend(['Sleep', tablet_62344.tablet_alias, '60s'])
    sp = utils.run_bg(args, stdout=PIPE, stderr=PIPE)

    # wait for it to start, and let's kill it
    time.sleep(4.0)
    utils.run(['pkill', 'vtaction'])
    out, err = sp.communicate()

    # check the vtctl command got the right remote error back
    if "vtaction interrupted by signal" not in err:
      self.fail("cannot find expected output in error: " + err)
    logging.debug("vtaction was interrupted correctly:\n" + err)

    tablet_62344.kill_vttablet()

  def mysqlctld(self, cmd, extra_my_cnf=None, verbose=False, extra_args=None):
    """Runs a mysqlctld command.

    Args:
      cmd: the command to run.
      extra_my_cnf: list of extra mycnf files to use
      verbose: passed to mysqlctld.
      extra_args: passed to mysqlctld.
    Returns:
      the result of run_bg.
    """
    extra_env = {}
    all_extra_my_cnf = get_all_extra_my_cnf(extra_my_cnf)
    if all_extra_my_cnf:
      extra_env['EXTRA_MY_CNF'] = ':'.join(all_extra_my_cnf)
    args = environment.binary_args('mysqlctld') + [
        '-log_dir', environment.vtlogroot,
        '-tablet_uid', str(self.tablet_uid),
        '-mysql_port', str(self.mysql_port),
        '-socket_file', os.path.join(self.tablet_dir, 'mysqlctl.sock')]
    self._add_dbconfigs(self.default_db_dba_config, args)
    if verbose:
      args.append('-alsologtostderr')
    if extra_args:
      args.extend(extra_args)
    args.extend(cmd)
    return utils.run_bg(args, extra_env=extra_env)

  def test_vtgate_qps(self):
    # create the topology
    utils.run_vtctl('CreateKeyspace test_keyspace')
    t = tablet.Tablet(tablet_uid=1, cell="nj")
    t.init_tablet("master", "test_keyspace", "0")
    t.update_addrs()
    utils.run_vtctl('RebuildKeyspaceGraph test_keyspace', auto_log=True)

    # start vtgate and the qps-er
    vtgate_proc, vtgate_port = utils.vtgate_start(
        extra_args=['-cpu_profile', os.path.join(environment.tmproot,
                                                 'vtgate.pprof')])
    qpser = utils.run_bg(environment.binary_args('zkclient2') + [
        '-server', 'localhost:%u' % vtgate_port,
        '-mode', 'qps',
        '-zkclient_cpu_profile', os.path.join(environment.tmproot, 'zkclient2.pprof'),
        'test_nj', 'test_keyspace'])
    qpser.wait()

    # get the vtgate vars, make sure we have what we need
    v = utils.get_vars(vtgate_port)

    # some checks on performance / stats
    rpcCalls = v['TopoReaderRpcQueryCount']['test_nj']
    if rpcCalls < MIN_QPS * 10:
      self.fail('QPS is too low: %u < %u' % (rpcCalls / 10, MIN_QPS))
    else:
      logging.debug("Recorded qps: %u", rpcCalls / 10)
    utils.vtgate_kill(vtgate_proc)

  def __init__(self, name):
    import environment  # pylint: disable=g-import-not-at-top
    import utils  # pylint: disable=g-import-not-at-top

    self.port_base = environment.reserve_ports(2)

    self.name = name
    self.hostname = 'localhost'
    self.client_port = self.port_base
    self.peer_port = self.port_base + 1
    self.client_addr = 'http://%s:%d' % (self.hostname, self.client_port)
    self.peer_addr = 'http://%s:%d' % (self.hostname, self.peer_port)
    self.api_url = self.client_addr + '/v2'

    dirname = 'etcd_' + self.name
    self.data_dir = os.path.join(environment.vtdataroot, dirname)

    self.proc = utils.run_bg([
        'etcd', '-name', self.name,
        '-advertise-client-urls', self.client_addr,
        '-initial-advertise-peer-urls', self.peer_addr,
        '-listen-client-urls', self.client_addr,
        '-listen-peer-urls', self.peer_addr,
        '-initial-cluster', '%s=%s' % (self.name, self.peer_addr),
        '-data-dir', self.data_dir],
                             stdout=open(os.path.join(
                                 environment.vtlogroot,
                                 dirname + '.stdout'),
                                         'w'),
                             stderr=open(os.path.join(
                                 environment.vtlogroot,
                                 dirname + '.stderr'),
                                         'w'),)

  def test_primecache(self):
    utils.run_vtctl(['CreateKeyspace', 'test_keyspace'])

    master.init_tablet( 'master',  'test_keyspace', '0')
    replica.init_tablet('idle')

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

    master.create_db('vt_test_keyspace')
    master.start_vttablet(wait_for_state=None)
    replica.start_vttablet(wait_for_state=None)

    master.wait_for_vttablet_state('SERVING')
    replica.wait_for_vttablet_state('NOT_SERVING') # DB doesn't exist

    self._create_data()

    # we use clone to not prime the mysql cache on the slave db
    utils.run_vtctl(['Clone', '-force', '-server-mode',
                     master.tablet_alias, replica.tablet_alias],
                    auto_log=True)

    # sync the buffer cache, and clear it. This will prompt for user's password
    utils.run(['sync'])
    utils.run(['sudo', 'bash', '-c', 'echo 1 > /proc/sys/vm/drop_caches'])

    # we can now change data on the master for 30s, while slave is stopped.
    # master's binlog will be in OS buffer cache now.
    replica.mquery('', 'slave stop')
    self._change_random_data()

    use_primecache = True # easy to test without
    if use_primecache:
      # starting vtprimecache, sleeping for a couple seconds
      args = environment.binary_args('vtprimecache') + [
              '-db-config-dba-uname', 'vt_dba',
              '-db-config-dba-charset', 'utf8',
              '-db-config-dba-dbname', 'vt_test_keyspace',
              '-db-config-app-uname', 'vt_app',
              '-db-config-app-charset', 'utf8',
              '-db-config-app-dbname', 'vt_test_keyspace',
              '-relay_logs_path', replica.tablet_dir+'/relay-logs',
              '-mysql_socket_file', replica.tablet_dir+'/mysql.sock',
              '-log_dir', environment.vtlogroot,
              '-worker_count', '4',
              '-alsologtostderr',
             ]
      vtprimecache = utils.run_bg(args)
      time.sleep(2)

    # start slave, see how longs it takes to catch up on replication
    replica.mquery('', 'slave start')
    self.catch_up()

    if use_primecache:
      # TODO(alainjobart): read and check stats
      utils.kill_sub_process(vtprimecache)

    tablet.kill_tablets([master, replica])

def _check_db_addr(db_addr, expected_addr):
    # Run in the background to capture output.
    proc = utils.run_bg(
        utils.vtroot + "/bin/vtctl --alsologtostderr -zk.local-cell=test_nj Resolve " + db_addr, stdout=PIPE
    )
    stdout = proc.communicate()[0].strip()
    if stdout != expected_addr:
        raise utils.TestError("wrong zk address", db_addr, stdout, expected_addr)

    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()